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::ReceiverString, // 输出端发送 Agent 事件给展示层 event_tx: mpsc::SenderAgentEvent, } impl StreamingAgent { /// 创建流式 Agent fn new(input_rx: mpsc::ReceiverString, event_tx: mpsc::SenderAgentEvent) - 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::SenderString, event_tx: mpsc::SenderAgentEvent, ) { // 识别意图 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::ReceiverString, tool_tx: mpsc::SenderString, event_tx: mpsc::SenderAgentEvent, ) { 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::ReceiverString, result_tx: mpsc::SenderString, event_tx: mpsc::SenderAgentEvent, ) { 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::ReceiverString, event_tx: mpsc::SenderAgentEvent, ) { 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 --|IntentRecognizedbr/立刻展示| S[展示层br/用户屏幕] C --|ToolSelectedbr/立刻展示| S D --|ToolProgressbr/中间结果| S E --|OutputTokenbr/逐词输出| S style S fill:#6f6三、流式输出到客户端Server-Sent Events 的实战对接前端展示流式结果最常用的方案是 SSEServer-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!(typeintentintent{}confidence{}, intent, confidence) } AgentEvent::ToolSelected { tool_name, reason } { format!(typetoolname{}reason{}, tool_name, reason) } AgentEvent::ToolProgress { step, partial_result } { format!(typeprogressstep{}result{}, step, partial_result) } AgentEvent::ToolCompleted { result } { format!(typecompletedresult{}, result) } AgentEvent::OutputToken { token } { format!(typetokencontent{}, token) } AgentEvent::Done { format!(typedone) } }; Event::default().data(data) } /// SSE handler将 Agent 的流式事件推送到客户端 async fn sse_handler() - Sseimpl StreamItem ResultEvent, 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: typeintent Note over C: 思考中识别意图 A--S: ToolSelected (700ms后) S--C: SSE: typetool Note over C: 使用工具search_api A--S: OutputToken (逐词) S--C: SSE: typetokencontentRust S--C: SSE: typetokencontent1.75 Note over C: 逐词显示 ✅ A--S: Done S--C: SSE: typedone 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: VecString, event_tx: mpsc::SenderAgentEvent, ) - VecString { // 为每个工具创建独立的执行任务 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: VecString 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::SenderAgentEvent) { 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 --|ToolProgressbr/开始执行| S[展示层] D --|ToolProgressbr/开始执行| S F --|ToolCompletedbr/Rust结果| S G --|ToolCompletedbr/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在所有子任务间传播取消信号可以在 CtrlC 时干净地停掉所有下游 IO。五、总结Agent 的流式架构核心是管道模式每个处理环节是独立的管道节点节点之间用 channel 连接数据逐步流动中间结果随时展示。具体做法AgentEvent 枚举定义所有可能的中间事件类型每个管道节点处理完就发送事件。channel tokio::spawn每个节点独立运行用 channel 串联。channel 的容量就是背压控制点。SSE 对接把 AgentEvent 转成 SSE 格式推送到前端前端用 EventSource 接收并逐步展示。并行工具执行多个工具用join_all并行运行结果流式汇总比串行快 N 倍。作为的 Rust 学习者流式架构让我真正理解了异步编程的价值——不是为了快而是为了流。串行架构追求的是最终结果快流式架构追求的是中间过程可见。对于 Agent 这种多步骤场景过程可见比结果快更重要——用户等 4 秒看到完整回答和逐步看到思考过程体验完全不同。我的教训是一开始就设计流式架构不要后期硬加。串行→流式的改造比一开始就流式的设计成本高很多——因为 channel 和事件类型要渗透到每个环节改起来是全局性的。先画管道图Mermaid再写代码效率会高很多。