AG-UI 协议完全学习指南

AG-UI（Agent-User Interaction Protocol，智能体用户交互协议）是由 CopilotKit 发布的开源、轻量级、基于事件的协议，用于标准化 AI Agent 与前端应用之间的实时交互。

┌─────────────────────────────────────────────────────────────────┐
│                     AI Agent 交互协议生态                        │
├─────────────────────────────────────────────────────────────────┤
│   MCP = AI 的"工具箱"接口（Agent↔工具）                          │
│   A2A = AI 的"同事间"通信（Agent↔Agent）                         │
│   AG-UI = AI 的"用户界面"桥梁（Agent↔用户） ← 本文重点           │
└─────────────────────────────────────────────────────────────────┘

一、核心基础（Foundation）

1.1 定义与核心定位 ★重点★

什么是 AG-UI？

AG-UI（Agent-User Interaction Protocol）是一个：

开放的：完全开源，MIT 协议
轻量级的：协议设计简洁，易于实现
基于事件的：所有通信都通过标准化事件流进行

一句话定义：

AG-UI 是一个开放、轻量、事件驱动的协议，通过标准 HTTP 或可选的二进制通道，以流式方式传输 JSON 事件，实现 AI Agent 与用户界面的实时交互。

核心定位图

         ┌─────────────────────────────────────┐
         │           前端应用 (UI)              │
         │   React / Vue / Mobile / Terminal   │
         └──────────────────┬──────────────────┘
                            │
                     ┌──────▼──────┐
                     │   AG-UI     │  ← 标准化协议层
                     │   Protocol  │
                     └──────┬──────┘
                            │
         ┌──────────────────▼──────────────────┐
         │           AI Agent 后端              │
         │  LangGraph / CrewAI / OpenAI / 自建  │
         └─────────────────────────────────────┘

1.2 设计初衷与核心问题 ★重点★

解决的核心痛点

问题	传统方案	AG-UI 解决方案
碎片化	每个 Agent 框架有自己的 UI 接口	统一的事件协议标准
实时性差	请求-响应模式，用户等待结果	流式事件，实时反馈
状态不同步	Agent 状态与 UI 状态割裂	双向状态同步机制
扩展性差	紧耦合，换框架需重写 UI	协议标准化，框架可替换

1.3 适用场景 ★重点★

✅ 适用场景

场景	说明	示例
AI 聊天应用	需要流式输出、实时反馈	ChatGPT 类应用
AI 协作工具	人机协作、人在回路	AI 辅助编程
多步骤工作流	展示进度和中间状态	数据分析、报告生成
生成式 UI	Agent 动态生成界面	表单生成、图表展示

❌ 不适用场景

场景	原因	替代方案
Agent 间通信	AG-UI 专注于 Agent↔用户	使用 A2A 协议
工具调用定义	AG-UI 不定义工具能力	使用 MCP 协议
简单 API 调用	无需流式、无需状态同步	标准 REST API

1.4 与其他协议的对比 ★重点★

对比维度	AG-UI	MCP	A2A
发起者	CopilotKit	Anthropic	Google
核心定位	Agent↔用户交互	Agent↔工具连接	Agent↔Agent 协作
消息格式	JSON 事件流	JSON-RPC 2.0	JSON-RPC 2.0
传输方式	HTTP/SSE/WebSocket	stdio/HTTP/SSE	HTTP
核心能力	流式聊天、状态同步	工具调用、资源访问	任务委派、Agent 发现

★ 核心记忆点：三者是互补关系，不是竞争关系！

二、技术规范（Technical Specification）

2.1 核心架构 ★重点★

┌─────────────────────────────────────────────────────────────────┐
│                       AG-UI 核心架构                            │
├─────────────────────────────────────────────────────────────────┤
│  Application (应用层)     → 用户界面（React/Vue/Mobile）        │
│  AG-UI Client (客户端层)  → HttpAgent、事件处理                 │
│  Protocol (协议层)        → HTTP SSE / WebSocket / Binary       │
│  Agent (代理层)           → LangGraph / CrewAI / OpenAI         │
│  Secure Proxy (可选)      → API Key 管理、安全代理              │
└─────────────────────────────────────────────────────────────────┘

核心设计原则

原则	说明
事件驱动	所有通信基于事件流，定义 16+ 种标准事件
双向交互	支持 Agent↔前端双向通信
传输无关	支持 SSE/WebSocket/HTTP
框架无关	可对接任何 Agent 框架

2.2 数据传输格式 ★重点★ 【强制要求】

基础事件结构

typescript

interface BaseEvent {
  type: EventType;        // 【必选】事件类型
  timestamp?: number;     // 【可选】毫秒时间戳
  rawEvent?: any;         // 【可选】原始事件
}

事件类型枚举（16种核心事件）

类别	事件	说明
生命周期	RUN_STARTED, RUN_FINISHED, RUN_ERROR, STEP_STARTED, STEP_FINISHED	运行控制
文本消息	TEXT_MESSAGE_START, TEXT_MESSAGE_CONTENT, TEXT_MESSAGE_END	流式文本
工具调用	TOOL_CALL_START, TOOL_CALL_ARGS, TOOL_CALL_END, TOOL_CALL_RESULT	工具执行
状态管理	STATE_SNAPSHOT, STATE_DELTA, MESSAGES_SNAPSHOT	状态同步
特殊事件	RAW, CUSTOM	扩展能力

2.3 通信协议 ★重点★

支持的传输方式

传输方式	特点	适用场景
HTTP SSE	单向流式、兼容性好	Web 应用、调试
HTTP Binary	高性能、节省带宽	生产环境
WebSocket	双向实时、低延迟	实时协作

SSE 事件格式【强制要求】

data: {"type":"RUN_STARTED","threadId":"t1","runId":"r1"}\n\n
data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"m1","delta":"Hello"}\n\n
data: {"type":"RUN_FINISHED","threadId":"t1","runId":"r1"}\n\n

三、交互流程（Interaction Flow）

3.1 完整交互链路 ★重点★

用户输入 → RUN_STARTED → [文本流/工具调用/状态更新] → RUN_FINISHED
                                    ↓
                         STEP_STARTED → STEP_FINISHED（可选）

3.2 核心事件详解 ★重点★

3.2.1 生命周期事件

json

// RUN_STARTED
{"type": "RUN_STARTED", "threadId": "t1", "runId": "r1"}

// RUN_FINISHED
{"type": "RUN_FINISHED", "threadId": "t1", "runId": "r1", "result": {...}}

// RUN_ERROR
{"type": "RUN_ERROR", "message": "API rate limit", "code": "RATE_LIMIT"}

3.2.2 文本消息事件（Start → Content* → End）

json

{"type": "TEXT_MESSAGE_START", "messageId": "m1", "role": "assistant"}
{"type": "TEXT_MESSAGE_CONTENT", "messageId": "m1", "delta": "你好"}
{"type": "TEXT_MESSAGE_CONTENT", "messageId": "m1", "delta": "，我是AI"}
{"type": "TEXT_MESSAGE_END", "messageId": "m1"}

3.2.3 工具调用事件（Start → Args* → End → Result）

json

{"type": "TOOL_CALL_START", "toolCallId": "c1", "toolCallName": "search"}
{"type": "TOOL_CALL_ARGS", "toolCallId": "c1", "delta": "{\"query\":\"天气\"}"}
{"type": "TOOL_CALL_END", "toolCallId": "c1"}
{"type": "TOOL_CALL_RESULT", "toolCallId": "c1", "content": "晴天 25°C"}

3.2.4 状态管理事件

json

// 完整快照
{"type": "STATE_SNAPSHOT", "snapshot": {"user": {...}, "task": {...}}}

// 增量更新（JSON Patch RFC 6902）
{"type": "STATE_DELTA", "delta": [
  {"op": "replace", "path": "/task/progress", "value": 75}
]}

3.3 错误处理与重连

错误码	含义	处理方式
`RATE_LIMIT`	API 限流	指数退避重试
`TIMEOUT`	请求超时	立即重试
`CONNECTION_LOST`	连接丢失	自动重连

四、开发适配（Development）

4.1 支持的框架生态 ★重点★

类别	框架
Agent 框架	LangGraph, CrewAI, Google ADK, AWS Strands, Pydantic AI
SDK	TypeScript, Python, Kotlin, Go, Java, Rust
客户端	CopilotKit, React Native, Terminal

4.2 Python 服务端示例 ★重点★

python

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json, uuid
from datetime import datetime

app = FastAPI()

def timestamp():
    return int(datetime.now().timestamp() * 1000)

def encode_event(event: dict) -> str:
    return f"data: {json.dumps(event)}\n\n"

async def generate_events(messages: list):
    run_id, msg_id = str(uuid.uuid4()), str(uuid.uuid4())
    
    # 1. RUN_STARTED
    yield encode_event({"type": "RUN_STARTED", "runId": run_id})
    
    # 2. TEXT_MESSAGE_START
    yield encode_event({"type": "TEXT_MESSAGE_START", 
                        "messageId": msg_id, "role": "assistant"})
    
    # 3. TEXT_MESSAGE_CONTENT（模拟流式输出）
    for word in ["你好", "，", "我是", "AI", "助手", "！"]:
        yield encode_event({"type": "TEXT_MESSAGE_CONTENT",
                            "messageId": msg_id, "delta": word})
    
    # 4. TEXT_MESSAGE_END
    yield encode_event({"type": "TEXT_MESSAGE_END", "messageId": msg_id})
    
    # 5. RUN_FINISHED
    yield encode_event({"type": "RUN_FINISHED", "runId": run_id})

@app.post("/agent")
async def run_agent(request: dict):
    return StreamingResponse(
        generate_events(request.get("messages", [])),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}
    )

4.3 TypeScript 客户端示例

typescript

class AGUIClient {
  private baseUrl: string;
  
  async run(input: { messages: any[] }) {
    const response = await fetch(`${this.baseUrl}/agent`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(input),
    });
    
    const reader = response.body?.getReader();
    const decoder = new TextDecoder();
    let buffer = '';
    
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      
      buffer += decoder.decode(value, { stream: true });
      const lines = buffer.split('\n\n');
      buffer = lines.pop() || '';
      
      for (const line of lines) {
        if (line.startsWith('data: ')) {
          const event = JSON.parse(line.slice(6));
          this.handleEvent(event);
        }
      }
    }
  }
  
  handleEvent(event: any) {
    switch (event.type) {
      case 'TEXT_MESSAGE_CONTENT':
        process.stdout.write(event.delta);
        break;
      case 'RUN_ERROR':
        console.error('Error:', event.message);
        break;
    }
  }
}

五、实战与问题排查

5.1 典型落地场景

场景	关键事件流
AI 聊天	RUN_STARTED → TEXT_MESSAGE_* → RUN_FINISHED
AI 编程	RUN_STARTED → TOOL_CALL_* → TEXT_MESSAGE_* → RUN_FINISHED
多步骤任务	RUN_STARTED → STEP_* → STEP_* → RUN_FINISHED

5.2 常见问题排查 ★重点★

问题 1：事件延迟/批量到达

bash

# 检查点：
1. 响应头是否正确：Content-Type: text/event-stream
2. 是否禁用缓冲：X-Accel-Buffering: no
3. Nginx 配置：proxy_buffering off

问题 2：连接频繁断开

python

# 解决方案：添加心跳
async def event_generator_with_heartbeat():
    while True:
        yield "data: {...}\n\n"
        await asyncio.sleep(15)
        yield ": keepalive\n\n"  # 心跳（注释行）

问题 3：状态同步异常

python

# 解决方案：使用 JSON Patch 库
import jsonpatch
patch = jsonpatch.JsonPatch(delta['delta'])
new_state = patch.apply(current_state)

5.3 调试工具

工具	用途
curl	`curl -N http://localhost:8000/agent ...`
Chrome DevTools	Network → EventStream 标签
AG-UI Inspector	官方可视化调试工具

六、核心知识点速记清单

📋 30秒速记

项目	内容
协议全称	Agent-User Interaction Protocol
发起组织	CopilotKit
核心特性	开放、轻量、事件驱动
传输方式	HTTP SSE / WebSocket / Binary
消息格式	流式 JSON 事件
核心定位	AI Agent ↔ 前端应用的桥梁
事件数量	16+ 种标准事件
状态同步	快照 + JSON Patch 增量

📋 公式总结

AG-UI = 事件驱动 + 流式传输 + 双向状态同步

事件流模式：
- 文本：START → CONTENT* → END
- 工具：START → ARGS* → END → RESULT
- 生命周期：RUN_STARTED → [...] → RUN_FINISHED/ERROR

协议关系：
- MCP：Agent ↔ 工具（垂直）
- A2A：Agent ↔ Agent（水平）
- AG-UI：Agent ↔ 用户（前端）

七、开发避坑指南

❌ 常见错误

错误	正确做法
SSE 不加 `\n\n`	每个事件必须以 `\n\n` 结尾
忘记禁用缓冲	设置 `X-Accel-Buffering: no`
不处理断线重连	实现指数退避重试
状态直接覆盖	区分 SNAPSHOT（覆盖）和 DELTA（补丁）
忽略 RUN_ERROR	必须处理并提示用户

✅ 最佳实践

实践	说明
使用心跳	每 15-30 秒发送 `: keepalive\n\n`
事件顺序处理	按接收顺序处理，ID 关联同一流
优雅降级	SSE 不支持时回退到轮询
状态压缩	大状态用 DELTA，初始化用 SNAPSHOT

八、学习资源

官方资源

资源	地址
官方文档	https://docs.ag-ui.com
GitHub	https://github.com/ag-ui-protocol/ag-ui
Discord	官方社区交流

快速开始

bash

# 创建 AG-UI 应用
npx create-ag-ui-app my-agent-app

# 安装 Python SDK
pip install ag-ui

总结：AG-UI 是 AI Agent 生态中连接 Agent 与用户界面的关键协议，与 MCP（工具连接）、A2A（Agent 协作）形成完整的协议三角。掌握 AG-UI 的事件驱动模型和状态同步机制，是构建现代 AI 应用的必备技能。

AG-UI 协议完全学习指南 ​

一、核心基础（Foundation） ​

1.1 定义与核心定位 ★重点★ ​

什么是 AG-UI？ ​

核心定位图 ​

1.2 设计初衷与核心问题 ★重点★ ​

解决的核心痛点 ​

1.3 适用场景 ★重点★ ​

✅ 适用场景 ​

❌ 不适用场景 ​

1.4 与其他协议的对比 ★重点★ ​

二、技术规范（Technical Specification） ​

2.1 核心架构 ★重点★ ​

核心设计原则 ​

2.2 数据传输格式 ★重点★ 【强制要求】 ​

基础事件结构 ​

事件类型枚举（16种核心事件） ​

2.3 通信协议 ★重点★ ​

支持的传输方式 ​

SSE 事件格式 【强制要求】 ​

三、交互流程（Interaction Flow） ​

3.1 完整交互链路 ★重点★ ​

3.2 核心事件详解 ★重点★ ​

3.2.1 生命周期事件 ​

3.2.2 文本消息事件（Start → Content* → End） ​

3.2.3 工具调用事件（Start → Args* → End → Result） ​

3.2.4 状态管理事件 ​

3.3 错误处理与重连 ​

四、开发适配（Development） ​

4.1 支持的框架生态 ★重点★ ​

4.2 Python 服务端示例 ★重点★ ​

4.3 TypeScript 客户端示例 ​

五、实战与问题排查 ​

5.1 典型落地场景 ​

5.2 常见问题排查 ★重点★ ​

问题 1：事件延迟/批量到达 ​

问题 2：连接频繁断开 ​

问题 3：状态同步异常 ​

5.3 调试工具 ​

六、核心知识点速记清单 ​

📋 30秒速记 ​

📋 公式总结 ​

七、开发避坑指南 ​

❌ 常见错误 ​

✅ 最佳实践 ​

八、学习资源 ​

官方资源 ​

快速开始 ​

AG-UI 协议完全学习指南

一、核心基础（Foundation）

1.1 定义与核心定位 ★重点★

什么是 AG-UI？

核心定位图

1.2 设计初衷与核心问题 ★重点★

解决的核心痛点

1.3 适用场景 ★重点★

✅ 适用场景

❌ 不适用场景

1.4 与其他协议的对比 ★重点★

二、技术规范（Technical Specification）

2.1 核心架构 ★重点★

核心设计原则

2.2 数据传输格式 ★重点★ 【强制要求】

基础事件结构

事件类型枚举（16种核心事件）

2.3 通信协议 ★重点★

支持的传输方式

SSE 事件格式【强制要求】

三、交互流程（Interaction Flow）

3.1 完整交互链路 ★重点★

3.2 核心事件详解 ★重点★

3.2.1 生命周期事件

3.2.2 文本消息事件（Start → Content* → End）

3.2.3 工具调用事件（Start → Args* → End → Result）

3.2.4 状态管理事件

3.3 错误处理与重连

四、开发适配（Development）

4.1 支持的框架生态 ★重点★

4.2 Python 服务端示例 ★重点★

4.3 TypeScript 客户端示例

五、实战与问题排查

5.1 典型落地场景

5.2 常见问题排查 ★重点★

问题 1：事件延迟/批量到达

问题 2：连接频繁断开

问题 3：状态同步异常

5.3 调试工具

六、核心知识点速记清单

📋 30秒速记

📋 公式总结

七、开发避坑指南

❌ 常见错误

✅ 最佳实践

八、学习资源

官方资源

快速开始