FastAPI与SSE

浏览器要边跑 Agent 边刷 UI，HTTP 长连接 SSE（Server-Sent Events，服务端推送事件）比轮询合适。FastAPI StreamingResponse 包装 graph.astream，按 stream_mode=updates 推送 JSON 行。

---

1. 定位

维度	内容
角色	HTTP 服务层集成
输入 → 输出	POST /chat → SSE event stream
核心 API	FastAPI、StreamingResponse、astream
依赖 LangChain	可选 messages；本篇 mock 节点

---

2. 图拓扑

节点表

节点名	职责	写
work	模拟一步	msg

边表

源	目标	类型
START	work	固定
work	END	固定

---

3. 请求生命周期

1. 客户端 POST {thread_id, input}
2. FastAPI 构造 configurable.thread_id
3. async for ev in graph.astream(..., stream_mode="updates")
4. 每 event 格式化为 data: {json}\n\n
5. 客户端 EventSource 解析

---

4. 原理

4.1 SSE 格式

data: {"work":{"msg":"done"}}\n\n

4.2 thread_id 来源

请求体或 Header 传入，映射 config["configurable"]["thread_id"]。

4.3 checkpointer

多轮对话 compile 时加 checkpointer，SSE 仅改 transport。

---

5. 最小可运行示例

import asyncio
import json
from typing import TypedDict

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    msg: str


def work(_: State) -> dict:
    return {"msg": "done"}


builder = StateGraph(State)
builder.add_node("work", work)
builder.add_edge(START, "work")
builder.add_edge("work", END)
graph = builder.compile(checkpointer=InMemorySaver())

app = FastAPI()


async def sse_gen(thread_id: str):
    cfg = {"configurable": {"thread_id": thread_id}}
    async for ev in graph.astream({"msg": ""}, cfg, stream_mode="updates"):
        yield f"data: {json.dumps(ev, default=str)}\n\n"


@app.post("/run")
async def run(thread_id: str = "demo"):
    return StreamingResponse(sse_gen(thread_id), media_type="text/event-stream")


# uvicorn module:app --reload

---

6. 执行追踪

SSE 序号	payload	state.msg
1	{"work":{"msg":"done"}}	done

---

重要配置参数

参数	类型 / 默认	作用与影响	参考起点	配置指导
stream_mode="updates"	str	SSE 粒度	每节点	或 values
media_type	text/event-stream	SSE 标准	固定	勿 application/json
thread_id 参数	str	会话	UUID	鉴权绑定 user
checkpointer	—	多轮	InMemorySaver	生产 Postgres
反向代理缓冲	nginx	断流	X-Accel-Buffering: no	生产必配
CORS	FastAPI	浏览器	前端域	限源

---

7. 易踩坑

1. sync invoke 堵死 worker：必须用 astream。
2. nginx 缓冲 SSE：客户端迟迟收不到 event。
3. JSON 序列化 Message 失败：default=str 或自定义 encoder。

---

小结

• FastAPI StreamingResponse + astream = LangGraph SSE 出口。
• thread_id 从 HTTP 层传入 config。
• 生产注意 代理缓冲 与 checkpointer 后端。

---

参考链接

• Streaming
• FastAPI StreamingResponse