LangGraph工作流集成FastAPI实战：构建可生产AI应用

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 FrameworkFlask 太轻缺开箱即用的 Pydantic 集成、异步支持弱、自动生成 OpenAPI 文档能力简陋Django REST Framework 太重ORM 绑定深、启动慢、对纯函数式 workflow 的侵入性强。FastAPI 的胜出点非常具体Pydantic v2 原生深度集成LangGraph 的 state 本质是 TypedDict 或 Pydantic ModelFastAPI 的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(gt0, le5)会自动渲染成 Swagger UI 的交互式文档。产品、测试、前端同学点开就能试不用再维护一份脱离代码的 Postman collection。提示别被 “FastAPI 很快” 的名字误导。它的核心优势不是吞吐量数字而是开发效率和可维护性。在 AI workflow 场景下90% 的瓶颈在 LLM 调用延迟而非框架本身。选 FastAPI是选它减少你写样板代码的时间而不是选它扛住每秒 10 万请求。2.3 架构分层从底层到接口每一层都必须可控我画过不下 20 张架构草图最终稳定下来的四层结构是State 层WorkFlowState类用 Pydantic v2 的BaseModel实现不是 TypedDict好处是自带.model_dump()序列化、.model_validate()反序列化、字段校验。字段命名全部用snake_case和 FastAPI 的 JSON 字段风格一致避免大小写转换 bug。Node 层每个节点是独立函数签名严格为def node_name(state: WorkFlowState) - dict。绝不允许节点之间直接调用如node_b(node_a(state))所有通信必须经 state。节点内只做一件事调用 LLM / 查询 DB / 发送邮件并将结果写入 state 字典。Graph 层用StateGraph(WorkFlowState)构建显式定义add_node()、add_edge()、add_conditional_edges()。特别注意add_conditional_edges()的 condition 函数必须返回字符串节点名不能返回None或布尔值否则 LangGraph 会静默失败。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(ge0.0, le1.0) class WorkFlowState(BaseModel): # 用户原始输入只读 user_input: str Field(..., min_length1, max_length10000) # 流程控制字段必须初始化 current_step: str Field(defaultextract_clauses) retry_count: int Field(default0, ge0, le3) start_time: datetime Field(default_factorydatetime.now) # 各节点产出用 Optional 包裹明确表示“此阶段未执行” extracted_clauses: Optional[List[Clause]] None template_score: Optional[float] Field(defaultNone, ge0.0, le100.0) revision_suggestions: Optional[str] None # 错误追踪用 list 记录每次失败不覆盖 errors: List[Dict[str, Any]] Field(default_factorylist) # 上下文透传字段供调试用 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 提取条款又查数据库比对模板还发邮件。正例拆成三个 nodeextract_clauses、score_against_template、send_notification。好处是单个 node 失败不影响其他可以单独对extract_clauses做压力测试后续加“缓存条款提取结果”功能只改这一个 node。第二幂等性相同输入的 state多次调用 node 必须产生相同输出 state。LLM 调用天然不幂等temperature0.7 时结果会变所以必须显式控制。我在所有 LLM node 里强制加configurable{llm_temperature: 0.0}并通过 FastAPI 的config参数透传。同时node 函数开头加校验if state.extracted_clauses is not None: return state避免重复执行。第三可观测性每个 node 必须记录关键指标且日志格式统一。我用 structlog 替代原生 loggingnode 函数开头固定三行import structlog logger structlog.get_logger() async def extract_clauses(state: WorkFlowState) - dict: logger.info(node_start, nodeextract_clauses, trace_idstate.trace_id, input_lenlen(state.user_input)) start_time time.time() try: # ... LLM 调用逻辑 ... clauses await llm_client.invoke(prompt) logger.info(node_success, nodeextract_clauses, clause_countlen(clauses), duration_ms(time.time()-start_time)*1000) return {extracted_clauses: clauses} except Exception as e: logger.error(node_failure, nodeextract_clauses, errorstr(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_failuretagELK 里用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_templateLangGraph 静默忽略流程卡在当前节点不动日志里没有任何提示。陷阱二condition 函数不能有副作用。它只能读 state不能改 state。所有 state 修改必须在 node 函数里完成。否则会出现 condition 读到旧 statenode 又改了 state逻辑错乱。解决方案是所有 condition 函数必须单元测试。我用 pytest 写了 12 个测试用例覆盖state.extracted_clausesNone、state.retry_count3、state.template_score95.5等所有边界组合。测试代码比 condition 函数本身还长但值得。提示别用lambda写 condition 函数它无法被 pytest 捕获无法打日志无法 debug。必须用具名函数哪怕只是def _route_for_test(state): ...。4. 实操过程与核心环节实现4.1 项目初始化与依赖管理用 Poetry 锁死版本LangGraph 和 FastAPI 生态更新极快上周还能跑的代码这周langgraph0.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.1StateGraph的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(ge0.0, le1.0) class WorkFlowState(BaseModel): user_input: str Field(..., min_length10, max_length5000) extracted_clauses: Optional[List[Clause]] None errors: List[Dict[str, Any]] Field(default_factorylist) 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( modelclaude-3-haiku-20240307, temperature0.0, max_tokens1024, ) 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(titleContract Review API, version1.0.0) # CORS 配置生产环境需细化 origin app.add_middleware( CORSMiddleware, allow_origins[*], allow_credentialsTrue, 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_inputinput_data.user_input, trace_idtrace_id ) try: # 调用 workflow result await workflow_app.ainvoke(initial_state) # 检查错误 if result.errors: raise HTTPException( status_code400, detailfWorkflow 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_code500, detailstr(e))第五步启动服务# 安装依赖 poetry install # 启动需设置 ANTHROPIC_API_KEY 环境变量 export ANTHROPIC_API_KEYyour-key-here poetry run uvicorn src.api.main:app --reload --host 0.0.0.0:8000访问http://localhost:8000/docsSwagger 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_codee.status_code, detaile.detail, pathrequest.url.path) return JSONResponse( status_codee.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, errorstr(e), tracebacktraceback.format_exc(), pathrequest.url.path) return JSONResponse( status_code500, 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), timeouttimeout ) return result except asyncio.TimeoutError: raise HTTPException(status_code408, detailfWorkflow execution timed out after {timeout} seconds) except Exception as e: raise e然后在 API 路由里调用safe_invoke(initial_state, timeout45.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_code400, detailInput contains prohibited content) # ... rest of logic这些补丁加完代码量增加 300 行但生产稳定性提升一个数量级。没有它们你的 workflow 可能被一个超长输入拖垮内存或被一个scriptalert(1)/script注入搞崩。5. 常见问题与排查技巧实录5.1 LangGraph 状态丢失为什么我的 state 字段在下一个 node 里是 None这是最高频问题90% 的原因是state 字段名拼写错误或大小写不一致。LangGraph 的 state merge 是浅合并shallow merge它不会报错只会静默忽略不存在的字段。排查步骤在每个 node 函数开头加日志logger.info(state_keys, keyslist(state.__dict__.keys()))比较前一个 node 的输出字典 key 和当前 node 的 state 字段名。常见错误node A 返回{extracted_clauses: [...]}snake_case但 state 类里定义的是extractedClauses: Optional[List[Clause]]camelCaseLangGraph 找不到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]] None5.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_typetext/event-stream)但stream_generator没有yield或者yield的不是 bytes。没设置正确的 headersCache-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_inputinput_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 fdata: {json.dumps(chunk)}\n\n except Exception as e: yield fevent: error\ndata: {json.dumps({error: str(e)})}\n\n return StreamingResponse( event_generator(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, } )前端接收示例JavaScriptconst 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 编排。终极排查表现象最可能原因快速验证方法解决方案workflow 卡住不动无日志condition 函数返回了未注册的节点名检查route_xxx函数返回值是否在add_node()中注册过在 condition 函数里加logger.info(route_result, resultreturn_value)state.errors为空但流程失败node 函数没返回{errors: [...]}字典在 node 开头加logger.info(node_input_state, statestate.model_dump())确保所有异常分支都返回含errors的 dictAPI 返回 500日志显示ValidationErrorPydantic 模型字段类型不匹配如 str 赋给 int 字段查看日志中的ValidationError详情定位哪个字段在 state 类里加model_config {coerce_numbers_to_str: True}或修正类型本地跑得通线上报ModuleNotFoundErrorPoetry lock 文件未提交或 CI 未运行poetry install登录线上机器poetry show查看已安装包确保 CI 流程包含poetry export -f requirements.txt requirements.txt并在部署时pip install -r requirements.txt最后分享一个小技巧在src/workflow/graph.py里加一个debug_mode开关DEBUG_MODE True # 生产环境设为 False if DEBUG_MODE: # 启用 LangGraph 的 debug 日志 import logging logging.getLogger(langgraph).setLevel(logging.DEBUG)