AG-UI：Agent用户交互协议

作者：easy

15 次浏览

1 次

2025-8-15

编辑推荐:

本文主要介绍了Agent用户交互协议相关内容。希望对你的学习有帮助。
本文来自于微信公众号奇舞精选，由火龙果软件Linda编辑，推荐。

随着多智能体系统（Multi-Agent Systems）和大语言模型（LLMs）技术的飞速发展，AI Agent 的工作能力越来越强。然而，在落地到真实应用中时，我们却发现一项关键能力仍旧缺失——如何让智能体与用户进行顺畅、高效的交互？

CopilotKit 团队提出的 AG-UI 协议（Agent-User Interaction Protocol），正是为此而生。

一、为什么需要 AG-UI？

当前市面上的 AI 系统，大多聚焦在后端的 Agent 执行、工具调用、模型编排等逻辑层面。然而当这些智能体要接入前端界面、嵌入产品时，开发者却面临种种困境：

每个 Agent 框架（如 LangGraph、CrewAI、Mastra 等）都有自己的事件机制和 API；

不同模型接口格式不同，前端开发需重复适配；

实时流式交互困难，前后端同步机制不统一；

用户难以对 Agent 执行过程进行实时干预或控制；

工具调用、状态共享、安全校验等逻辑无统一抽象。

这些问题直接影响 Agent 系统的可用性与用户体验。

二、AG-UI 是什么？

AG-UI 是一个开源、轻量的协议，旨在规范 AI Agent 与前端用户界面之间的通信流程。

协议亮点：

基于标准 HTTP 通信（支持 SSE / WebSocket / 二进制）；

所有数据通过 JSON 编码的事件（Event）进行传输；

支持前后端实时双向通信；

提供 TypeScript 与 Python SDK，可快速集成；

可兼容主流 Agent 框架（LangGraph、CrewAI、Mastra、AG2 等）；

架构

三、AG-UI 核心能力

能力模型	说明
✅统一事件流	所有交互统一采用结构化的 JSON 事件格式，降低前后端适配成本
✅实时交互	支持 token-by-token 的流式推送，提供极致响应式的用户体验
✅工具编排	Agent 执行过程中的 Tool 调用全过程均可被标准事件表示并渲染
✅状态共享	提供完整快照 STATE_SNAPSHOT 与增量更新 STATE_DELTA，高效同步状态
✅并发控制与中断	支持线程管理、任务取消、重启等机制，提升系统的可控性与稳定性
✅安全控制	协议内建权限管理、身份认证等机制，适配企业级安全需求

这些能力构成了 AG-UI 成为生产级 Agent 应用的关键基础。

四、事件机制

AG-UI 协议中一切交互都围绕“事件”进行组织。核心事件类别包括：

生命周期事件（如 RUN_STARTED, RUN_FINISHED），监控Agent运行进度。

文本消息事件（如 TEXT_MESSAGE_START, TEXT_MESSAGE_CONTENT, TEXT_MESSAGE_END），处理文本流式内容的事件。

工具调用事件（如 TOOL_CALL_START, TOOL_CALL_ARGS, TOOL_CALL_END），管理 Agent 对工具的执行。

状态管理事件（如 STATE_SNAPSHOT, STATE_DELTA），同步 Agent 和 UI 之间的状态。

特殊事件（如 RAW, CUSTOM）

这一机制不仅提供了强大可扩展性，还简化了前端的 UI 渲染逻辑。

五、与其他协议的关系

AG-UI 与 A2A（Agent-to-Agent）和 MCP（Model Context Protocol）形成互补：

A2A：主要促进智能体（agent-to-agent）之间的通信和协作；

MCP：主要解决跨不同模型之间工具调用的标准化和上下文处理问题；

AG-UI：主要处理由用户（人）参与的交互以及流式更新用户界面；

六、工作流程

AG-UI 的工作流程基于事件驱动架构，主要包括以下几个步骤：

前端发送请求：

用户在前端界面（如聊天窗口）输入信息。

前端应用将用户输入封装为 RunAgentInput 类型的 JSON 请求，发送到后端的 /awp 端点。

后端处理请求：

后端接收到请求后，解析 RunAgentInput，提取 thread_id、run_id 和用户消息等信息。

后端启动 AI 代理的处理流程，并准备通过事件流向前端发送处理状态和结果。

事件流通信：

后端通过 Server-Sent Events（SSE）或 WebSocket 等协议，向前端持续发送事件。

事件包括但不限于：

(1)RunStartedEvent：表示代理开始处理请求。

(2)TextMessageContentEvent：代理生成的文本内容。

(3)RunFinishedEvent：代理完成处理。

4. 前端更新界面：

前端接收到事件后，根据事件类型和内容，实时更新用户界面，展示代理的处理过程和结果。

例子

创建一个基础服务

mkdir awp-endpoint && cd awp-endpoint
npm init -y
npm install typescript ts-node @types/node @types/express --save-dev
npx tsc --init

npm install express openai @ag-ui/core @ag-ui/encoder uuid
npm install @types/uuid --save-dev

import express from"express"
import { Request, Response } from"express"

const app = express()

app.use(express.json())

app.post("/awp", (req: Request, res: Response) => {
  res.json({ message: "Hello World" })
})

app.listen(8000, () => {
console.log("Server running on http://localhost:8000")
})

解析 AG-UI 输入并添加事件流（SSE）

import express, { Request, Response } from"express"
import { RunAgentInputSchema, RunAgentInput, EventType } from"@ag-ui/core"
import { EventEncoder } from"@ag-ui/encoder"

const app = express()

app.use(express.json())

app.post("/awp", async (req: Request, res: Response) => {
try {
    // 解析并验证请求体
    const input: RunAgentInput = RunAgentInputSchema.parse(req.body)

    // 设置 SSE headers
    res.setHeader("Content-Type", "text/event-stream")
    res.setHeader("Cache-Control", "no-cache")
    res.setHeader("Connection", "keep-alive")

    // 创建事件 encoder
    const encoder = new EventEncoder()

    // 发送 started 事件
    const runStarted = {
      type: EventType.RUN_STARTED,
      threadId: input.threadId,
      runId: input.runId,
    }
    res.write(encoder.encode(runStarted))

    // 发送 finished 事件
    const runFinished = {
      type: EventType.RUN_FINISHED,
      threadId: input.threadId,
      runId: input.runId,
    }
    res.write(encoder.encode(runFinished))

    // 结束响应
    res.end()
  } catch (error) {
    res.status(422).json({ error: (error asError).message })
  }
})

app.listen(8000, () => {
console.log("Server running on http://localhost:8000")
})

集成 OpenAI

import express, { Request, Response } from"express"
import {
  RunAgentInputSchema,
  RunAgentInput,
  EventType,
  Message,
} from"@ag-ui/core"
import { EventEncoder } from"@ag-ui/encoder"
import { OpenAI } from"openai"
import { v4 as uuidv4 } from"uuid"

const app = express()

app.use(express.json())

app.post("/awp", async (req: Request, res: Response) => {
try {
    // 解析并验证请求体
    const input: RunAgentInput = RunAgentInputSchema.parse(req.body)

    // 设置 SSE headers
    res.setHeader("Content-Type", "text/event-stream")
    res.setHeader("Cache-Control", "no-cache")
    res.setHeader("Connection", "keep-alive")

    // 创建事件 encoder
    const encoder = new EventEncoder()

    // 发送 started 事件
    const runStarted = {
      type: EventType.RUN_STARTED,
      threadId: input.threadId,
      runId: input.runId,
    }
    res.write(encoder.encode(runStarted))


    // 初始化 OpenAI 客户端
    const client = new OpenAI()

    // 将 AG-UI 消息转换为 OpenAI 消息格式
    const openaiMessages = input.messages
      .filter((msg: Message) =>
        ["user", "system", "assistant"].includes(msg.role)
      )
      .map((msg: Message) => ({
        role: msg.role as"user" | "system" | "assistant",
        content: msg.content || "",
      }))

    // 生成消息 ID
    const messageId = uuidv4()

    // 发送 ‘文本消息开始’ 事件
    const textMessageStart = {
      type: EventType.TEXT_MESSAGE_START,
      messageId,
      role: "assistant",
    }
    res.write(encoder.encode(textMessageStart))

    // 创建流式传输完成请求
    const stream = await client.chat.completions.create({
      model: "gpt-3.5-turbo",
      messages: openaiMessages,
      stream: true,
    })

    // 处理流并发送 ‘文本消息内容’ 事件
    forawait (const chunk of stream) {
      if (chunk.choices[0]?.delta?.content) {
        const content = chunk.choices[0].delta.content
        const textMessageContent = {
          type: EventType.TEXT_MESSAGE_CONTENT,
          messageId,
          delta: content,
        }
        res.write(encoder.encode(textMessageContent))
      }
    }

    // 发送 ‘文本消息结束’ 事件
    const textMessageEnd = {
      type: EventType.TEXT_MESSAGE_END,
      messageId,
    }
    res.write(encoder.encode(textMessageEnd))


    // 发送 finished 事件
    const runFinished = {
      type: EventType.RUN_FINISHED,
      threadId: input.threadId,
      runId: input.runId,
    }
    res.write(encoder.encode(runFinished))

    // 结束响应
    res.end()
  } catch (error) {
    res.status(422).json({ error: (error asError).message })
  }
})

app.listen(8000, () => {
console.log("Server running on http://localhost:8000")
})