AG-UI Protocol 深度解析：AI Agent 與使用者介面的標準化橋樑

前言：Agentic 應用程式的通訊挑戰

當我們構建 AI Agent 應用程式時，一個核心挑戰浮現水面：如何讓運行中的 Agent 與使用者介面進行流暢的互動？傳統的請求-回應模式顯然不足以應對這個場景。Agent 可能需要數分鐘甚至數小時來完成複雜任務，在這個過程中使用者期待看到即時進度、中間結果，並希望能夠中斷、調整或提供反饋。

這正是 AG-UI（Agent-User Interaction Protocol） 試圖解決的問題。AG-UI 是一個開放、輕量級、基於事件的協議，旨在標準化 AI Agent 與使用者面對面應用程式之間的連接方式。本文將從架構設計、事件系統、整合範例等多個維度，深入探索這套協議的設計哲學與實作細節。

1. AG-UI 在 Agentic 協議棧中的定位

在深入技術細節之前，我們需要先理解 AG-UI 在整個 Agentic 生態系統中的位置。現代 AI Agent 應用涉及多層面的通訊需求，而不同的協議各司其職：

協議全稱職責角色
MCPModel Context Protocol為 Agent 提供工具與資料存取能力Agent ↔ 工具/知識庫
A2AAgent to Agent Protocol實現 Agent 之間的協作通訊Agent ↔ Agent
AG-UIAgent-User Interaction Protocol連接 Agent 與使用者介面Agent ↔ 使用者應用

這三個協議形成了互補的關係：MCP 讓 Agent 具備執行能力，A2A 讓多個 Agent 能夠協作，而 AG-UI 則將 Agent 的能力帶到使用者面前。沒有 AG-UI，Agent 的智慧將被困在後端，無法直接與使用者互動。

2. 核心設計理念

2.1 事件驅動架構

AG-UI 的核心設計圍繞「事件」這個概念展開。當 Agent 執行任務時，它會 emit 各種類型的事件，這些事件被傳送到前端應用程式，前端根據事件類型更新 UI。這個設計有幾個關鍵優勢：

第一，支援長期運行的任務。Agent 不需要在完成所有計算後才回應，而是可以邊計算邊發送事件，使用者能夠看到即時進度。

第二，非同步與串流。透過事件流，AG-UI 自然支援 Server-Sent Events（SSE）和 WebSocket 等串流協議，實現真正的即時通訊。

第三，狀態追蹤。每個事件都攜帶足夠的上下文資訊，前端可以準確追蹤對話狀態、工具呼叫進度等。

2.2 傳輸無關性

AG-UI 刻意與底層傳輸協議保持獨立。協議規範不強制要求使用 SSE 或 WebSocket，而是定義事件的格式與語義，具體的傳輸方式可以根據場景靈活選擇：

  • HTTP SSE：最常見的選擇，簡單易用，適合大多數場景
  • WebSocket：雙向通訊，適合需要即時雙向互動的場景
  • HTTP Long Polling：作為 SSE 的替代方案
  • 自定義傳輸：任何能夠傳輸事件的機制

這種設計讓 AG-UI 可以適配各種部署環境和技術棧。

2.3 格式靈活性

AG-UI 的一個重要特點是「寬鬆的格式匹配」。協議不要求後端發送的事件必須完全符合 AG-UI 規範，而是透過 middleware 層進行轉換。這意味著：

  • 現有的 Agent 框架（LangGraph、CrewAI 等）不需要大幅修改就能支援 AG-UI
  • 每個框架可以保持自己的事件格式，只需提供轉接層
  • 前端看到的永遠是標準化的 AG-UI 事件

3. 事件類型詳解

AG-UI 定義了大約 16 種標準事件類型，覆蓋了 Agent 與使用者互動的各個方面。這些事件可以分為幾個主要類別：

3.1 執行生命週期事件

enum EventType {
  RUN_STARTED = "RUN_STARTED", // Agent 開始執行
  RUN_FINISHED = "RUN_FINISHED", // Agent 完成執行
  RUN_ERROR = "RUN_ERROR", // 執行過程中的錯誤
  STEP_STARTED = "STEP_STARTED", // 單一步驟開始
  STEP_FINISHED = "STEP_FINISHED", // 單一步驟完成
}

這些事件讓前端能夠追蹤整體執行進度，顯示載入指示器或進度條。

3.2 訊息相關事件

enum EventType {
  TEXT_MESSAGE_START = "TEXT_MESSAGE_START", // 開始一條新訊息
  TEXT_MESSAGE_CONTENT = "TEXT_MESSAGE_CONTENT", // 訊息內容片段
  TEXT_MESSAGE_END = "TEXT_MESSAGE_END", // 訊息完成
}

透過這種分段傳輸機制，前端可以實現打字機效果，讓使用者感覺 Agent 正在「思考」而非等待回應。

3.3 工具呼叫事件

enum EventType {
  TOOL_CALL_START = "TOOL_CALL_START", // 開始呼叫工具
  TOOL_CALL_ARGS = "TOOL_CALL_ARGS", // 工具參數
  TOOL_CALL_END = "TOOL_CALL_END", // 工具呼叫完成
}

工具呼叫事件的支援讓前端可以顯示「Agent 正在使用某工具」的視覺提示，提升使用者信任感。

3.4 狀態管理事件

這是 AG-UI 最具特色的設計之一：

enum EventType {
  STATE_SNAPSHOT = "STATE_SNAPSHOT", // 完整狀態快照
  STATE_DELTA = "STATE_DELTA", // 狀態增量更新（JSON Patch）
  MESSAGES_SNAPSHOT = "MESSAGES_SNAPSHOT", // 對話歷史快照
}

STATE_SNAPSHOT 傳輸完整的狀態物件，適用於初始化或狀態重置。STATE_DELTA 則使用 RFC 6902 定義的 JSON Patch 格式，僅傳輸變更的部分：

{
  "type": "STATE_DELTA",
  "delta": [
    { "op": "add", "path": "/messages/-", "value": {...} },
    { "op": "replace", "path": "/data/count", "value": 42 }
  ]
}

這種設計大幅減少網路傳輸量，特別適合狀態頻繁更新的場景。

3.5 自定義與原始事件

enum EventType {
  CUSTOM = "CUSTOM", // 自定義事件類型
  RAW = "RAW", // 原始事件（直通）
}

這兩個事件類型提供了擴展性，讓開發者可以傳輸 AG-UI 規範未定義的事件，或直接傳遞框架特定的事件。

4. 技術架構與核心元件

4.1 客戶端 SDK 架構

AG-UI 的 TypeScript SDK 採用清晰的物件導向設計。核心抽象是 AbstractAgent 類別：

// 簡化版的核心類別結構
abstract class AbstractAgent {
  public threadId: string;
  public messages: Message[];
  public state: State;
  public subscribers: AgentSubscriber[] = [];
  public isRunning: boolean = false;
  private middlewares: Middleware[] = [];

  // 所有 Agent 必須實現的方法
  abstract run(input: RunAgentInput): Observable<BaseEvent>;

  // 執行並處理事件流
  public async runAgent(
    parameters?: RunAgentParameters,
    subscriber?: AgentSubscriber
  ): Promise<RunAgentResult> {
    const pipeline = pipe(
      () => this.run(input), // 執行 Agent
      transformChunks(this.debug), // 轉換區塊事件
      verifyEvents(this.debug), // 驗證事件格式
      (source$) => this.apply(input, source$, subscribers),
      (source$) => this.processApplyEvents(input, source$, subscribers),
      catchError((error) => this.onError(input, error, subscribers))
    );

    return await lastValueFrom(pipeline(of(null)));
  }

  // 使用 middleware 擴展功能
  public use(...middlewares: Middleware[]): this {
    this.middlewares.push(...middlewares);
    return this;
  }
}

AbstractAgent 採用 RxJS Observable 模式處理事件流，這帶來了幾個重要優勢：

  1. 組合性：多個事件流可以優雅地組合
  2. 錯誤處理：透過 catchError 運算子統一處理錯誤
  3. 取消支援：訂閱可以被取消，適用於使用者中斷場景
  4. 豐富的轉換：可以對事件流進行 map、filter、reduce 等操作

4.2 實現 HTTP Agent

對於大多數場景，HttpAgent 是最常用的客戶端實現：

export class HttpAgent extends AbstractAgent {
  public url: string;
  public headers: Record<string, string>;

  run(input: RunAgentInput): Observable<BaseEvent> {
    const httpEvents = runHttpRequest(this.url, this.requestInit(input));
    return transformHttpEventStream(httpEvents);
  }

  protected requestInit(input: RunAgentInput): RequestInit {
    return {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        Accept: "text/event-stream",
      },
      body: JSON.stringify(input),
      signal: this.abortController.signal,
    };
  }
}

使用範例：

const agent = new HttpAgent({
  url: "https://api.example.com/agent",
  threadId: "user-session-123",
});

const result = await agent.runAgent({
  messages: [{ role: "user", content: "分析這份報告" }],
  tools: [{ name: "search", description: "...", parameters: {...} }],
});

4.3 事件處理管道

當事件從 Agent 傳來時，客戶端會經過一系列處理步驟：

// 事件處理流程
const processEvents = (events$: Observable<BaseEvent>) => {
  return events$.pipe(
    concatMap(async (event) => {
      switch (event.type) {
        case EventType.TEXT_MESSAGE_START:
          // 建立新訊息
          messages.push({ id: messageId, role, content: "" });
          break;

        case EventType.TEXT_MESSAGE_CONTENT:
          // 附加內容片段
          targetMessage.content += event.delta;
          break;

        case EventType.STATE_DELTA:
          // 使用 JSON Patch 應用增量更新
          const result = applyPatch(state, event.delta, true, false);
          state = result.newDocument;
          break;

        case EventType.TOOL_CALL_START:
          // 記錄工具呼叫開始
          break;
      }
    })
  );
};

4.4 SSE 串流解析

SSE（Server-Sent Events）是 AG-UI 最常用的傳輸方式。客戶端需要解析 SSE 格式：

export const parseSSEStream = (
  source$: Observable<HttpEvent>
): Observable<any> => {
  const jsonSubject = new Subject<any>();
  const decoder = new TextDecoder("utf-8", { fatal: false });
  let buffer = "";

  source$.subscribe({
    next: (event) => {
      if (event.type === HttpEventType.DATA && event.data) {
        const text = decoder.decode(event.data, { stream: true });
        buffer += text;

        // 按雙換行分割事件
        const events = buffer.split(/\n\n/);
        buffer = events.pop() || "";

        for (const eventText of events) {
          const event = parseSSEEvent(eventText);
          if (event) jsonSubject.next(event);
        }
      }
    },
  });

  return jsonSubject.asObservable();
};

function parseSSEEvent(eventText: string): any | null {
  const lines = eventText.split("\n");
  const dataLines: string[] = [];

  for (const line of lines) {
    if (line.startsWith("data:")) {
      dataLines.push(line.slice(5).replace(/^ /, ""));
    }
  }

  if (dataLines.length > 0) {
    return JSON.parse(dataLines.join("\n"));
  }
  return null;
}

5. 框架整合實例

5.1 LangGraph 整合

LangGraph 是最流行的 Agent 框架之一，AG-UI 提供了完整的整合方案。整合的核心是將 LangGraph 的事件轉換為 AG-UI 事件：

export class LangGraphAgent extends AbstractAgent {
  private client: LangGraphClient;
  private messagesInProcess: MessagesInProgressRecord;

  async runAgentStream(
    input: RunAgentExtendedInput,
    subscriber: Subscriber<ProcessedEvents>
  ) {
    // 準備 LangGraph 串流
    const preparedStream = await this.prepareStream(
      {
        ...input,
        threadId: input.threadId || this.threadId,
      },
      streamMode
    );

    // 處理 LangGraph 事件並轉換為 AG-UI 格式
    await this.handleStreamEvents(
      preparedStream,
      input.threadId || this.threadId,
      subscriber,
      input
    );
  }

  private handleSingleEvent(event: any): void {
    switch (event.event) {
      case "on_chat_model_stream":
        // 轉換模型輸出為文字訊息內容
        this.dispatchEvent({
          type: EventType.TEXT_MESSAGE_CONTENT,
          messageId: currentMessage.id,
          delta: chunk.content,
        });
        break;

      case "on_tool_start":
        // 工具呼叫開始
        this.dispatchEvent({
          type: EventType.TOOL_CALL_START,
          toolCallId: toolCall.id,
          toolCallName: toolCall.name,
        });
        break;

      case "on_tool_end":
        // 工具呼叫完成
        this.dispatchEvent({
          type: EventType.TOOL_CALL_END,
          toolCallId: toolCall.id,
        });
        break;
    }
  }
}

// 使用 LangGraph Agent
const agent = new LangGraphAgent({
  graphId: "research-agent",
  deploymentUrl: "https://langgraph.example.com",
});

await agent.runAgent({
  messages: [{ role: "user", content: "研究 AI 發展趨勢" }],
});

5.2 CrewAI 整合

CrewAI 專注於多 Agent 協作，AG-UI 整合讓這些 Agent 的互動過程對使用者可見：

export class CrewAIAgent extends AbstractAgent {
  async run(input: RunAgentInput): Observable<BaseEvent> {
    // 啟動 CrewAI 工作流程
    const crewExecution = await this.crew.kickoff(input.messages);

    return new Observable((subscriber) => {
      // 轉發 CrewAI 事件為 AG-UI 事件
      for (const event of crewExecution.events) {
        const agEvent = this.translateToAGUI(event);
        subscriber.next(agEvent);
      }
      subscriber.complete();
    });
  }
}

5.3 自定義 HTTP Agent

如果你的 Agent 是自定義實現，可以直接使用 HTTP 介面：

import { HttpAgent, subscribe } from "@ag-ui/client";

// 建立 Agent 連接
const agent = new HttpAgent({
  url: "https://your-agent.example.com/stream",
  threadId: "conversation-001",
});

// 訂閱事件
agent.subscribe({
  onTextMessageContentEvent: (params) => {
    // 顯示 Agent 回覆片段
    appendToChat(params.messageId, params.delta);
  },

  onToolCallStartEvent: (params) => {
    // 顯示工具呼叫 UI
    showToolCallIndicator(params.toolCallName);
  },

  onRunFinishedEvent: () => {
    // 完成處理
    hideLoadingIndicator();
  },
});

// 執行
await agent.runAgent({
  messages: [{ role: "user", content: "幫我寫一段程式碼" }],
});

6. Middleware 系統

AG-UI 提供了強大的 middleware 系統，允許在事件處理過程中注入自定義邏輯：

// 自定義 Middleware 範例：過濾特定工具呼叫
export const filterToolCalls = ({ toolNames }: { toolNames: string[] }) => {
  return new (class implements Middleware {
    run(input: RunAgentInput, next: AbstractAgent): Observable<BaseEvent> {
      return next.run(input).pipe(
        filter((event) => {
          if (event.type === EventType.TOOL_CALL_START) {
            return toolNames.includes(event.toolCallName);
          }
          return true;
        })
      );
    }
  })();
};

// 使用 Middleware
agent.use(
  filterToolCalls({ toolNames: ["search", "calculator"] }),
  loggingMiddleware
);

常見的 middleware 應用場景：

  • 日誌記錄：追蹤所有事件用於除錯
  • 錯誤處理：統一處理錯誤事件
  • 事件轉換：修改事件內容
  • 過濾：移除不需要的事件
  • 節流：控制事件觸發頻率

7. Human-in-the-Loop 支援

AG-UI 的一個重要功能是支援 Human-in-the-Loop 互動，讓使用者在 Agent 執行過程中進行干預：

agent.subscribe({
  onCustomEvent: async (params) => {
    if (params.event.name === "approval_required") {
      // 顯示批准對話框
      const shouldProceed = await showApprovalDialog(
        params.event.description,
        params.event.details
      );

      // 根據用戶決定繼續或停止
      if (shouldProceed) {
        await agent.runAgent({
          forwardedProps: {
            command: { type: "continue", data: params.event.data },
          },
        });
      } else {
        await agent.runAgent({
          forwardedProps: {
            command: { type: "cancel" },
          },
        });
      }
    }
  },
});

這種模式適用於：

  • 敏感操作需要用戶批准
  • 選擇多個選項中的一個
  • 提供額外資訊給 Agent
  • 中斷並重新引導 Agent 方向

8. 狀態管理深度解析

8.1 Snapshot-Delta 模式

AG-UI 的狀態管理採用 Snapshot-Delta 模式，這是同步狀態的高效方式：

// 初始化時的全量狀態
{
  type: "STATE_SNAPSHOT",
  snapshot: {
    messages: [...],
    formData: { name: "", email: "" },
    selectedItems: [],
  }
}

// 使用者操作後的增量更新
{
  type: "STATE_DELTA",
  delta: [
    { op: "replace", path: "/formData/name", value: "John" },
    { op: "add", path: "/selectedItems/-", value: "item-1" },
  ]
}

這種設計的優勢：

  1. 頻寬效率：僅傳輸變更部分
  2. 一致性：每次更新都是原子性的
  3. 可追蹤：完整的變更歷史

8.2 多次執行支援

AG-UI 支援在單一對話中進行多次執行：

// 第一次執行
await agent.runAgent({
  messages: [{ role: "user", content: "先做這個" }],
});

// 第二次執行（狀態延續）
await agent.runAgent({
  messages: [{ role: "user", content: "再做那個" }],
});

// 狀態繼續累積，對話歷史完整保留

9. SDK 生態與多語言支援

AG-UI 提供多種程式語言的 SDK：

SDK語言狀態類型
TypeScriptJavaScript/TypeScript穩定官方
PythonPython穩定官方
KotlinAndroid/JVM穩定社群
GoGolang穩定社群
RustRust穩定社群
JavaJava穩定社群
DartFlutter穩定社群
.NETC#開發中社群

這種多語言支援讓 AG-UI 可以應用於各種技術環境，從 Web 應用到行動 App 乃至嵌入式系統。

10. 整合生態系統

AG-UI 已經與多個主流 Agent 框架完成整合：

官方整合（1st Party）

  • Microsoft Agent Framework
  • Google ADK
  • AWS Strands Agents
  • Mastra
  • Pydantic AI
  • LlamaIndex
  • Agno

合作夥伴整合

  • LangGraph
  • CrewAI

社群整合

  • Vercel AI SDK
  • OpenAI Agent SDK（開發中）
  • Cloudflare Agents（開發中）

每個整合都提供完整的文件和範例，開發者可以快速上手。

11. 實際應用場景

11.1 智慧客服系統

// 客服 Agent 支援多輪對話和工具查詢
const客服Agent = new HttpAgent({
  url: "https://support.example.com/agent",
  threadId: sessionId,
});

await客服Agent.runAgent({
  messages: [{ role: "user", content: "我想查詢訂單狀態" }],
  tools: [{ name: "queryOrder", ... }],
});

11.2 程式碼助手

// 程式碼助手 Agent 支援即時編譯和測試
const codeAgent = new LangGraphAgent({
  graphId: "code-assistant",
  deploymentUrl: "https://code.example.com",
});

// 串流顯示編碼過程
codeAgent.subscribe({
  onToolCallStartEvent: (params) => {
    showStatus(`正在執行 ${params.toolCallName}...`);
  },
});

11.3 資料分析助手

// 資料分析 Agent 支援圖表生成
const dataAgent = new MastraAgent({
  url: "https://data.example.com",
});

await dataAgent.runAgent({
  messages: [{ role: "user", content: "分析銷售趨勢" }],
});

12. 與其他協議的比較

12.1 AG-UI vs MCP

面向AG-UIMCP
目的Agent-使用者互動Agent-工具連接
方向雙向Agent → 工具
資料格式事件流請求-回應
狀態管理Snapshot-Delta無

12.2 AG-UI vs A2A

面向AG-UIA2A
目的Agent-使用者互動Agent-Agent 協作
參與者1 個 Agent + 前端多個 Agent
互動模式即時事件串流任務委派
狀態同步完整支援有限

13. 最佳實踐

13.1 Thread ID 管理

Thread ID 是維護對話狀態的關鍵：

// 保持 threadId 一致性
const threadId = sessionStorage.getItem("threadId") || generateUUID();
sessionStorage.setItem("threadId", threadId);

const agent = new HttpAgent({
  url: "https://agent.example.com",
  threadId,
});

13.2 錯誤處理

agent.use(errorHandlingMiddleware);

agent.subscribe({
  onRunErrorEvent: (params) => {
    logError(params.error);
    showUserFriendlyError(params.error.message);
  },
});

13.3 效能優化

// 使用狀態增量更新減少頻寬
agent.use(stateCompressionMiddleware());

// 過濾不需要的事件
agent.use(
  filterEvents({
    excludeTypes: ["STATE_SNAPSHOT"], // 已有的狀態不需要重複
  })
);

14. 低層通訊邏輯

理解 AG-UI 的低層通訊機制，有助於除錯和效能優化。以下從 HTTP 請求到事件處理的完整流程。

14.1 通訊流程總覽

┌─────────────────────────────────────────────────────────────────────────────┐
│                           AG-UI 通訊流程                                     │
└─────────────────────────────────────────────────────────────────────────────┘

  Frontend (FE)                              Backend (BE)
     │                                           │
     │  1. POST /agent/run                       │
     │  ┌─────────────────────────────────────┐  │
     │  │ Content-Type: application/json      │  │
     │  │ { messages: [...], tools: [...] }   │  │
     │  └─────────────────────────────────────┘  │
     │  ───────────────────────────────────────► │
     │                                           │
     │               2. SSE Response             │
     │  ┌─────────────────────────────────────┐  │
     │  │ content-type: text/event-stream     │◄─│
     │  │                                       │  │
     │  │ data: {"type":"RUN_STARTED",...}     │◄─│
     │  │ data: {"type":"ACTIVITY_SNAPSHOT",..}│◄─│
     │  │ data: {"type":"ACTIVITY_DELTA",...}  │◄─│
     │  │ data: {"type":"RUN_FINISHED",...}    │◄─│
     │  └─────────────────────────────────────┘  │
     │  ◄─────────────────────────────────────── │
     │                                           │

14.2 FE 發起請求

// 客戶端發起 HTTP 請求
fetch(url, {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    "Accept": "text/event-stream",  // 關鍵：要求串流回應
  },
  body: JSON.stringify({
    messages: [{ role: "user", content: "分析這筆交易" }],
    tools: [...],
    state: {},
  }),
});

14.3 BE 返回 SSE 串流

伺服器端持續發送事件，保持 HTTP 連接打開：

HTTP/1.1 200 OK
Content-Type: text/event-stream
Transfer-Encoding: chunked
Cache-Control: no-cache

data: {"type":"RUN_STARTED","runId":"run-123","threadId":"user-1"}

data: {"type":"ACTIVITY_SNAPSHOT","messageId":"trading-1","activityType":"TRADING_VIEW","content":{"symbol":"BTC","annotations":[]}}

data: {"type":"ACTIVITY_DELTA","messageId":"trading-1","patch":[{"op":"add","path":"/annotations/-","value":{"type":"buy","price":45000}}]}

data: {"type":"RUN_FINISHED","runId":"run-123"}

SSE 格式規則：

  • 事件之間用雙換行 \n\n 分隔
  • 只有 data: 開頭的行會被處理
  • 多行 data 會自動合併

14.4 FE 解析 SSE（低層實作）

// client/src/transform/sse.ts
export const parseSSEStream = (source$: Observable<HttpEvent>): Observable<any> => {
  const jsonSubject = new Subject<any>();
  const decoder = new TextDecoder("utf-8", { stream: true });
  let buffer = "";

  source$.subscribe({
    next: (event) => {
      if (event.type === HttpEventType.DATA && event.data) {
        // 1. 累積位元組
        const text = decoder.decode(event.data, { stream: true });
        buffer += text;

        // 2. 按 \n\n 分割完整事件
        const events = buffer.split(/\n\n/);
        buffer = events.pop() || "";

        // 3. 解析每個事件
        for (const eventText of events) {
          const json = parseSSEEvent(eventText);
          jsonSubject.next(json);
        }
      }
    },
  });

  function parseSSEEvent(eventText: string) {
    const lines = eventText.split("\n");
    const dataLines: string[] = [];

    for (const line of lines) {
      if (line.startsWith("data:")) {
        dataLines.push(line.slice(5).replace(/^ /, ""));
      }
    }

    return JSON.parse(dataLines.join("\n"));
  }

  return jsonSubject.asObservable();
};

14.5 事件類型驗證

解析後的事件會透過 Zod Schema 驗證確保類型安全：

// client/src/transform/http.ts
transformHttpEventStream(source$).subscribe({
  next: (json) => {
    const event = EventSchemas.parse(json);  // Zod schema validation
    eventSubject.next(event as BaseEvent);
  },
});

// 事件 schema 定義（摘自 core/src/events.ts）
export const ActivitySnapshotEventSchema = BaseEventSchema.extend({
  type: z.literal(EventType.ACTIVITY_SNAPSHOT),
  messageId: z.string(),
  activityType: z.string(),
  content: z.record(z.any()),
  replace: z.boolean().optional().default(true),
});

export const ActivityDeltaEventSchema = BaseEventSchema.extend({
  type: z.literal(EventType.ACTIVITY_DELTA),
  messageId: z.string(),
  activityType: z.string(),
  patch: z.array(z.any()),  // JSON Patch (RFC 6902)
});

14.6 Protocol Buffer 選項（高效能場景）

對於需要更低延遲的場景，AG-UI 支援 Protocol Buffer：

┌─────────────────────────────────────────────────────────────────┐
│  Protocol Buffer Frame Format                                   │
├─────────────────────────────────────────────────────────────────┤
│  [4 bytes: message length] [N bytes: protobuf data]            │
│        ▲ uint32 big-endian              ▲ AG-UI Event          │
└─────────────────────────────────────────────────────────────────┘

BE 發送：
  ┌──────────┬──────────────────┐
  │ 0x0000012C│ (proto data)     │
  │  (200)    │ [encoded event]  │
  └──────────┴──────────────────┘

Protocol Buffer 解析流程：

// client/src/transform/proto.ts
function processBuffer() {
  while (buffer.length >= 4) {
    // 1. 讀取 4 bytes header → messageLength
    const view = new DataView(buffer.buffer, buffer.byteOffset, 4);
    const messageLength = view.getUint32(0, false); // false = big-endian

    // 2. 檢查是否有完整訊息
    const totalLength = 4 + messageLength;
    if (buffer.length < totalLength) break;

    // 3. 提取並解碼
    const message = buffer.slice(4, totalLength);
    const event = proto.decode(message);
    eventSubject.next(event);

    // 4. 移除已處理的資料
    buffer = buffer.slice(totalLength);
  }
}

14.7 事件處理管道

RxJS Observable 串起整個處理流程：

// client/src/agent/agent.ts
public async runAgent(input: RunAgentInput): Promise<RunAgentResult> {
  const pipeline = pipe(
    () => this.run(input),                    // 執行 Agent
    transformChunks(this.debug),              // 轉換區塊事件
    verifyEvents(this.debug),                 // 驗證事件
    (source$) => this.apply(input, source$, subscribers),  // 應用事件
    catchError((error) => this.onError(input, error, subscribers)),
  );

  return await lastValueFrom(pipeline(of(null)));
}

14.8 Activity 事件實際流程

以 TradingView 增量更新為例的完整資料流：

1. FE 發送請求
   POST /agent/run
   Body: { "messages": [...], "tools": [...] }

2. BE 返回 SSE 串流
   data: {"type":"RUN_STARTED","runId":"abc123"}

   data: {"type":"ACTIVITY_SNAPSHOT","messageId":"chart-1",
          "activityType":"TRADING_VIEW",
          "content":{"symbol":"BTC-USD","annotations":[]}}

   data: {"type":"ACTIVITY_DELTA","messageId":"chart-1",
          "patch":[{"op":"add","path":"/annotations/-",
                    "value":{"type":"buy-signal","price":45000}}]}

   data: {"type":"ACTIVITY_DELTA","messageId":"chart-1",
          "patch":[{"op":"add","path":"/annotations/-",
                    "value":{"type":"sell-signal","price":47500}}]}

   data: {"type":"RUN_FINISHED"}

3. FE 解析並更新狀態
   - ACTIVITY_SNAPSHOT → 建立 chart-1 組件
   - ACTIVITY_DELTA → 增量新增 annotations
   - RUN_FINISHED → 完成

4. React 渲染更新
   - useState 更新 messages
   - TradingView 組件 re-render
   - annotations 增量新增到圖表

14.9 關鍵設計要點

層次技術作用
傳輸HTTP + SSE長連接、串流推送
編碼JSON 或 ProtoBuf事件序列化
驗證Zod Schema類型安全
處理RxJS Observable組合式事件處理
狀態Snapshot + Delta高效狀態同步

14.10 除錯技巧

# 查看原始 SSE 串流
curl -N -X POST http://localhost:3000/agent/run \
  -H "Content-Type: application/json" \
  -H "Accept: text/event-stream" \
  -d '{"messages":[{"role":"user","content":"hello"}]}'
// 在 FE 加入除錯 middleware
agent.use({
  run(input: RunAgentInput, next: AbstractAgent): Observable<BaseEvent> {
    return next.run(input).pipe(
      tap((event) => {
        console.log("[DEBUG]", event.type, event);
      }),
    );
  },
});

15. 未來發展方向

根據官方 Roadmap，AG-UI 正在發展以下方向：

  1. 更多 SDK 支援：.NET、Nim、Flowise、Langflow
  2. React Native 客戶端：支援行動應用
  3. AWS Bedrock Agents 整合：雲端 Agent 服務
  4. 效能優化：更高效的事件編碼
  5. 文件完善：更多範例和教程

16. 總結

AG-UI 填補了 AI Agent 生態系統中的一個重要空白：如何讓 Agent 與使用者介面進行流暢、自然的互動。透過事件驅動的架構、傳輸無關的設計、豐富的狀態管理機制，AG-UI 為構建現代 Agentic 應用提供了堅實的基礎。

無論是智慧客服、程式碼助手還是資料分析工具，AG-UI 都提供了一套標準化、可擴展的解決方案。從低層的 SSE 串流解析到高層的 Activity 增量更新，AG-UI 的設計處處體現對效能與開發體驗的考量。

如果你正在構建 Agent 應用，AG-UI 值得你認真考慮。它不僅解決了眼前的技術問題，更為你的應用提供了一個面向未來的架構基礎。

參考資源