作者：PySuper | 来源：zhengxingtao.com
日期：2026-09-15
目录
1. 为什么需要 AG-UI？
2. AG-UI 是什么
3. 协议架构：事件流模型
4. 协议三角：MCP + A2A + AG-UI
5. 事件类型详解
6. 实战1：用 CopilotKit + AG-UI 构建 Agent 驱动的 Web 应用
7. 实战2：将 LangGraph Agent 通过 AG-UI 接入前端
8. 适用场景 vs 不适用场景
9. 生态支持现状
10. 踩坑记录
11. 总结与展望
1. 为什么需要 AG-UI？
2024-2025 年，AI Agent 生态经历了爆发式增长——LangGraph、CrewAI、Mastra、AG2、AutoGen 等框架百花齐放。但一个关键问题始终悬而未决：
Agent 和前端 UI 之间缺乏标准通信方式。
具体来说，开发者面临这些困境：
plaintext
┌──────────────────────────────────────────────────────────────┐
│ Agent 前端接入的 5 大痛点 │
├──────────────────────────────────────────────────────────────┤
│ │
│ 1. 格式碎片化 │
│ ├─ LangGraph 用自己的 stream_events │
│ ├─ CrewAI 用自定义 JSON WebSocket │
│ ├─ Mastra 用 REST + SSE │
│ └─ 每个框架一个方言，前端需要逐个适配 │
│ │
│ 2. 实时流式困难 │
│ ├─ LLM 逐 token 输出，UI 需要 token-by-token 渲染 │
│ ├─ WebSocket 双工但缺乏标准化事件格式 │
│ └─ 纯轮询延迟太高，用户体验差 │
│ │
│ 3. 工具调用不可见 │
│ ├─ Agent 调用了什么工具？前端看不到 │
│ ├─ 需要用户审批的工具怎么暂停/恢复？ │
│ └─ 工具执行进度怎么实时展示？ │
│ │
│ 4. 状态同步混乱 │
│ ├─ Agent 内部状态怎么同步到前端？ │
│ ├─ 全量快照太重，增量更新没规范 │
│ └─ 前端修改状态怎么回传 Agent？ │
│ │
│ 5. 并发与中断 │
│ ├─ 多个 Agent 同时运行怎么管理？ │
│ ├─ 用户想中断某个 Agent 怎么发信号？ │
│ └─ 中断后恢复上下文如何保持？ │
│ │
└──────────────────────────────────────────────────────────────┘
结果就是：每个团队都在重复造轮子——写 WebSocket 适配器、解析自定义 JSON 格式、处理 SSE 重连逻辑。CopilotKit 团队自己在构建产品时也踩了这些坑，所以他们决定抽出一个通用协议。
据《Introducing AG-UI: The Protocol Where Agents Meet Users》（https://www.copilotkit.ai/blog/introducing-ag-ui-the-protocol-where-agents-meet-users），AG-UI 于 2025 年 5 月正式发布，CopilotKit 团队在获得 $27M A 轮融资后，将其开源并推动成为行业标准。
2. AG-UI 是什么
AG-UI（Agent-User Interaction Protocol）是一个开源、轻量的协议，定义了 AI Agent 与前端用户界面之间的标准通信方式。
核心设计原则：
plaintext
┌─────────────────────────────────────────────────────────────────┐
│ AG-UI 核心设计原则 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 1. 单请求 → 事件流 │
│ 客户端发一个 POST，后端回一个 SSE 事件流 │
│ 不需要 WebSocket，不需要自定义序列化 │
│ │
│ 2. 统一事件格式 │
│ 所有交互都编码为 JSON 事件 │
│ 每个 event 有 type 字段 + 最小化 payload │
│ 前端按 type 分发处理即可 │
│ │
│ 3. 基于 HTTP │
│ 走标准 HTTP/SSE，天然兼容： │
│ ├─ CORS、认证 Token、审计日志 │
│ ├─ 负载均衡、CDN、API Gateway │
│ └─ 可选二进制序列化用于性能敏感场景 │
│ │
│ 4. 框架无关 │
│ 后端可以是 LangGraph / CrewAI / Mastra / 自定义 │
│ 前端可以是 React / Vue / 原生 │
│ 只要遵循协议，组件可互换 │
│ │
└─────────────────────────────────────────────────────────────────┘
一句话总结：AG-UI 是 Agent 世界的"HTTP"——一个把 Agent 和 UI 解耦的通信协议。
3. 协议架构：事件流模型
3.1 请求-响应模型
plaintext
┌───────────┐ ┌───────────┐
│ │ POST /awp │ │
│ Frontend │ RunAgentInput (JSON) │ Agent │
│ (React) │ ──────────────────────────>│ Backend │
│ │ │ (Python) │
│ │ SSE Stream │ │
│ │ <──────────────────────────│ │
│ │ Event 1: RUN_STARTED │ │
│ │ Event 2: TEXT_MESSAGE_ │ │
│ │ START │ │
│ │ Event 3: TEXT_MESSAGE_ │ │
│ │ CONTENT │ │
│ │ Event 4: TOOL_CALL_START │ │
│ │ Event 5: TOOL_CALL_ARGS │ │
│ │ Event 6: TOOL_CALL_END │ │
│ │ Event 7: STATE_DELTA │ │
│ │ Event 8: RUN_FINISHED │ │
└───────────┘ └───────────┘
3.2 RunAgentInput 结构
前端发往 Agent 的请求体：
python
# RunAgentInput - 前端发给 Agent 的标准输入
{
"thread_id": "thread_abc123", # 会话线程 ID
"run_id": "run_xyz789", # 本次运行 ID
"messages": [ # 对话历史
{
"role": "user",
"content": "帮我分析这份销售报告"
},
{
"role": "assistant",
"content": "好的，让我来分析..."
}
],
"state": { # 共享状态
"current_document": "report.xlsx",
"analysis_mode": "detailed"
},
"tools": [ # 前端可用工具（可选）
{
"name": "show_chart",
"description": "在 UI 上展示图表",
"parameters": {
"type": "object",
"properties": {
"chart_type": {"type": "string"},
"data": {"type": "object"}
}
}
}
],
"context": [], # 上下文信息（可选）
"forwarded_props": {} # 转发属性（可选）
}
3.3 事件基础结构
所有 AG-UI 事件共享一个 BaseEvent 结构：
python
# BaseEvent - 所有事件的基类
{
"type": "TextMessageContent", # 必填，事件类型标识
"timestamp": 1705324800000, # 可选，Unix 毫秒时间戳
"rawEvent": null # 可选，原始事件载荷
}
3.4 CopilotKit 中的 Proxy 模式
CopilotKit 使用代理模式，前端不直连 Agent：
plaintext
┌──────────┐ useAgent() ┌──────────────────────┐
│ │ ────────────────> │ ProxiedAgent │
│ React │ 返回 │ (前端代理) │
│ 组件 │ <──────────────── │ ├─ 同一 AbstractAgent │
│ │ AbstractAgent │ ├─ 同一 subscribe API │
└──────────┘ │ └─ 同一属性接口 │
└──────────┬───────────┘
│
HTTP POST + SSE
│
┌──────────▼───────────┐
│ CopilotKit Runtime │
│ (服务端) │
└──────────┬───────────┘
│
AG-UI 事件流
│
┌──────────▼───────────┐
│ 真实 Agent Backend │
│ (LangGraph/CrewAI) │
└──────────────────────┘
这样做的好处是前端代码完全不关心 Agent 跑在哪里——本地、远程、云端都一样。
4. 协议三角：MCP + A2A + AG-UI
这三个协议分别解决了 Agent 生态中不同层级的通信问题：
plaintext
┌──────────────────────────────────────────────────────┐
│ │
│ ┌─────────────┐ │
│ │ AG-UI │ Agent ↔ User │
│ │ (交互层) │ "用户怎么和Agent交互" │
│ └──────┬──────┘ │
│ │ │
│ ┌──────┴──────┐ │
│ │ A2A │ Agent ↔ Agent │
│ │ (协作层) │ "Agent之间怎么协作" │
│ └──────┬──────┘ │
│ │ │
│ ┌──────┴──────┐ │
│ │ MCP │ Agent ↔ Tools │
│ │ (工具层) │ "Agent怎么调用工具" │
│ └─────────────┘ │
│ │
└──────────────────────────────────────────────────────┘
自底向上：
MCP → 定义 Agent 与工具/API 的通信
A2A → 定义 Agent 与 Agent 之间的通信
AG-UI → 定义 Agent 与用户/前端的通信
据《AG-UI Is Redefining the Agent–User Interaction Layer》（https://webflow.copilotkit.ai/blog/ag-ui-is-redefining-the-agent-user-interaction-layer），A2A 和 AG-UI 已完成官方握手（handshake），首次将 Agent 间通信和 Agent-用户通信层连接起来。
它们不是竞争关系，而是互补关系。 一个完整的 Agent 系统通常三层都需要：
plaintext
场景示例：用户让 AI 团队协作完成市场调研
1. 用户输入任务 (AG-UI 层)
├─ 前端通过 AG-UI 发送 RUN_STARTED 事件
└─ Agent 接收并开始执行
2. 多 Agent 协作 (A2A 层)
├─ 主 Agent 拆分任务给 子Agent
├─ 子Agent A: 数据收集
├─ 子Agent B: 竞品分析
└─ 子Agent C: 报告生成
3. Agent 调用外部工具 (MCP 层)
├─ 通过 MCP 调用搜索引擎 API
├─ 通过 MCP 读取数据库
└─ 通过 MCP 调用图表工具
4. 结果回传用户 (AG-UI 层)
├─ TEXT_MESSAGE_CONTENT: "调研完成，这是报告..."
├─ TOOL_CALL_END: 渲染图表
└─ RUN_FINISHED: 运行完成
5. 事件类型详解
AG-UI 定义了以下标准事件类别：
5.1 生命周期事件
plaintext
┌──────────────────────────────────────────────────────┐
│ Agent 运行生命周期 │
├──────────────────────────────────────────────────────┤
│ │
│ RUN_STARTED ──> STEP_STARTED ──> STEP_FINISHED │
│ │ │ │
│ │ ├─ (多个 Step 可嵌套) │
│ │ │ │
│ ├──────────────┴──> RUN_FINISHED │
│ │ │
│ └──(异常)──> RUN_ERROR │
│ │
└──────────────────────────────────────────────────────┘
python
# RUN_STARTED 事件
{
"type": "RunStarted",
"thread_id": "thread_abc123",
"run_id": "run_xyz789",
"parent_run_id": null, # 父运行 ID（子 Agent 时有值）
"input": {"query": "..."} # 可选的输入信息
}
# STEP_STARTED 事件 - 标记一个处理步骤开始
{
"type": "StepStarted",
"step_name": "analyzing_data" # 步骤名称
}
# STEP_FINISHED 事件
{
"type": "StepFinished",
"step_name": "analyzing_data"
}
# RUN_FINISHED 事件
{
"type": "RunFinished",
"thread_id": "thread_abc123",
"run_id": "run_xyz789",
"result": {"summary": "..."} # 可选的运行结果
}
# RUN_ERROR 事件
{
"type": "RunError",
"message": "API rate limit exceeded",
"code": "RATE_LIMIT" # 可选错误码
}
5.2 文本消息事件
支持 token-by-token 流式输出：
python
# TEXT_MESSAGE_START - 开始一条新消息
{
"type": "TextMessageStart",
"message_id": "msg_001",
"role": "assistant"
}
# TEXT_MESSAGE_CONTENT - 逐块推送文本
{
"type": "TextMessageContent",
"message_id": "msg_001",
"delta": "你好" # 增量文本
}
# TEXT_MESSAGE_CONTENT - 下一块
{
"type": "TextMessageContent",
"message_id": "msg_001",
"delta": "，我正在分析"
}
# TEXT_MESSAGE_END - 消息结束
{
"type": "TextMessageEnd",
"message_id": "msg_001"
}
5.3 工具调用事件
python
# TOOL_CALL_START - Agent 开始调用工具
{
"type": "ToolCallStart",
"tool_call_id": "tc_001",
"tool_call_name": "search_database",
"parent_message_id": "msg_001" # 可选，关联的消息
}
# TOOL_CALL_ARGS - 逐块推送工具参数
{
"type": "ToolCallArgs",
"tool_call_id": "tc_001",
"delta": "{\"query\": \"sales data Q3\"}" # 增量 JSON
}
# TOOL_CALL_END - 工具调用阶段完成
{
"type": "ToolCallEnd",
"tool_call_id": "tc_001"
}
# TOOL_CALL_RESULT - 工具返回结果
{
"type": "ToolCallResult",
"tool_call_id": "tc_001",
"content": "{\"results\": [...]}",
"message_id": "msg_002",
"role": "tool"
}
5.4 状态管理事件
python
# STATE_SNAPSHOT - 完整状态快照（首次同步或全量刷新）
{
"type": "StateSnapshot",
"snapshot": {
"current_step": 3,
"total_steps": 5,
"files_processed": ["report1.csv", "report2.csv"],
"progress": 0.6
}
}
# STATE_DELTA - 增量状态更新（JSON Patch 格式）
{
"type": "StateDelta",
"delta": [
{"op": "replace", "path": "/progress", "value": 0.8},
{"op": "add", "path": "/files_processed/-", "value": "report3.csv"}
]
}
5.5 Activity 事件
用于展示 Agent 的中间活动/进度指示：
python
# ACTIVITY_SNAPSHOT - 活动快照
{
"type": "ActivitySnapshot",
"message_id": "activity_001",
"activity_type": "thinking",
"content": {
"step": "Searching database...",
"progress": 45
},
"replace": false # 是否替换之前的 activity
}
# ACTIVITY_DELTA - 活动增量更新
{
"type": "ActivityDelta",
"message_id": "activity_001",
"activity_type": "thinking",
"patch": [
{"op": "replace", "path": "/progress", "value": 78}
]
}
5.6 自定义事件
python
# CUSTOM 事件 - 扩展用
{
"type": "Custom",
"name": "chart_update",
"value": {
"chart_type": "bar",
"data_points": [10, 20, 30]
}
}
5.7 事件类型汇总表
表格
6. 实战1：用 CopilotKit + AG-UI 构建 Agent 驱动的 Web 应用
6.1 项目结构
plaintext
ag-ui-demo/
├── frontend/ # React 前端
│ ├── package.json
│ ├── src/
│ │ ├── App.tsx
│ │ ├── main.tsx
│ │ └── components/
│ │ ├── AgentChat.tsx
│ │ └── ProgressPanel.tsx
│ └── vite.config.ts
├── backend/ # Python 后端
│ ├── requirements.txt
│ ├── main.py
│ └── agent.py
└── docker-compose.yml
6.2 后端：创建 AG-UI Agent 服务
python
# backend/requirements.txt
fastapi==0.115.0
uvicorn==0.32.0
copilotkit==0.1.40
langchain-openai==0.3.0
langchain-core==0.3.0
python-dotenv==1.0.1
python
# backend/agent.py
"""
AG-UI Agent 实现
定义 Agent 的工具和行为，通过 AG-UI 协议与前端通信
"""
import os
import json
from datetime import datetime
from typing import Any
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import SystemMessage
from copilotkit.integrations.fastapi import add_fastapi_endpoint
from copilotkit import CopilotKitSDK, LangGraphAgent
from copilotkit import Action
# 定义 Agent 可用的工具
@tool
def search_knowledge_base(query: str) -> str:
"""搜索知识库获取相关信息"""
# 模拟知识库搜索
knowledge = {
"定价策略": "我们提供三种定价方案：基础版¥99/月，专业版¥299/月，企业版¥999/月",
"退款政策": "购买后30天内可申请全额退款，需提供购买凭证",
"技术支持": "技术支持热线：400-xxx-xxxx，工作时间 9:00-18:00",
"产品功能": "核心功能包括：数据分析、报表生成、API集成、团队协作",
}
results = []
for key, value in knowledge.items():
if query in key or key in query:
results.append(f"【{key}】{value}")
if not results:
return f"未找到与 '{query}' 相关的信息，建议联系人工客服"
return "\n".join(results)
@tool
def create_support_ticket(
customer_name: str,
issue_type: str,
description: str,
priority: str = "normal"
) -> str:
"""创建客服工单"""
ticket_id = f"TK-{datetime.now().strftime('%Y%m%d%H%M%S')}"
ticket = {
"ticket_id": ticket_id,
"customer_name": customer_name,
"issue_type": issue_type,
"description": description,
"priority": priority,
"status": "open",
"created_at": datetime.now().isoformat()
}
# 实际场景中存入数据库
print(f"[Ticket Created] {json.dumps(ticket, ensure_ascii=False)}")
return f"工单创建成功！工单号: {ticket_id}，我们将尽快处理。"
@tool
def check_order_status(order_id: str) -> str:
"""查询订单状态"""
# 模拟订单查询
orders = {
"ORD-20260901": {"status": "已发货", "tracking": "SF1234567890", "eta": "2026-09-17"},
"ORD-20260905": {"status": "处理中", "tracking": None, "eta": "2026-09-20"},
"ORD-20260910": {"status": "已签收", "tracking": "SF9876543210", "eta": None},
}
if order_id in orders:
info = orders[order_id]
result = f"订单号: {order_id}\n状态: {info['status']}"
if info["tracking"]:
result += f"\n物流单号: {info['tracking']}"
if info["eta"]:
result += f"\n预计到达: {info['eta']}"
return result
return f"未找到订单号 {order_id}，请确认订单号是否正确"
# 创建 LLM 实例
llm = ChatOpenAI(
model="gpt-4o",
temperature=0.3,
api_key=os.environ.get("OPENAI_API_KEY")
)
# 绑定工具
llm_with_tools = llm.bind_tools(
[search_knowledge_base, create_support_ticket, check_order_status]
)
# 系统提示词
SYSTEM_PROMPT = """你是一个专业的客服助手 Agent。你可以：
1. 搜索知识库回答常见问题（使用 search_knowledge_base 工具）
2. 查询订单状态（使用 check_order_status 工具）
3. 创建客服工单（使用 create_support_ticket 工具）
工作流程：
- 先判断用户意图
- 如果是常见问题，搜索知识库
- 如果是订单问题，查询订单状态
- 如果需要人工介入，创建工单
- 始终用友好、专业的语气回复
注意：创建工单前必须确认用户信息，不要随意填写。
"""
# 创建 CopilotKit Agent
def create_agent():
"""创建 AG-UI 兼容的 Agent"""
from langgraph.prebuilt import create_react_agent
agent = create_react_agent(
llm,
tools=[search_knowledge_base, create_support_ticket, check_order_status],
prompt=SYSTEM_PROMPT,
)
return agent
python
# backend/main.py
"""
AG-UI Agent FastAPI 服务端
通过 CopilotKit SDK 将 Agent 暴露为 AG-UI 端点
"""
import os
from dotenv import load_dotenv
from fastapi import FastAPI
from copilotkit.integrations.fastapi import add_fastapi_endpoint
from copilotkit import CopilotKitSDK, LangGraphAgent
from agent import create_agent
load_dotenv()
app = FastAPI(title="AG-UI Customer Service Agent")
# 创建 Agent 实例
agent = create_agent()
# 初始化 CopilotKit SDK
sdk = CopilotKitSDK(
agents=[
LangGraphAgent(
name="customer_service_agent",
agent=agent,
description="客服助手 Agent，支持知识库查询、订单查询、工单创建",
)
]
)
# 添加 AG-UI 端点
add_fastapi_endpoint(app, sdk, "/api/copilotkit")
# 健康检查
@app.get("/health")
async def health():
return {"status": "ok", "protocol": "AG-UI"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"main:app",
host="0.0.0.0",
port=8000,
reload=True
)
6.3 前端：React + CopilotKit
json
// frontend/package.json
{
"name": "ag-ui-frontend",
"private": true,
"version": "0.0.1",
"type": "module",
"scripts": {
"dev": "vite",
"build": "tsc && vite build"
},
"dependencies": {
"@copilotkit/react-core": "^1.3.0",
"@copilotkit/react-ui": "^1.3.0",
"@copilotkit/runtime-client-http": "^1.3.0",
"react": "^18.3.0",
"react-dom": "^18.3.0"
},
"devDependencies": {
"@types/react": "^18.3.0",
"@types/react-dom": "^18.3.0",
"@vitejs/plugin-react": "^4.3.0",
"typescript": "^5.5.0",
"vite": "^5.4.0"
}
}
tsx
// frontend/src/App.tsx
import { CopilotKit } from "@copilotkit/react-core";
import { CopilotPopup } from "@copilotkit/react-ui";
import "@copilotkit/react-ui/styles.css";
import { ProgressPanel } from "./components/ProgressPanel";
import { AgentChat } from "./components/AgentChat";
function App() {
return (
<CopilotKit runtimeUrl="http://localhost:8000/api/copilotkit">
<div className="app-container">
<header className="app-header">
<h1>AG-UI 智能客服</h1>
<span className="protocol-badge">AG-UI Protocol</span>
</header>
<main className="app-main">
<div className="chat-area">
<AgentChat />
</div>
<aside className="side-panel">
<ProgressPanel />
</aside>
</main>
<CopilotPopup
instructions="你是一个专业的客服助手，帮助用户解决问题。"
labels={{
title: "客服助手",
initial: "你好！有什么可以帮你的？"
}}
/>
</div>
</CopilotKit>
);
}
export default App;
tsx
// frontend/src/components/AgentChat.tsx
import { useAgent } from "@copilotkit/react-core";
import { useEffect, useState } from "react";
interface StepInfo {
name: string;
status: "running" | "finished" | "error";
}
interface ToolCallInfo {
name: string;
args: string;
status: "started" | "finished";
}
export function AgentChat() {
const { agent } = useAgent({ agentId: "customer_service_agent" });
const [streamingText, setStreamingText] = useState("");
const [steps, setSteps] = useState<StepInfo[]>([]);
const [toolCalls, setToolCalls] = useState<ToolCallInfo[]>([]);
const [agentState, setAgentState] = useState<Record<string, any>>({});
useEffect(() => {
if (!agent) return;
const subscription = agent.subscribe({
// 监听所有事件
onEvent({ event }) {
console.log("[AG-UI Event]", event.type, event);
},
// 流式文本
onTextMessageContentEvent({ event, textMessageBuffer }) {
setStreamingText(textMessageBuffer);
},
// 步骤追踪
onStepStartedEvent({ event }) {
setSteps(prev => [
...prev,
{ name: event.step_name, status: "running" }
]);
},
onStepFinishedEvent({ event }) {
setSteps(prev =>
prev.map(s =>
s.name === event.step_name
? { ...s, status: "finished" }
: s
)
);
},
// 工具调用
onToolCallStartEvent({ event, toolCallName }) {
setToolCalls(prev => [
...prev,
{ name: toolCallName, args: "", status: "started" }
]);
},
onToolCallEndEvent({ event, toolCallName, toolCallArgs }) {
setToolCalls(prev =>
prev.map((tc, i) =>
i === prev.length - 1
? { ...tc, args: toolCallArgs, status: "finished" }
: tc
)
);
},
// 状态同步
onStateSnapshotEvent({ event, agent }) {
setAgentState(agent.state);
},
// 运行完成
onRunFinishedEvent({ event }) {
console.log("Agent run finished:", event.run_id);
},
});
return () => subscription.unsubscribe();
}, [agent]);
return (
<div className="agent-chat">
{/* 流式文本输出 */}
{streamingText && (
<div className="streaming-text">
<div className="avatar">🤖</div>
<div className="message">{streamingText}</div>
</div>
)}
{/* 步骤追踪 */}
{steps.length > 0 && (
<div className="steps-tracker">
<h4>执行步骤</h4>
{steps.map((step, i) => (
<div key={i} className={`step step-${step.status}`}>
{step.status === "running" ? "⏳" : "✅"} {step.name}
</div>
))}
</div>
)}
{/* 工具调用展示 */}
{toolCalls.length > 0 && (
<div className="tool-calls">
<h4>工具调用</h4>
{toolCalls.map((tc, i) => (
<div key={i} className={`tool-call tc-${tc.status}`}>
<span>{tc.status === "started" ? "🔧" : "✅"}</span>
<span className="tool-name">{tc.name}</span>
{tc.args && (
<pre className="tool-args">{tc.args}</pre>
)}
</div>
))}
</div>
)}
{/* Agent 状态 */}
{Object.keys(agentState).length > 0 && (
<div className="agent-state">
<h4>Agent 状态</h4>
<pre>{JSON.stringify(agentState, null, 2)}</pre>
</div>
)}
</div>
);
}
tsx
// frontend/src/components/ProgressPanel.tsx
import { useAgent } from "@copilotkit/react-core";
import { useEffect, useState } from "react";
export function ProgressPanel() {
const { agent } = useAgent({ agentId: "customer_service_agent" });
const [isRunning, setIsRunning] = useState(false);
const [lastActivity, setLastActivity] = useState<string>("");
useEffect(() => {
if (!agent) return;
const subscription = agent.subscribe({
onRunStartedEvent() {
setIsRunning(true);
setLastActivity("Agent 开始运行...");
},
onRunFinishedEvent() {
setIsRunning(false);
setLastActivity("Agent 运行完成");
},
onRunErrorEvent({ event }) {
setIsRunning(false);
setLastActivity(`运行错误: ${event.message}`);
},
onToolCallStartEvent({ toolCallName }) {
setLastActivity(`调用工具: ${toolCallName}`);
},
onStepStartedEvent({ event }) {
setLastActivity(`开始步骤: ${event.step_name}`);
},
});
return () => subscription.unsubscribe();
}, [agent]);
return (
<div className="progress-panel">
<h3>Agent 监控面板</h3>
<div className="status-row">
<span>状态：</span>
<span className={isRunning ? "status-running" : "status-idle"}>
{isRunning ? "运行中" : "空闲"}
</span>
</div>
<div className="activity-row">
<span>最近活动：</span>
<span>{lastActivity || "等待输入..."}</span>
</div>
<div className="message-count">
消息数: {agent?.messages?.length ?? 0}
</div>
</div>
);
}
6.4 Docker 部署
yaml
# docker-compose.yml
version: '3.8'
services:
backend:
build: ./backend
ports:
- "8000:8000"
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
volumes:
- ./backend:/app
command: uvicorn main:app --host 0.0.0.0 --port 8000 --reload
frontend:
build: ./frontend
ports:
- "5173:5173"
volumes:
- ./frontend:/app
- /app/node_modules
command: npm run dev -- --host
depends_on:
- backend
7. 实战2：将 LangGraph Agent 通过 AG-UI 接入前端
这个实战展示如何将已有的 LangGraph Agent 改造为 AG-UI 兼容端点，重点展示事件流的精细控制。
7.1 自定义 LangGraph Agent
python
# backend/research_agent.py
"""
研究型 LangGraph Agent
支持多步骤研究、实时进度汇报、中间结果预览
通过 AG-UI 事件流与前端实时交互
"""
import os
import json
import asyncio
from typing import TypedDict, Annotated, Sequence
from datetime import datetime
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from copilotkit.integrations.fastapi import add_fastapi_endpoint
from copilotkit import CopilotKitSDK, LangGraphAgent
# --- 定义状态 ---
class ResearchState(TypedDict):
messages: Annotated[Sequence[BaseMessage], add_messages]
research_topic: str
findings: list[str]
current_phase: str # "planning" | "searching" | "analyzing" | "synthesizing"
progress: float # 0.0 ~ 1.0
# --- 定义工具 ---
from langchain_core.tools import tool
@tool
def web_search(query: str) -> str:
"""搜索互联网获取信息"""
# 模拟搜索结果
mock_results = {
"AI Agent": "AI Agent 市场预计 2026 年达到 50 亿美元，年增长率 45%。主要玩家包括 OpenAI、Anthropic、Google。",
"MCP协议": "MCP（Model Context Protocol）由 Anthropic 提出，已有 1000+ 服务器实现，覆盖主流 SaaS 平台。",
"AG-UI": "AG-UI 是 CopilotKit 提出的 Agent-用户交互协议，GitHub 星标 9000+，支持 7 种 SDK 语言。",
}
results = []
for key, value in mock_results.items():
if key.lower() in query.lower() or query.lower() in key.lower():
results.append(value)
if not results:
return f"搜索 '{query}' 暂无直接结果，建议缩小搜索范围。"
return "\n".join(results)
@tool
def analyze_data(data_description: str) -> str:
"""分析数据并生成洞察"""
return f"数据分析完成：{data_description}\n关键发现：趋势向上，但存在季节性波动。建议持续监控。"
@tool
def generate_report(topic: str, findings: str) -> str:
"""生成研究报告"""
return f"""## {topic} 研究报告
### 摘要
{findings}
### 建议
1. 持续关注市场动态
2. 建立技术评估框架
3. 制定分阶段实施计划
---
报告生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}
"""
# --- 定义节点 ---
llm = ChatOpenAI(model="gpt-4o", temperature=0.2)
tools = [web_search, analyze_data, generate_report]
llm_with_tools = llm.bind_tools(tools)
def planning_node(state: ResearchState) -> dict:
"""规划研究步骤"""
messages = state["messages"]
topic = messages[-1].content if messages else ""
prompt = f"""你是一个研究助手。用户想要研究：{topic}
请制定一个研究计划，包括：
1. 需要搜索哪些信息
2. 需要分析哪些数据
3. 最终生成什么样的报告
用简洁的格式列出研究步骤。"""
response = llm.invoke([
SystemMessage(content=prompt),
HumanMessage(content=topic)
])
return {
"messages": [response],
"research_topic": topic,
"current_phase": "searching",
"progress": 0.2
}
def research_node(state: ResearchState) -> dict:
"""执行研究搜索"""
findings = state.get("findings", [])
# 使用工具进行搜索
response = llm_with_tools.invoke(state["messages"])
# 提取搜索结果
if hasattr(response, "tool_calls") and response.tool_calls:
for tc in response.tool_calls:
findings.append(f"[{tc['name']}] 已执行搜索")
return {
"messages": [response],
"findings": findings,
"current_phase": "analyzing",
"progress": 0.5
}
def analysis_node(state: ResearchState) -> dict:
"""分析研究结果"""
findings_text = "\n".join(state.get("findings", []))
prompt = f"""基于以下研究结果，进行深度分析：
{findings_text}
请总结关键洞察和趋势。"""
response = llm.invoke([
SystemMessage(content=prompt),
*state["messages"]
])
return {
"messages": [response],
"current_phase": "synthesizing",
"progress": 0.8
}
def synthesis_node(state: ResearchState) -> dict:
"""综合生成报告"""
findings_text = "\n".join(state.get("findings", []))
topic = state.get("research_topic", "")
response = llm_with_tools.invoke([
SystemMessage(content="请根据研究结果生成最终报告。使用 generate_report 工具。"),
*state["messages"],
HumanMessage(content=f"请为 '{topic}' 生成完整研究报告。关键发现：{findings_text}")
])
return {
"messages": [response],
"current_phase": "completed",
"progress": 1.0
}
# --- 构建图 ---
def should_continue(state: ResearchState) -> str:
"""决定下一步"""
phase = state.get("current_phase", "")
if phase == "searching":
return "research"
elif phase == "analyzing":
return "analysis"
elif phase == "synthesizing":
return "synthesis"
return END
# 创建 LangGraph
workflow = StateGraph(ResearchState)
# 添加节点
workflow.add_node("planning", planning_node)
workflow.add_node("research", research_node)
workflow.add_node("analysis", analysis_node)
workflow.add_node("synthesis", synthesis_node)
workflow.add_node("tools", ToolNode(tools))
# 设置边
workflow.set_entry_point("planning")
workflow.add_conditional_edges("planning", should_continue)
workflow.add_edge("research", "tools")
workflow.add_edge("tools", "analysis")
workflow.add_edge("analysis", "synthesis")
workflow.add_edge("synthesis", END)
# 编译
research_agent = workflow.compile()
7.2 注册到 AG-UI
python
# backend/research_main.py
"""
研究 Agent 的 AG-UI 服务端
"""
from fastapi import FastAPI
from copilotkit.integrations.fastapi import add_fastapi_endpoint
from copilotkit import CopilotKitSDK, LangGraphAgent
from research_agent import research_agent
app = FastAPI(title="AG-UI Research Agent")
sdk = CopilotKitSDK(
agents=[
LangGraphAgent(
name="research_agent",
agent=research_agent,
description="研究助手 Agent，支持多步骤研究、数据分析、报告生成",
)
]
)
add_fastapi_endpoint(app, sdk, "/api/copilotkit")
if __name__ == "__main__":
import uvicorn
uvicorn.run("research_main:app", host="0.0.0.0", port=8001, reload=True)
7.3 前端订阅研究进度
tsx
// 前端监听研究 Agent 的进度
import { useAgent } from "@copilotkit/react-core";
import { useEffect, useState } from "react";
interface ResearchProgress {
phase: string;
progress: number;
findings: string[];
}
export function ResearchTracker() {
const { agent } = useAgent({ agentId: "research_agent" });
const [progress, setProgress] = useState<ResearchProgress>({
phase: "idle",
progress: 0,
findings: []
});
useEffect(() => {
if (!agent) return;
const subscription = agent.subscribe({
// 监听状态快照 - 获取研究进度
onStateSnapshotEvent({ agent }) {
setProgress({
phase: agent.state?.current_phase ?? "idle",
progress: agent.state?.progress ?? 0,
findings: agent.state?.findings ?? []
});
},
// 监听状态增量更新
onStateDeltaEvent({ event }) {
// 根据 JSON Patch 更新本地状态
console.log("State delta:", event.delta);
},
// 步骤追踪
onStepStartedEvent({ event }) {
console.log(`Research step started: ${event.step_name}`);
},
onStepFinishedEvent({ event }) {
console.log(`Research step finished: ${event.step_name}`);
},
// 工具调用 - 展示 Agent 在用什么工具
onToolCallStartEvent({ toolCallName }) {
console.log(`Tool invoked: ${toolCallName}`);
},
});
return () => subscription.unsubscribe();
}, [agent]);
const phaseLabels: Record<string, string> = {
planning: "📋 规划中",
searching: "🔍 搜索中",
analyzing: "📊 分析中",
synthesizing: "📝 生成报告",
completed: "✅ 完成"
};
return (
<div className="research-tracker">
<h3>研究进度</h3>
{/* 进度条 */}
<div className="progress-bar">
<div
className="progress-fill"
style={{ width: `${progress.progress * 100}%` }}
/>
</div>
{/* 当前阶段 */}
<div className="phase">
{phaseLabels[progress.phase] || progress.phase}
</div>
{/* 研究发现 */}
{progress.findings.length > 0 && (
<div className="findings">
<h4>研究发现</h4>
<ul>
{progress.findings.map((f, i) => (
<li key={i}>{f}</li>
))}
</ul>
</div>
)}
</div>
);
}
8. 适用场景 vs 不适用场景
8.1 适用 AG-UI 的场景
plaintext
┌──────────────────────────────────────────────────────────────┐
│ ✅ 适合 AG-UI 的场景 │
├──────────────────────────────────────────────────────────────┤
│ │
│ 1. 用户协作型 Agent │
│ ├─ AI 客服（实时对话 + 工具调用 + 状态更新） │
│ ├─ 代码助手（终端协作 + 文件编辑 + 运行结果） │
│ └─ 数据分析师（进度汇报 + 图表渲染 + 中间预览） │
│ │
│ 2. 需要中间干预的任务 │
│ ├─ Human-in-the-loop 审批流程 │
│ ├─ 敏感操作确认（删除、支付、发布） │
│ └─ 多步骤任务的检查点恢复 │
│ │
│ 3. 多 Agent 编排场景 │
│ ├─ Agent 团队协作，用户需要看到整体进度 │
│ ├─ 子 Agent 结果需要实时展示 │
│ └─ Agent 间协作 + 用户交互同时存在 │
│ │
│ 4. 需要实时反馈的长时任务 │
│ ├─ 大规模代码重构（进度百分比 + 当前文件） │
│ ├─ 研究报告生成（阶段展示 + 中间结果预览） │
│ └─ 数据 ETL 流程（步骤追踪 + 数据预览） │
│ │
│ 共同特征：Agent 和用户需要实时协作/通信 │
│ │
└──────────────────────────────────────────────────────────────┘
8.2 不适用 AG-UI 的场景
plaintext
┌──────────────────────────────────────────────────────────────┐
│ ❌ 不需要 AG-UI 的场景 │
├──────────────────────────────────────────────────────────────┤
│ │
│ 1. 后台批处理任务 │
│ ├─ 定时数据同步（无需用户交互） │
│ ├─ 日志分析（跑完看结果就行） │
│ └─ 批量邮件发送（纯后端执行） │
│ │
│ 2. 一次性 API 调用 │
│ ├─ 简单问答（请求-响应模式足够） │
│ ├─ 图片生成（等结果返回即可） │
│ └─ 翻译任务（无中间步骤） │
│ │
│ 3. 纯后端 Agent-to-Agent 协作 │
│ ├─ 不涉及用户界面 │
│ ├─ 用 A2A 协议更合适 │
│ └─ 无需 SSE 流式推送 │
│ │
│ 4. 纯工具调用 │
│ ├─ Agent 只需要调 API，不需要和用户交互 │
│ ├─ 用 MCP 协议更合适 │
│ └─ 无需展示中间状态 │
│ │
│ 判断标准：如果没有"用户需要看到 Agent 在干什么"的需求， │
│ AG-UI 就是过度设计。 │
│ │
└──────────────────────────────────────────────────────────────┘
9. 生态支持现状
据《AG-UI Is Redefining the Agent–User Interaction Layer》和 Microsoft Agent Framework 文档，截至 2026 年中期：
9.1 支持方
表格
9.2 SDK 语言支持
表格
9.3 关键里程碑
plaintext
时间线：
2025-05 ─── AG-UI 正式发布
2025-07 ─── CopilotKit $27M A 轮融资
2025-09 ─── Google ADK 接入
2025-10 ─── GitHub 9000+ Stars，周安装量 12万
2025-11 ─── A2A ↔ AG-UI 官方握手
2026-04 ─── Microsoft Agent Framework 集成
2026-06 ─── CopilotKit 200万+ 周 Agent-用户交互
2026-09 ─── 7 种语言 SDK 稳定版
10. 踩坑记录
坑1：SSE 连接在 Nginx 反代下断开
问题： Nginx 默认会缓冲代理响应，导致 SSE 事件被缓存而不是实时推送到前端。
nginx
# ❌ 错误配置 - SSE 会被缓冲
location /api/copilotkit {
proxy_pass http://backend:8000;
}
# ✅ 正确配置 - 禁用缓冲，启用长连接
location /api/copilotkit {
proxy_pass http://backend:8000;
proxy_http_version 1.1;
proxy_set_header Connection '';
proxy_buffering off;
proxy_cache off;
proxy_read_timeout 86400s; # SSE 长连接超时
proxy_send_timeout 86400s;
chunked_transfer_encoding on;
}
坑2：useAgent Hook 获取不到 agent 实例
问题： 组件挂载时 agent 为 undefined，直接调用 agent.subscribe() 报错。
tsx
// ❌ 错误写法 - agent 可能还没初始化
function MyComponent() {
const { agent } = useAgent();
agent.subscribe({ ... }); // TypeError: Cannot read properties of undefined
}
// ✅ 正确写法 - 等待 agent 初始化
function MyComponent() {
const { agent } = useAgent();
useEffect(() => {
if (!agent) return; // 先判断
const subscription = agent.subscribe({ ... });
return () => subscription.unsubscribe();
}, [agent]); // 依赖 agent
}
坑3：STATE_DELTA 的 JSON Patch 不生效
问题： 后端发送了 STATE_DELTA 事件，但前端状态没有更新。
原因： JSON Patch 的 path 必须以 / 开头，且路径必须与现有状态结构匹配。
python
# ❌ 错误 - 路径不以 / 开头
{
"type": "StateDelta",
"delta": [
{"op": "replace", "path": "progress", "value": 0.8}
]
}
# ✅ 正确 - 路径以 / 开头
{
"type": "StateDelta",
"delta": [
{"op": "replace", "path": "/progress", "value": 0.8}
]
}
# ❌ 错误 - 向不存在的数组添加元素用 add
# 如果 files_processed 不存在，需要先创建
{
"type": "StateDelta",
"delta": [
{"op": "add", "path": "/files_processed/-", "value": "new.csv"}
]
}
# ✅ 正确 - 确保先有 StateSnapshot 初始化状态
# 先发送 StateSnapshot 创建完整状态结构
# 再发送 StateDelta 做增量更新
坑4：多 Agent 场景下事件流混淆
问题： 多个 Agent 同时运行，前端收到的事件无法区分来自哪个 Agent。
tsx
// ❌ 错误 - 全局监听不区分 Agent
const { agent } = useAgent(); // 默认获取第一个 agent
agent.subscribe({
onEvent({ event }) {
// 无法区分事件来源
}
});
// ✅ 正确 - 用 agentId 指定特定 Agent
const { agent: researchAgent } = useAgent({ agentId: "research_agent" });
const { agent: supportAgent } = useAgent({ agentId: "support_agent" });
useEffect(() => {
if (!researchAgent) return;
const sub = researchAgent.subscribe({
onEvent({ event }) {
console.log("Research agent event:", event);
}
});
return () => sub.unsubscribe();
}, [researchAgent]);
坑5：Human-in-the-Loop 实现不当导致死锁
问题： Agent 等待用户审批，但前端没有正确发送审批响应，导致 Agent 永远挂起。
python
# ❌ 错误 - 没有超时处理
@tool
def delete_user(user_id: str) -> str:
"""删除用户 - 需要审批"""
# 如果前端不响应审批，这里永远阻塞
approval = wait_for_approval(user_id)
if approval:
return f"用户 {user_id} 已删除"
return "操作已取消"
# ✅ 正确 - 设置超时 + 默认行为
import asyncio
@tool
def delete_user(user_id: str) -> str:
"""删除用户 - 需要审批"""
try:
approval = asyncio.wait_for(
wait_for_approval(user_id),
timeout=300 # 5分钟超时
)
if approval:
return f"用户 {user_id} 已删除"
return "操作已取消"
except asyncio.TimeoutError:
return "审批超时，操作已取消（安全默认行为）"
坑6：CORS 配置问题
问题： 前端（localhost:5173）请求后端（localhost:8000）的 AG-UI 端点被 CORS 拦截。
python
# ✅ 正确 - FastAPI 添加 CORS 中间件
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:5173"], # 开发环境
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
expose_headers=["Content-Type"], # SSE 需要暴露的 header
)
坑7：事件顺序依赖导致 UI 闪烁
问题： STATE_SNAPSHOT 和 STATE_DELTA 事件到达顺序不确定，导致 UI 闪烁（先显示旧状态再更新）。
tsx
// ✅ 解决方案 - 用版本号确保事件有序处理
const [stateVersion, setStateVersion] = useState(0);
const [agentState, setAgentState] = useState({});
useEffect(() => {
if (!agent) return;
const subscription = agent.subscribe({
onStateSnapshotEvent({ event }) {
// 全量快照 - 重置状态
setAgentState(event.snapshot);
setStateVersion(v => v + 1);
},
onStateDeltaEvent({ event }) {
// 增量更新 - 应用 JSON Patch
setAgentState(prev => applyPatch(prev, event.delta));
setStateVersion(v => v + 1);
},
});
return () => subscription.unsubscribe();
}, [agent]);
11. 总结与展望
AG-UI 解决了什么问题？
AG-UI 的核心价值在于标准化。就像 HTTP 标准化了客户端和服务器的通信一样，AG-UI 标准化了 Agent 和前端的通信。
没有 AG-UI 的世界：
每个 Agent 框架有自己的流式输出格式
前端为每个后端写适配器
工具调用不可见，状态同步靠猜测
Human-in-the-loop 需要自己实现全套
有 AG-UI 的世界：
统一的事件流模型，一套代码适配所有后端
工具调用、状态更新、进度汇报都有标准事件
Human-in-the-loop 是协议原生支持
切换后端框架（LangGraph → CrewAI）前端零改动
局限性
协议还年轻：2025年5月才发布，部分场景的事件定义还在迭代
生态覆盖不均：TypeScript/Python SDK 成熟，其他语言相对滞后
性能瓶颈：SSE 在极端高并发场景下不如 WebSocket 灵活
调试工具缺失：事件流的调试目前主要靠 console.log，缺乏可视化工具
展望
据 CopilotKit 团队的规划，AG-UI 接下来的重点方向是：
Rust SDK：高性能后端场景
更深的 MCP 集成：AG-UI 事件流中直接嵌入 MCP 工具发现和调用
多模态事件：图片、音频、视频的流式交互事件
协作编辑：Agent 和用户同时编辑同一文档的状态同步
AG-UI 是 Agent 生态从"能用"到"好用"的关键一环。当 Agent 需要和人协作时，一个标准化的交互协议不再是锦上添花，而是基础设施。
——PySuper | zhengxingtao.com
参考链接：
评论区