【Agent专题】Agent翻译官：一文说透AG-UI ，先收藏了！

最近我看到了一个新鲜有趣的东西：AG-UI。看起来太有意思了，我忍不住想跟大家分享一下。

简单来说，它就像是让那些酷炫的 AI 应用变得自然顺滑的“神秘调味料”。

你不是技术宅？别担心！我会用生活化的语言解释清楚。

咱们先从头开始，什么是 AI Agent？大部分人可能已经听说过它，但为了方便理解，我们再重新捋一遍。

想象你在一个聚会上，有一个人什么都知道。

想听歌？他立刻给你推荐播放列表。
想找附近咖啡店？他秒给你导航。

而且这个人不仅能回答你的问题，还能直接帮你做事。这就是“AI Agent”——一个超聪明、由人工智能驱动的软件程序。

理解了吧？Agent 不只是能对话，它还能操作，比如上网搜索、控制设备等。不过，这些 Agent 往往藏在应用的“后端”，我们平时是看不到的。

那我们怎么跟它们互动呢？这就得说到“前端”了。前端，简单说，就是你在一个应用或网站上看得见、点得到、输入得了的部分：

Instagram 里的彩色图片流
Google 地图上的搜索栏和地图
客服网站上的聊天窗口

前端是应用的“脸面”，而 AI Agent 是背后的“大脑”。没有前端，你就没法与 AI Agent 对话。

就好比你去餐厅点菜，但却没有服务员来接单。

所以，我们需要一种机制，把“大脑”与“脸面”连接起来。这时候，AGUI 就登场了，它是一位“神助攻”。

AGUI 到底是干嘛的？（就像是 AI 世界的万能翻译官）。你有没有试过和一个不会讲你语言的人交流？费劲又尴尬。

这正是开发者在没有统一标准时，试图让 AI Agent 与前端沟通时的真实写照：混乱、费时、令人抓狂。

AGUI 就像一个万能翻译官，甚至更像一个训练有素的餐厅服务员。它是一套“协议”（别被这个词吓到，就是一套大家约定好的规则）。

有了 AGUI，AI Agent 和前端就可以无障碍沟通。无论它们是用什么语言开发的，只要都遵循 AGUI 的规则，就能轻松配合。

就像在餐厅点菜，你对服务员说：“我想吃汉堡和薯条”，服务员把你的需求准确传达给后厨，然后把做好了的菜端给你。AGUI 就是那套确保点菜无误、上菜无误的“流程系统”。

AGUI 是如何工作的？（举个例子你就懂）假设你正在使用一个旅行 app 计划行程。你输入：“帮我查下下周五飞纽约的便宜机票。”AGUI 在背后是这样运作的：

你发送请求：你在 app 上输入需求（前端）。app 通过 AGUI 把请求传给 AI Agent。
Agent 开始处理：Agent 开始工作，并发送事件给前端，比如“我开始处理请求了”。
实时反馈：AI Agent 持续发送事件，比如“正在查航班信息…”，你就像看到打字提示一样，感觉有人在实时回复你。
调用工具：AI 需要查航班，就发送“正在查找航班”这样的事件，前端显示进度条或加载动画。
完成任务：Agent 最后发出：“机票已预订成功！以下是确认信息。”前端展示结果，你就能开始收拾行李了。

AGUI 就像交通规则一样，确保所有环节顺利配合。

它使用一套标准事件，例如：

run_started message_started tool_call run_finished

前端只需监听这些事件，便能让整个体验像自然对话一样流畅。

那么，AGUI 为啥这么重要？你可能会想，如果我不写代码，AGUI 跟我有啥关系？其实它的意义远不止开发者的便利。在 AGUI 出现之前，开发者每接一个项目，就得重新造“连接 AI 和前端”的桥，费时费力。

现在有了 AGUI，就像有了现成的高速公路，谁都能跑。这代表什么？未来更多 AI 应用能快速落地，提升你生活的各个方面：

旅行安排：你用 app 一问，AI 就能帮你订机票、酒店，甚至推荐附近的咖啡厅。
智能家居：你说“把空调调到 24 度”，AI 不仅听懂，还能实时反馈：“空调已开启。”
客服体验：不只是答疑，还能真正操作，比如退款、预约。

甚至小企业也能用 AGUI 做智能服务！

想象你开了一家蛋糕店，用 AGUI 就可以做一个智能订蛋糕小助手：

客户对话输入：“我要一个 20 人份的巧克力蛋糕”
AI 检查库存、给出设计建议
实时反馈：“正在确认是否有足够的巧克力酱”
订单完成：“蛋糕已完成，请来取货\~”

整个过程用户不需要懂 AI，也不需要下载什么复杂应用，对话就能搞定一切！AGUI 的核心功能亮点。

实时响应流：AI 的回答像打字一样逐字呈现，用户体验更自然。
工具编排：AI 使用工具时，前端能实时显示进度和状态。
标准事件系统：AI 与前端使用统一的“事件语言”，避免理解偏差。
框架自由度高：无论前端还是后端，开发者可以用任何语言或框架，只要遵循 AGUI 协议即可。

而且，AGUI 是 开源的！这意味着任何人都可以免费使用、修改，社区可以共同改进，就像一本共享食谱，谁都能往里加点“好味道”。

如果你是开发者，还可以使用如 Windsurf 或 Cursor 这样的 AI 编程助手，帮助你更快基于 AGUI 构建产品。

开发者看这里：前后端该怎么用？

✅ 前端开发者：

AG-UI 支持 事件驱动、流式输出、状态控制，你只需要监听标准事件，比如：

RunStarted
TextMessageStart
ToolCallStart
RunFinished

来看一个 Node.js + TypeScript 的 CLI 示例：

import * as readline from "readline" import { agent } from "./agent" import { randomUUID } from "node:crypto" const rl = readline.createInterface({ input: process.stdin, output: process.stdout, }) async function chatLoop() { console.log("🤖 AG-UI Assistant started!") console.log("Type your messages and press Enter. Press Ctrl+D to quit.\n") return new Promise<void>((resolve) => { const promptUser = () => { rl.question("> ", async (input) => { if (input.trim() === "") { promptUser() return } console.log("") rl.pause() agent.messages.push({ id: randomUUID(), role: "user", content: input.trim(), }) try { await agent.runAgent( {}, { onTextMessageStartEvent() { process.stdout.write("🤖 Assistant: ") }, onTextMessageContentEvent({ event }) { process.stdout.write(event.delta) }, onTextMessageEndEvent() { console.log("\n") }, } ) } catch (error) { console.error("❌ Error:", error) } rl.resume() promptUser() }) } rl.on("close", () => { console.log("\n👋 Thanks for using AG-UI Assistant!") resolve() }) promptUser() }) } async function main() { await chatLoop() } main().catch(console.error)

是不是清爽又实用？

✅ 后端开发者：

你的 API 服务可以封装成 AG-UI 协议支持，比如用 FastAPI + OpenAI：

import os import uuid import uvicorn from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse from ag_ui.core import ( RunAgentInput, EventType, RunStartedEvent, RunFinishedEvent, RunErrorEvent, ) from ag_ui.encoder import EventEncoder from openai import OpenAI app = FastAPI(title="AG-UI OpenAI Server") client = OpenAI() @app.post("/") async def agentic_chat_endpoint(input_data: RunAgentInput, request: Request): accept_header = request.headers.get("accept") encoder = EventEncoder(accept=accept_header) async def event_generator(): try: yield encoder.encode( RunStartedEvent( type=EventType.RUN_STARTED, thread_id=input_data.thread_id, run_id=input_data.run_id ) ) stream = client.chat.completions.create( model="gpt-4o", stream=True, tools=[ { "type": "function", "function": { "name": tool.name, "description": tool.description, "parameters": tool.parameters, } } for tool in input_data.tools ] if input_data.tools else None, messages=[ { "role": message.role, "content": message.content or "", **({"tool_calls": message.tool_calls} if message.role == "assistant" and hasattr(message, 'tool_calls') and message.tool_calls else {}), **({"tool_call_id": message.tool_call_id} if message.role == "tool" and hasattr(message, 'tool_call_id') else {}), } for message in input_data.messages ], ) message_id = str(uuid.uuid4()) for chunk in stream: if chunk.choices[0].delta.content: yield encoder.encode({ "type": EventType.TEXT_MESSAGE_CHUNK, "message_id": message_id, "delta": chunk.choices[0].delta.content, }) elif chunk.choices[0].delta.tool_calls: tool_call = chunk.choices[0].delta.tool_calls[0] yield encoder.encode({ "type": EventType.TOOL_CALL_CHUNK, "tool_call_id": tool_call.id, "tool_call_name": tool_call.function.name if tool_call.function else None, "parent_message_id": message_id, "delta": tool_call.function.arguments if tool_call.function else None, }) yield encoder.encode( RunFinishedEvent( type=EventType.RUN_FINISHED, thread_id=input_data.thread_id, run_id=input_data.run_id ) ) except Exception as error: yield encoder.encode( RunErrorEvent( type=EventType.RUN_ERROR, message=str(error) ) ) return StreamingResponse( event_generator(), media_type=encoder.get_content_type() ) def main(): port = int(os.getenv("PORT", "8000")) uvicorn.run("example_server:app", host="0.0.0.0", port=port, reload=True) if __name__ == "__main__": main()