AI Agent 的流式架构：从输入到输出的全链路数据流设计-开发者社区

AI Agent 的流式架构：从输入到输出的全链路数据流设计

我做第一个 AI Agent 项目时，把所有逻辑都塞在一个函数里——输入→处理→输出，串行执行。后来加了"流式输出"需求，整个架构要大改，改了三天才勉强能跑。那时候才意识到：Agent 的数据流设计应该在第一天就做好，而不是后期硬塞。这篇就把我从"串行杂乱"到"流式清晰"的架构演进过程写出来。

一、为什么 Agent 需要"流式"？——从用户体验到技术实现

AI Agent 的"流式"有两层含义：

用户体验层：用户不想等 10 秒看到完整回答，而是希望每个 token 逐步出现在屏幕上——像 ChatGPT 那样"边思考边输出"。
技术层：Agent 的处理链路很长（输入→意图识别→工具选择→工具执行→结果整合→输出），中间每个环节都可能产生可展示的中间结果。如果等所有环节都完成才输出，用户体验极差。

// ❌ 串行处理：所有步骤完成后才返回结果 async fn serial_agent(user_input: &str) -> String { // 第一步：意图识别（耗时 500ms） let intent = recognize_intent(user_input).await; // 第二步：选择工具（耗时 200ms） let tool = select_tool(&intent).await; // 第三步：执行工具（耗时 2-5秒） let tool_result = execute_tool(&tool).await; // 第四步：整合结果（耗时 300ms） let final_output = synthesize_result(&tool_result).await; // 用户等了 3-6秒才看到结果！ final_output }

sequenceDiagram participant U as 用户 participant A as Agent U->>A: "帮我查 Rust 最新版本" Note over A: 意图识别 500ms Note over A: 工具选择 200ms Note over A: 执行工具 3000ms Note over A: 整合结果 300ms A-->>U: 完整回答（等待4秒） Note over U,U: 用户等了4秒才看到任何输出 ❌ style A fill:#f96

二、流式架构的核心：Channel + Stream + 管道模式

流式架构的关键是：把每个处理环节变成独立的"管道节点"，节点之间用 channel 连接，数据像流水一样从输入端逐步流向输出端。每个节点处理完一部分数据，立刻通过 channel 发给下游，同时也可以发给"展示层"——用户就能看到中间结果。

use tokio::sync::mpsc; use tokio_stream::{StreamExt, wrappers::ReceiverStream}; /// Agent 事件类型：每个处理环节都可能产生可展示的事件 #[derive(Debug, Clone)] enum AgentEvent { // 意图识别完成 IntentRecognized { intent: String, confidence: f32 }, // 工具选择完成 ToolSelected { tool_name: String, reason: String }, // 工具执行中间结果 ToolProgress { step: String, partial_result: String }, // 工具执行完成 ToolCompleted { result: String }, // 最终输出（逐token） OutputToken { token: String }, // 流式输出完成 Done, } /// 流式 Agent 架构：每个环节通过 channel 向下游传递事件 struct StreamingAgent { // 输入端：接收用户消息 input_rx: mpsc::Receiver<String>, // 输出端：发送 Agent 事件给展示层 event_tx: mpsc::Sender<AgentEvent>, } impl StreamingAgent { /// 创建流式 Agent fn new(input_rx: mpsc::Receiver<String>, event_tx: mpsc::Sender<AgentEvent>) -> Self { Self { input_rx, event_tx } } /// 启动流式处理：每个环节独立运行，通过 channel 串联 async fn run(&mut self) { while let Some(user_input) = self.input_rx.recv().await { // 为每个用户输入创建一条完整的处理管道 self.process_stream(user_input).await; } } /// 单条输入的流式处理管道 async fn process_stream(&self, user_input: String) { // 管道1：意图识别 → 工具选择 let (intent_tx, intent_rx) = mpsc::channel(10); // 管道2：工具选择 → 工具执行 let (tool_tx, tool_rx) = mpsc::channel(10); // 管道3：工具执行 → 结果整合 → 输出 let (result_tx, result_rx) = mpsc::channel(10); let event_tx = self.event_tx.clone(); // 每个管道节点独立 spawn，数据流式传递 let intent_stage = tokio::spawn(intent_recognizer(user_input, intent_tx, event_tx.clone())); let tool_stage = tokio::spawn(tool_selector(intent_rx, tool_tx, event_tx.clone())); let execute_stage = tokio::spawn(tool_executor(tool_rx, result_tx, event_tx.clone())); let output_stage = tokio::spawn(output_synthesizer(result_rx, event_tx)); // 等待所有管道完成 let _ = tokio::join!(intent_stage, tool_stage, execute_stage, output_stage); } } /// 管道节点1：意图识别（完成后向下游和展示层发送事件） async fn intent_recognizer( input: String, intent_tx: mpsc::Sender<String>, event_tx: mpsc::Sender<AgentEvent>, ) { // 识别意图 let intent = recognize_intent(&input).await; // 立刻通知展示层：用户可以看到"我理解了你的意图" event_tx.send(AgentEvent::IntentRecognized { intent: intent.clone(), confidence: 0.92, }).await.unwrap(); // 把意图传给下游 intent_tx.send(intent).await.unwrap(); } /// 管道节点2：工具选择 async fn tool_selector( mut intent_rx: mpsc::Receiver<String>, tool_tx: mpsc::Sender<String>, event_tx: mpsc::Sender<AgentEvent>, ) { let intent = intent_rx.recv().await.unwrap(); let tool = select_tool(&intent).await; // 通知展示层：用户可以看到"我选择了XX工具" event_tx.send(AgentEvent::ToolSelected { tool_name: tool.clone(), reason: format!("基于意图 '{}' 选择", intent), }).await.unwrap(); tool_tx.send(tool).await.unwrap(); } /// 管道节点3：工具执行（可能产生多个中间结果） async fn tool_executor( mut tool_rx: mpsc::Receiver<String>, result_tx: mpsc::Sender<String>, event_tx: mpsc::Sender<AgentEvent>, ) { let tool = tool_rx.recv().await.unwrap(); let result = execute_tool_streaming(&tool, &event_tx).await; result_tx.send(result).await.unwrap(); } /// 管道节点4：结果整合 → 流式输出 async fn output_synthesizer( mut result_rx: mpsc::Receiver<String>, event_tx: mpsc::Sender<AgentEvent>, ) { let result = result_rx.recv().await.unwrap(); // 逐 token 输出（模拟 LLM 的流式生成） for token in result.split_whitespace() { event_tx.send(AgentEvent::OutputToken { token: token.to_string(), }).await.unwrap(); tokio::time::sleep(std::time::Duration::from_millis(50)).await; // 模拟生成延迟 } event_tx.send(AgentEvent::Done).await.unwrap(); }

flowchart LR A["用户输入"] --> B["意图识别<br/>500ms"] B --> C["工具选择<br/>200ms"] C --> D["工具执行<br/>2-5s"] D --> E["结果整合<br/>300ms"] B -->|IntentRecognized<br/>立刻展示| S["展示层<br/>用户屏幕"] C -->|ToolSelected<br/>立刻展示| S D -->|ToolProgress<br/>中间结果| S E -->|OutputToken<br/>逐词输出| S style S fill:#6f6

三、流式输出到客户端：Server-Sent Events 的实战对接

前端展示流式结果，最常用的方案是 SSE（Server-Sent Events）。后端把AgentEvent转成 SSE 格式，前端用 EventSource API 接收。

use axum::{ response::sse::{Event, Sse}, routing::get, Router, }; use futures::stream::Stream; use std::convert::Infallible; /// 把 AgentEvent 转成 SSE 格式 fn event_to_sse(event: AgentEvent) -> Event { let data = match event { AgentEvent::IntentRecognized { intent, confidence } => { format!("type=intent&intent={}&confidence={}", intent, confidence) } AgentEvent::ToolSelected { tool_name, reason } => { format!("type=tool&name={}&reason={}", tool_name, reason) } AgentEvent::ToolProgress { step, partial_result } => { format!("type=progress&step={}&result={}", step, partial_result) } AgentEvent::ToolCompleted { result } => { format!("type=completed&result={}", result) } AgentEvent::OutputToken { token } => { format!("type=token&content={}", token) } AgentEvent::Done => { format!("type=done") } }; Event::default().data(data) } /// SSE handler：将 Agent 的流式事件推送到客户端 async fn sse_handler() -> Sse<impl Stream<Item = Result<Event, Infallible>>> { let (event_tx, event_rx) = mpsc::channel(100); // 启动 Agent 流式处理 let (input_tx, input_rx) = mpsc::channel(10); let mut agent = StreamingAgent::new(input_rx, event_tx); tokio::spawn(async move { agent.run().await }); // 发送测试输入 input_tx.send("帮我查 Rust 最新版本".to_string()).await.unwrap(); // 将 event_rx 转为 SSE Stream let stream = ReceiverStream::new(event_rx).map(event_to_sse).map(Ok); Sse::new(stream) } /// 路由配置 fn app() -> Router { Router::new().route("/api/chat/stream", get(sse_handler)) }

前端接收 SSE 的代码（JavaScript，不是 Rust）：

// 前端接收 SSE 流式事件 const eventSource = new EventSource('/api/chat/stream'); eventSource.onmessage = (event) => { const data = new URLSearchParams(event.data); switch (data.get('type')) { case 'intent': // 显示：Agent 识别了你的意图 appendMessage(`思考中：识别意图 "${data.get('intent')}"`); break; case 'tool': // 显示：Agent 选择了工具 appendMessage(`使用工具：${data.get('name')}`); break; case 'progress': // 显示：工具执行进度 appendMessage(`执行中：${data.get('step')}`); break; case 'token': // 显示：逐词输出结果 appendToken(data.get('content')); // 关键：逐词追加！ break; case 'done': // 流式输出完成 eventSource.close(); break; } };

sequenceDiagram participant C as 客户端 participant S as Axum SSE participant A as 流式 Agent C->>S: GET /api/chat/stream S->>A: 启动流式管道 A-->>S: IntentRecognized (500ms后) S-->>C: SSE: type=intent Note over C: "思考中：识别意图" A-->>S: ToolSelected (700ms后) S-->>C: SSE: type=tool Note over C: "使用工具：search_api" A-->>S: OutputToken (逐词) S-->>C: SSE: type=token&content=Rust S-->>C: SSE: type=token&content=1.75 Note over C: 逐词显示 ✅ A-->>S: Done S-->>C: SSE: type=done style C fill:#6f6

四、多工具并行：Agent 同时调用多个工具的流式设计

很多 Agent 场景需要"并行调用多个工具"——比如用户问"Rust 和 Go 的最新版本分别是什么"，Agent 需要同时查两个 API。串行执行太慢（4秒+4秒=8秒），并行执行只需要4秒，但流式展示更复杂。

use tokio::sync::mpsc; use futures::future::join_all; /// 并行工具执行：多个工具同时运行，结果流式汇总 async fn parallel_tool_execution( tools: Vec<String>, event_tx: mpsc::Sender<AgentEvent>, ) -> Vec<String> { // 为每个工具创建独立的执行任务 let tasks: Vec<_> = tools.iter().map(|tool| { let event_tx = event_tx.clone(); tokio::spawn(async move { // 通知展示层：开始执行工具 event_tx.send(AgentEvent::ToolProgress { step: format!("开始执行 {}", tool), partial_result: String::new(), }).await.ok(); // 执行工具（每个工具独立运行） let result = execute_tool(tool).await; // 通知展示层：工具完成 event_tx.send(AgentEvent::ToolCompleted { result: result.clone(), }).await.ok(); result }) }).collect(); // 等待所有工具并行完成 let results: Vec<String> = join_all(tasks) .await .into_iter() .map(|r| r.unwrap()) .collect(); results // 返回所有工具的结果 } /// 使用示例：Agent 同时查询 Rust 和 Go 版本 async fn multi_tool_agent(user_input: &str, event_tx: mpsc::Sender<AgentEvent>) { let intent = recognize_intent(user_input).await; // 根据意图选择多个工具 let tools = vec![ "search_rust_version".to_string(), "search_go_version".to_string(), ]; // 并行执行所有工具 let results = parallel_tool_execution(tools, event_tx.clone()).await; // 整合多个工具的结果 let output = synthesize_multi_results(&results).await; // 流式输出最终答案 for token in output.split_whitespace() { event_tx.send(AgentEvent::OutputToken { token: token.to_string(), }).await.ok(); } event_tx.send(AgentEvent::Done).await.ok(); } async fn synthesize_multi_results(results: &[String]) -> String { format!("查询结果：{}", results.join("；")) }

flowchart TD A["用户输入<br/>查 Rust 和 Go 版本"] --> B["意图识别"] B --> C["选择工具1: search_rust"] B --> D["选择工具2: search_go"] C --> E["并行执行<br/>2秒"] D --> E E --> F["结果1: Rust 1.75"] E --> G["结果2: Go 1.22"] F --> H["整合结果"] G --> H H --> I["流式输出<br/>逐词展示"] C -->|ToolProgress<br/>开始执行| S["展示层"] D -->|ToolProgress<br/>开始执行| S F -->|ToolCompleted<br/>Rust结果| S G -->|ToolCompleted<br/>Go结果| S style E fill:#6cf style S fill:#6f6

我在多工具并行时踩过一个 SSE 顺序的坑：工具 A 先完成、工具 B 后完成，但因为 tokio spawn 的调度顺序，ToolCompleted 事件 B 先到达 SSE handler。前端按到达顺序展示，用户看到"Go结果→Rust结果"，和实际执行顺序相反。后来给每个事件加了 sequence number，前端按 seq 排序。

并行工具调用还需要关注取消传播：如果主请求被用户取消，所有正在进行的子工具调用也应及时中断，否则浪费 token。用一个CancellationToken在所有子任务间传播取消信号，可以在 Ctrl+C 时干净地停掉所有下游 IO。