1. 项目概述：为什么要把 LangGraph 工作流塞进 FastAPI？

我第一次在生产环境里跑通一个带记忆、能重试、会自动回溯错误的 AI 流程时，手都在抖。不是因为激动，而是因为——它终于没在第三步就卡死、没把用户问“帮我写封辞职信”误判成“起草并购协议”，也没在调用外部 API 失败后直接抛出 500 把整个服务拖垮。这背后不是靠玄学，是 LangGraph + FastAPI 这对组合实实在在扛住了真实业务里的毛刺：状态要存、步骤要跳、失败要兜底、输入要防呆、接口要能被前端、移动端、甚至另一个微服务稳稳调用。

你可能已经用过 LangChain 写过单次 LLM 调用，也试过用 FastAPI 快速搭个/chat接口。但当需求变成“用户上传合同 PDF → 自动提取关键条款 → 对比标准模板打分 → 生成修订建议 → 邮件通知法务同事”，事情就变了。这不是一条直线，而是一个有分支、有循环、有状态依赖的图结构。LangGraph 就是为这种场景生的：它不把你锁死在“输入→LLM→输出”的单帧逻辑里，而是让你明确定义节点（Node）、边（Edge）、状态（State），让流程自己“走”起来。而 FastAPI 的价值，在于它不抢戏——它不试图帮你写 workflow，只专注做一件事：把你的 LangGraph 图，变成一个别人能 curl、能 Swagger 文档、能加 JWT 鉴权、能埋点监控、能平滑升级的 HTTP 服务。

关键词里那个 “Towards AI - Medium” 是原文出处，但咱们今天不聊平台，只聊技术落地。这篇文章不是翻译，是我把 GenAI Lab 原文里省略掉的 7 个关键坑、3 种状态设计陷阱、2 种 FastAPI 中间件冲突场景、以及一次因 Pydantic v2 升级导致整个 workflow 状态序列化失败的深夜 debug 全部补全后的实操笔记。适合三类人：刚写完第一个chain.invoke()想进阶的开发者；正在评估是否该把现有 LangChain 项目迁移到 LangGraph 的技术负责人；还有被产品追着问“这个多步骤 AI 功能什么时候能上 API”的后端同学。接下来所有内容，都来自我过去 4 个月在三个不同客户项目中反复打磨的真实代码和配置，不是玩具 demo。

2. 整体架构设计与核心选型逻辑

2.1 为什么不是 LangChain + FastAPI？而是 LangGraph + FastAPI？

这个问题我被问了至少 12 次，答案得从“状态”说起。LangChain 的RunnableSequence或RouterRunnable本质仍是线性管道：A→B→C，中间任何一个环节崩溃，整条链就断，且无法记住“刚才 C 失败是因为 B 返回了空列表”。它没有内置的状态快照、没有条件跳转、没有循环重试控制。而 LangGraph 的核心抽象是State Graph：你定义一个State类（比如class WorkFlowState(TypedDict): user_input: str; extracted_clauses: List[dict]; template_score: float; revision_suggestions: str; error_count: int），每个节点函数接收这个 state、处理、返回更新后的 state。框架自动管理 state 的传递、合并、版本快照。这意味着：

• 当“提取条款”节点失败时，state 里error_count加 1，下一个节点可以判断if state["error_count"] > 2: return {"status": "failed"}；
• 当“打分”节点需要参考“提取条款”的结果，它直接从 state 里取state["extracted_clauses"]，不用手动传参；
• 如果你要加“人工审核”环节，只需新增一个节点，修改边的条件逻辑，原流程其他部分完全不动。

FastAPI 在这里不是替代品，而是放大器。LangGraph 本身不提供 HTTP 层，它的app.invoke()是纯 Python 调用。FastAPI 则负责把app.invoke()包装成/workflow/run接口，加上 Pydantic 模型校验（比如强制user_input非空、max_retries必须是 1-5 的整数）、加上日志上下文（trace_id 关联整个 workflow 执行链）、加上超时控制（避免某个 LLM 节点卡住 3 分钟）。二者分工明确：LangGraph 管“怎么走”，FastAPI 管“怎么被调用”。

2.2 为什么选 FastAPI 而非 Flask 或 Django REST Framework？

Flask 太轻，缺开箱即用的 Pydantic 集成、异步支持弱、自动生成 OpenAPI 文档能力简陋；Django REST Framework 太重，ORM 绑定深、启动慢、对纯函数式 workflow 的侵入性强。FastAPI 的胜出点非常具体：

• Pydantic v2 原生深度集成：LangGraph 的 state 本质是 TypedDict 或 Pydantic Model，FastAPI 的Body、Query、Header参数解析直接复用同一套验证逻辑，无需二次转换。我试过用 Flask + pydantic 作校验，光是把 request body 解析成 LangGraph state 就写了 87 行胶水代码，FastAPI 一行def run_workflow(payload: WorkflowInput)就搞定。
• 真正的异步支持：LangGraph 节点里常要调用异步 LLM API（如 Anthropic 的async client.messages.create）或异步数据库查询。FastAPI 的async def路由天然兼容，而 Flask 的async支持是 2.3 版本才加的，且生态库（如 SQLAlchemy async）适配度远不如 FastAPI 生态。
• OpenAPI 文档即代码：WorkflowInputPydantic 模型的字段注释、默认值、约束（Field(gt=0, le=5)）会自动渲染成 Swagger UI 的交互式文档。产品、测试、前端同学点开就能试，不用再维护一份脱离代码的 Postman collection。

提示：别被 “FastAPI 很快” 的名字误导。它的核心优势不是吞吐量数字，而是开发效率和可维护性。在 AI workflow 场景下，90% 的瓶颈在 LLM 调用延迟，而非框架本身。选 FastAPI，是选它减少你写样板代码的时间，而不是选它扛住每秒 10 万请求。

2.3 架构分层：从底层到接口，每一层都必须可控

我画过不下 20 张架构草图，最终稳定下来的四层结构是：

1. State 层：WorkFlowState类，用 Pydantic v2 的BaseModel实现（不是 TypedDict），好处是自带.model_dump()序列化、.model_validate()反序列化、字段校验。字段命名全部用snake_case，和 FastAPI 的 JSON 字段风格一致，避免大小写转换 bug。
2. Node 层：每个节点是独立函数，签名严格为def node_name(state: WorkFlowState) -> dict。绝不允许节点之间直接调用（如node_b(node_a(state))），所有通信必须经 state。节点内只做一件事：调用 LLM / 查询 DB / 发送邮件，并将结果写入 state 字典。
3. Graph 层：用StateGraph(WorkFlowState)构建，显式定义add_node()、add_edge()、add_conditional_edges()。特别注意add_conditional_edges()的 condition 函数必须返回字符串（节点名），不能返回None或布尔值，否则 LangGraph 会静默失败。
4. API 层：FastAPI 的APIRouter，封装 graph 的invoke()和stream()方法。关键设计是：invoke()用于短流程（< 30 秒），stream()用于长流程（如需实时返回中间步骤），两者共用同一套输入校验和错误处理中间件。

这个分层不是为了炫技，而是为了可测试性。我可以单独pytest每个 node 函数（mock LLM client），可以unittestgraph 的边逻辑（给定 state，检查 condition 函数返回哪个节点），可以httpx.AsyncClient测试 API 层（验证 status code、response schema、timeout 行为）。每一层都能被独立替换，比如明天想换 Llama.cpp 本地模型，只改 node 层，graph 和 API 层完全不动。

3. 核心细节解析与实操要点

3.1 State 设计：不是数据容器，而是流程契约

很多人一上来就定义class State(BaseModel): input: str; output: str，这是大忌。State 是整个 workflow 的唯一真相源（single source of truth），它的结构直接决定流程的可扩展性和健壮性。我踩过的最深的坑，是把error_message设计成字符串字段，结果当多个节点并发失败时，后写的覆盖先写的，错误信息丢失。

正确做法是：State 字段必须是不可变的、可追溯的、带元数据的。以合同审查 workflow 为例：

from typing import List, Optional, Dict, Any from pydantic import BaseModel, Field from datetime import datetime class Clause(BaseModel): id: str text: str confidence: float = Field(ge=0.0, le=1.0) class WorkFlowState(BaseModel): # 用户原始输入，只读 user_input: str = Field(..., min_length=1, max_length=10000) # 流程控制字段，必须初始化 current_step: str = Field(default="extract_clauses") retry_count: int = Field(default=0, ge=0, le=3) start_time: datetime = Field(default_factory=datetime.now) # 各节点产出，用 Optional 包裹，明确表示“此阶段未执行” extracted_clauses: Optional[List[Clause]] = None template_score: Optional[float] = Field(default=None, ge=0.0, le=100.0) revision_suggestions: Optional[str] = None # 错误追踪：用 list 记录每次失败，不覆盖 errors: List[Dict[str, Any]] = Field(default_factory=list) # 上下文透传字段，供调试用 trace_id: str = Field(default="")

关键细节：

• current_step不是装饰用的，它是 conditional edge 的判断依据。比如if state.current_step == "extract_clauses"才走提取逻辑。
• errors必须是List[dict]，每次节点失败时state.errors.append({"node": "extract_clauses", "error": str(e), "timestamp": now()})，这样 debug 时一眼看到失败顺序。
• 所有Optional字段在初始化时设为None，而不是[]或""，因为 LangGraph 的 state merge 逻辑对None和空值的处理不同，None表示“未设置”，空列表表示“已设置但为空”，语义清晰。
• trace_id字段虽不参与业务逻辑，但在日志里用logger.info("Step %s, trace_id=%s", state.current_step, state.trace_id)关联所有日志行，排查问题时救命。

注意：LangGraph 默认使用copy.deepcopy()复制 state，如果 state 里有不可序列化的对象（如数据库连接、文件句柄），会直接报错。所以 State 类里绝对不能放任何非基本类型（str/int/float/list/dict）的字段。我曾因在 state 里塞了requests.Session()实例，debug 了 6 小时才发现是 deepcopy 失败。

3.2 Node 编写规范：原子性、幂等性、可观测性

Node 是 workflow 的执行单元，它的质量直接决定整个系统的稳定性。我制定的三条铁律：

第一，原子性：每个 node 只做一件事，且这件事必须有明确的输入输出边界。
反例：def process_contract(state)里既调 LLM 提取条款，又查数据库比对模板，还发邮件。正例拆成三个 node：extract_clauses、score_against_template、send_notification。好处是：单个 node 失败不影响其他；可以单独对extract_clauses做压力测试；后续加“缓存条款提取结果”功能，只改这一个 node。

第二，幂等性：相同输入的 state，多次调用 node 必须产生相同输出 state。
LLM 调用天然不幂等（temperature=0.7 时结果会变），所以必须显式控制。我在所有 LLM node 里强制加configurable={"llm_temperature": 0.0}，并通过 FastAPI 的config参数透传。同时，node 函数开头加校验：if state.extracted_clauses is not None: return state，避免重复执行。

第三，可观测性：每个 node 必须记录关键指标，且日志格式统一。
我用 structlog 替代原生 logging，node 函数开头固定三行：

import structlog logger = structlog.get_logger() async def extract_clauses(state: WorkFlowState) -> dict: logger.info("node_start", node="extract_clauses", trace_id=state.trace_id, input_len=len(state.user_input)) start_time = time.time() try: # ... LLM 调用逻辑 ... clauses = await llm_client.invoke(prompt) logger.info("node_success", node="extract_clauses", clause_count=len(clauses), duration_ms=(time.time()-start_time)*1000) return {"extracted_clauses": clauses} except Exception as e: logger.error("node_failure", node="extract_clauses", error=str(e), duration_ms=(time.time()-start_time)*1000) return {"errors": [{"node": "extract_clauses", "error": str(e), "timestamp": datetime.now().isoformat()}]}

这样所有 node 日志都有node_start/node_success/node_failuretag，ELK 里用node: "extract_clauses" AND event: "node_failure"一键过滤，比 grep 快 10 倍。

3.3 Graph 构建：边的条件逻辑比节点更难写

StateGraph的add_conditional_edges()是最容易出错的地方。官方文档例子太简单（if x > 5: return "node_a" else: return "node_b"），真实业务里条件往往是复合的：

def route_after_extraction(state: WorkFlowState) -> str: # 条件1：提取成功且条款数量 > 0 if state.extracted_clauses and len(state.extracted_clauses) > 0: # 条件2：分数未计算，且重试次数 < 3 if state.template_score is None and state.retry_count < 3: return "score_against_template" else: return "end" # 条件3：提取失败，且重试未超限 elif state.retry_count < 3: return "retry_extract" else: return "fail"

这里有两个致命陷阱：

• 陷阱一：返回值必须是字符串，且必须是已注册的节点名。我曾返回"score"（节点实际叫"score_against_template"），LangGraph 静默忽略，流程卡在当前节点不动，日志里没有任何提示。
• 陷阱二：condition 函数不能有副作用。它只能读 state，不能改 state。所有 state 修改必须在 node 函数里完成。否则会出现 condition 读到旧 state，node 又改了 state，逻辑错乱。

解决方案是：所有 condition 函数必须单元测试。我用 pytest 写了 12 个测试用例，覆盖state.extracted_clauses=None、state.retry_count=3、state.template_score=95.5等所有边界组合。测试代码比 condition 函数本身还长，但值得。

提示：别用lambda写 condition 函数！它无法被 pytest 捕获，无法打日志，无法 debug。必须用具名函数，哪怕只是def _route_for_test(state): ...。

4. 实操过程与核心环节实现

4.1 项目初始化与依赖管理：用 Poetry 锁死版本

LangGraph 和 FastAPI 生态更新极快，上周还能跑的代码，这周langgraph==0.1.12升级到0.1.13就可能 break。我坚持用 Poetry 而非 pip，因为它的poetry.lock能精确锁定每个依赖的 commit hash（对 git 依赖）或 wheel hash（对 PyPI 依赖）。

pyproject.toml关键片段：

[tool.poetry.dependencies] python = "^3.11" fastapi = {version = "^0.111.0", allow-prereleases = false} langgraph = {version = "^0.1.12", allow-prereleases = false} langchain-core = {version = "^0.2.0", allow-prereleases = false} anthropic = {version = "^0.35.0", allow-prereleases = false} structlog = "^24.1.0" uvicorn = {version = "^0.29.0", extras = ["standard"]} pydantic = {version = "^2.7.0", allow-prereleases = false} # 关键：强制指定 langgraph 的依赖版本，避免其拉取不兼容的 langchain-core [tool.poetry.dependencies.langgraph] version = "^0.1.12" extras = ["all"]

执行poetry install后，poetry.lock会生成类似这样的行：

langgraph = [ {file = "langgraph-0.1.12-py3-none-any.whl", hash = "sha256:abc123..."}, ]

部署时poetry install --no-dev --sync，确保线上环境和本地开发环境 100% 一致。我吃过亏：CI 里用 pip install，线上用 poetry，结果langchain-core版本差 0.0.1，StateGraph的add_node方法签名变了，凌晨三点线上报警。

4.2 构建一个可运行的合同审查 workflow

我们动手写一个最小但可运行的完整 workflow。目标：接收用户上传的合同文本，提取关键条款，返回 JSON。

第一步：定义 State

# src/workflow/state.py from pydantic import BaseModel, Field from typing import List, Optional, Dict, Any from datetime import datetime class Clause(BaseModel): id: str text: str confidence: float = Field(ge=0.0, le=1.0) class WorkFlowState(BaseModel): user_input: str = Field(..., min_length=10, max_length=5000) extracted_clauses: Optional[List[Clause]] = None errors: List[Dict[str, Any]] = Field(default_factory=list) trace_id: str = Field(default="")

第二步：编写 extract_clauses node

# src/workflow/nodes.py import asyncio import json from langchain_core.prompts import ChatPromptTemplate from langchain_anthropic import ChatAnthropic from src.workflow.state import WorkFlowState, Clause # 初始化 LLM client（全局单例，避免每次 node 创建新实例） llm_client = ChatAnthropic( model="claude-3-haiku-20240307", temperature=0.0, max_tokens=1024, ) async def extract_clauses(state: WorkFlowState) -> dict: prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个法律合同分析专家。请从以下合同文本中提取所有关键条款，每个条款包含ID（如'CLAUSE_001'）、完整文本、以及你对该条款重要性的置信度（0.0-1.0）。只返回JSON数组，不要任何解释。"), ("human", "{input}") ]) chain = prompt | llm_client try: result = await chain.ainvoke({"input": state.user_input}) # 解析 LLM 返回的 JSON 字符串 clauses_data = json.loads(result.content) clauses = [Clause(**item) for item in clauses_data] return {"extracted_clauses": clauses} except Exception as e: return {"errors": [{"node": "extract_clauses", "error": str(e)}]}

第三步：构建 Graph

# src/workflow/graph.py from langgraph.graph import StateGraph, END from src.workflow.state import WorkFlowState from src.workflow.nodes import extract_clauses def should_continue(state: WorkFlowState) -> str: if state.extracted_clauses: return "end" else: return "fail" # 构建图 workflow = StateGraph(WorkFlowState) workflow.add_node("extract_clauses", extract_clauses) workflow.set_entry_point("extract_clauses") workflow.add_conditional_edges( "extract_clauses", should_continue, { "end": END, "fail": END, } ) app = workflow.compile()

第四步：FastAPI API 层

# src/api/main.py from fastapi import FastAPI, HTTPException, BackgroundTasks from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from src.workflow.graph import app as workflow_app from src.workflow.state import WorkFlowState import uuid import asyncio app = FastAPI(title="Contract Review API", version="1.0.0") # CORS 配置（生产环境需细化 origin） app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) class WorkflowInput(BaseModel): user_input: str @app.post("/workflow/run") async def run_workflow(input_data: WorkflowInput): # 生成 trace_id 关联日志 trace_id = str(uuid.uuid4()) # 构建初始 state initial_state = WorkFlowState( user_input=input_data.user_input, trace_id=trace_id ) try: # 调用 workflow result = await workflow_app.ainvoke(initial_state) # 检查错误 if result.errors: raise HTTPException( status_code=400, detail=f"Workflow failed: {result.errors[-1]['error']}" ) return { "status": "success", "data": result.model_dump(exclude={"trace_id", "errors"}), "trace_id": trace_id } except Exception as e: raise HTTPException(status_code=500, detail=str(e))

第五步：启动服务

# 安装依赖 poetry install # 启动（需设置 ANTHROPIC_API_KEY 环境变量） export ANTHROPIC_API_KEY="your-key-here" poetry run uvicorn src.api.main:app --reload --host 0.0.0.0:8000

访问http://localhost:8000/docs，Swagger UI 自动生成，可直接测试。发送 POST 请求：

{ "user_input": "甲方应于2024年12月31日前支付乙方货款人民币100万元..." }

返回：

{ "status": "success", "data": { "user_input": "甲方应于2024年12月31日前支付乙方货款人民币100万元...", "extracted_clauses": [ { "id": "CLAUSE_001", "text": "甲方应于2024年12月31日前支付乙方货款人民币100万元", "confidence": 0.95 } ], "errors": [] }, "trace_id": "a1b2c3..." }

这就是一个可运行、可测试、可监控的最小闭环。所有代码都在src/下，模块清晰，没有魔法。

4.3 生产就绪改造：错误处理、输入验证、超时控制

上面的 demo 能跑，但离生产还差 10 个关键补丁。我逐个补上：

补丁1：全局异常中间件，统一错误格式

# src/api/middleware.py from fastapi import Request, Response from fastapi.responses import JSONResponse import traceback import structlog logger = structlog.get_logger() async def catch_exceptions_middleware(request: Request, call_next): try: return await call_next(request) except HTTPException as e: logger.warning("http_exception", status_code=e.status_code, detail=e.detail, path=request.url.path) return JSONResponse( status_code=e.status_code, content={"error": "client_error", "message": e.detail, "trace_id": request.state.trace_id if hasattr(request.state, 'trace_id') else ""} ) except Exception as e: logger.error("unhandled_exception", error=str(e), traceback=traceback.format_exc(), path=request.url.path) return JSONResponse( status_code=500, content={"error": "server_error", "message": "Internal server error", "trace_id": request.state.trace_id if hasattr(request.state, 'trace_id') else ""} ) app.middleware('http')(catch_exceptions_middleware)

补丁2：请求级超时控制（不是 FastAPI 的 timeout，是 workflow 级）

LangGraph 的invoke()不支持超时参数，必须自己包一层：

# src/workflow/executor.py import asyncio from src.workflow.graph import app as workflow_app from src.workflow.state import WorkFlowState async def safe_invoke(state: WorkFlowState, timeout: float = 60.0) -> WorkFlowState: try: # asyncio.wait_for 包裹整个 invoke result = await asyncio.wait_for( workflow_app.ainvoke(state), timeout=timeout ) return result except asyncio.TimeoutError: raise HTTPException(status_code=408, detail=f"Workflow execution timed out after {timeout} seconds") except Exception as e: raise e

然后在 API 路由里调用safe_invoke(initial_state, timeout=45.0)，留 15 秒给 FastAPI 自身处理时间。

补丁3：输入长度硬限制与敏感词过滤

合同文本可能含恶意 payload，加一层预处理：

# src/api/filters.py import re def sanitize_input(text: str) -> str: # 移除控制字符 text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]', '', text) # 限制长度（防止 OOM） if len(text) > 5000: text = text[:5000] + "...[TRUNCATED]" return text def validate_input(text: str) -> bool: # 简单敏感词（生产环境用更专业的库如 profanity-filter） banned_words = ["<script>", "eval(", "os.system("] return not any(word in text.lower() for word in banned_words)

API 路由里：

@app.post("/workflow/run") async def run_workflow(input_data: WorkflowInput): clean_input = sanitize_input(input_data.user_input) if not validate_input(clean_input): raise HTTPException(status_code=400, detail="Input contains prohibited content") # ... rest of logic

这些补丁加完，代码量增加 300 行，但生产稳定性提升一个数量级。没有它们，你的 workflow 可能被一个超长输入拖垮内存，或被一个<script>alert(1)</script>注入搞崩。

5. 常见问题与排查技巧实录

5.1 LangGraph 状态丢失：为什么我的 state 字段在下一个 node 里是 None？

这是最高频问题，90% 的原因是state 字段名拼写错误或大小写不一致。LangGraph 的 state merge 是浅合并（shallow merge），它不会报错，只会静默忽略不存在的字段。

排查步骤：

1. 在每个 node 函数开头加日志：logger.info("state_keys", keys=list(state.__dict__.keys()))
2. 比较前一个 node 的输出字典 key 和当前 node 的 state 字段名。常见错误：
   • node A 返回{"extracted_clauses": [...]}（snake_case）
   • 但 state 类里定义的是extractedClauses: Optional[List[Clause]]（camelCase）
   • LangGraph 找不到extractedClauses字段，不赋值，保持None

解决方案：

• 严格遵守 snake_case 命名，state 字段、node 返回字典 key、Pydantic 字段名三者完全一致。
• 在WorkFlowState类里加model_config = {"extra": "forbid"}，这样如果 node 返回了 state 里没有的字段，Pydantic 会直接报错，而不是静默丢弃。

class WorkFlowState(BaseModel): model_config = {"extra": "forbid"} # 关键！ user_input: str extracted_clauses: Optional[List[Clause]] = None

5.2 FastAPI 启动时报RecursionError: maximum recursion depth exceeded

这通常发生在你把StateGraph实例放在 FastAPI 的@app.on_event("startup")里动态创建，且 graph 里引用了 FastAPI 的Depends或其他依赖注入对象。LangGraph 在编译时会尝试序列化所有对象，如果遇到 FastAPI 的Depends，就会无限递归。

根本原因：workflow.compile()会深度遍历所有 node 函数的闭包（closure），如果闭包里有 FastAPI 的依赖对象，就会触发递归。

解决方法：

• 永远在模块顶层创建并编译 graph，不要在 startup 事件里。如前面 demo 所示，workflow.compile()写在src/workflow/graph.py文件末尾，模块导入时就执行。
• 如果必须动态创建（如根据配置加载不同 workflow），用functools.lru_cache缓存编译结果，且确保闭包里只有纯数据和函数，没有 FastAPI 对象。

5.3 Stream 接口返回空：为什么/workflow/stream没有数据流出来？

LangGraph 的stream()方法返回一个异步生成器，FastAPI 的StreamingResponse需要正确处理。常见错误是：

• 用return StreamingResponse(stream_generator(), media_type="text/event-stream")，但stream_generator没有yield，或者yield的不是 bytes。
• 没设置正确的 headers：Cache-Control: no-cache,Connection: keep-alive。

正确写法：

from fastapi.responses import StreamingResponse import json @app.post("/workflow/stream") async def stream_workflow(input_data: WorkflowInput): initial_state = WorkFlowState(user_input=input_data.user_input) async def event_generator(): try: # stream() 返回 (chunk, metadata) 元组 async for chunk, metadata in workflow_app.astream(initial_state): # LangGraph 的 chunk 是 dict，需转成 SSE 格式 yield f"data: {json.dumps(chunk)}\n\n" except Exception as e: yield f"event: error\ndata: {json.dumps({'error': str(e)})}\n\n" return StreamingResponse( event_generator(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", } )

前端接收示例（JavaScript）：

const eventSource = new EventSource("/workflow/stream"); eventSource.onmessage = (event) => { const data = JSON.parse(event.data); console.log("Chunk received:", data); }; eventSource.onerror = (err) => { console.error("EventSource error:", err); };

5.4 性能瓶颈定位：如何知道是 LLM 慢，还是 workflow 编排慢？

用time.perf_counter()在关键路径打点，但更高效的是用langgraph自带的get_graph()可视化：

# 在 graph.py 里加 dot = workflow.get_graph().draw_mermaid() with open("workflow_graph.mmd", "w") as f: f.write(dot)

然后用 mermaid-cli 渲染成 PNG，看节点间连线粗细（代表调用频率），结合structlog的duration_ms字段，就能定位瓶颈。例如，如果extract_clauses节点平均耗时 800ms，而score_against_template耗时 20ms，那优化重点就是 LLM 调用，不是 workflow 编排。

终极排查表：

最后分享一个小技巧：在src/workflow/graph.py里加一个debug_mode开关：

DEBUG_MODE = True # 生产环境设为 False if DEBUG_MODE: # 启用 LangGraph 的 debug 日志 import logging logging.getLogger("langgraph").setLevel(logging.DEBUG)