Agent-10-05.FastAPI与Agent流式服务

系列：00 索引 · 上一篇：04 配置驱动 · 下一篇：06 混合 RAG

---

1. 行业常见问题

现象	痛点
CLI 只能单人本机用	团队无法共享 Agent
同步 HTTP 等 30s+ 超时	网关/浏览器断连
前端自己拼 LLM 密钥	密钥泄露
业务逻辑写在路由里	难测、难换 UI

---

2. 该技术如何解决

后端：FastAPI（或同类）提供 Agent 编排入口，密钥留在服务端。
流式：SSE（Server-Sent Events）或 WebSocket 推送 token/tool 事件，改善体验。
契约：GET /agents、POST /chat 等与 UI 解耦。

---

3. 核心原理

3.1 推荐 API 形态

端点	作用
GET /agents	列出可对话角色
POST /agents/{id}/chat	发送消息，SSE 返回 token
POST /runs/{id}/approve	HITL（见 Agent-10-08）

3.2 SSE 事件类型（示例）

{"type": "token", "text": "部分输出"}
{"type": "tool_start", "name": "search_pubmed"}
{"type": "tool_end", "name": "search_pubmed", "ok": true}
{"type": "done"}

---

4. 典型实现与代码示例

Web 层在 agent/uni/api/main.py（单 Agent SSE）与 agent/orchestrator/streaming.py（LangGraph 流水线 SSE）实现；底层仍走 AgentRuntime.chat(on_event=...) 或 graph.stream(stream_mode="updates")。

4.1 单 Agent 聊天 SSE

# agent/uni/api/main.py（节选）
@app.post("/agents/{agent_id}/chat")
async def chat_agent(agent_id: str, body: ChatRequest):
    runtime = discovery.load_runtime(agent_id)

    async def event_gen():
        queue: asyncio.Queue = asyncio.Queue()

        def on_event(ev: dict):
            queue.put_nowait(ev)

        task = asyncio.create_task(asyncio.to_thread(runtime.chat, body.message, on_event=on_event))
        while True:
            ev = await queue.get()
            yield sse_line(ev)
            if ev.get("type") == "done":
                break
        await task

    return StreamingResponse(event_gen(), media_type="text/event-stream", ...)

tool_start / tool_end / token 由 run_tool_loop 的 on_event 推送。

4.2 LangGraph 编排 SSE

# agent/orchestrator/streaming.py — stream_pipeline_updates()
for chunk in graph.stream(state, config, stream_mode="updates"):
    yield sse({"type": "node_update", "node": ..., "payload": chunk})
# HITL 中断后：Command(resume=True) 续跑

4.3 工程验收

# 启动 Web（单 Agent + Pipeline 页）
uv run python -m agent.uni.api.main
# 浏览器：/ 单 Agent；/pipeline 多节点编排

# CLI 等价验收（无浏览器）
uv run python -m agent.orchestrator.cli run --symbol BRCA1 --mock-llm

---

5. 替代方案与优缺点

前端/后端组合	优点	缺点
FastAPI + SSE + HTMX	同仓、学习成本低	复杂交互弱
FastAPI + Vue/React SPA	体验好、组件化	构建链、CORS
WebSocket 双向	适合打断、协作编辑	运维与重连复杂
gRPC streaming	高性能内部服务	浏览器支持需网关
仅 stdio MCP	本地 IDE 最佳	非 Web 产品

---

6. 自检题

1. 为什么 Agent 服务通常不把 API Key 下发给浏览器？
2. SSE 与 WebSocket 在本场景各适合什么交互？
3. X-Accel-Buffering: no 解决什么问题？

---

7. 延伸阅读

• FastAPI StreamingResponse
• 下一篇：Agent-10-06 混合检索 RAG