Python+LangGraph构建可审计业务Agent工作流

1. 项目概述这不是写个脚本而是让业务逻辑自己跑起来“Python Agent SDK把业务场景转化为自动化工作流的完整流程”——这句话里藏着一个被很多人忽略的关键动词“转化”。不是“用Python写个自动化脚本”也不是“调用几个API凑成一个链路”而是把真实业务中那些需要人反复判断、跨系统切换、查文档翻记录、在Excel里拖拽公式、再发邮件确认的“活儿”原样搬进代码里让它具备感知、决策、执行、纠错、重试、归档的完整闭环能力。我做过银行对公信贷审批辅助系统也搭过电商售后自动定责引擎还重构过制造业设备报修工单分派流程所有这些项目的起点都不是“我们要做个AI应用”而是业务方拍着桌子说“这个环节每天要花3个人2小时错一次就要赔5000块你们能不能让它别出错”核心关键词Python在这里不是语法糖或胶水语言而是整个工作流的底盘——它提供足够灵活的类型系统支撑多源数据融合比如把ERP里的JSON、CRM里的CSV、邮件正文里的表格片段统一喂给Agent有成熟的异步IO生态应对高并发任务调度比如同时处理200个客户投诉工单更重要的是它的包管理机制pipvenv和模块化设计让不同业务线的Agent可以像乐高一样插拔组合。而Agent SDK特指LangGraph这类面向状态机建模的框架解决的是传统自动化最头疼的问题流程不是线性的。现实业务里“审批通过→发合同→寄快递”这种直线流程少之又少更多是“客户补充材料→转法务复核→若超3天未响应→自动触发催办→若仍无反馈→升级至主管”。LangGraph用图结构Graph显式定义节点Node和边Edge把“条件分支”“循环重试”“人工介入点”全部变成可版本控制、可单元测试、可监控告警的代码实体。这个流程真正落地时适合三类人第一类是业务分析师他们不需要会写Python但能用YAML描述清楚“当订单金额5万且客户等级为VIP时跳过初审直接进入风控模型打分”第二类是后端工程师他们负责把业务规则编译成LangGraph的StateSchema并对接内部API网关第三类是运维同学他们关注的是如何把Agent工作流打包成Docker镜像部署到K8s集群里设置CPU内存限制配置Prometheus指标采集。它不是给程序员炫技的玩具而是让业务逻辑第一次真正拥有了“可执行、可审计、可迭代”的数字生命体。我见过最典型的失败案例就是团队用Flask搭了个Web界面前端点按钮触发Python脚本调用OpenAI API结果业务方提了17个新需求开发改了43次代码最后连谁在哪天改了哪行都查不清——因为“流程”根本没在代码里只存在于产品经理的脑回路和飞书文档的评论区里。2. 核心思路拆解为什么必须放弃“函数链式调用”转向状态机驱动很多刚接触Agent开发的人第一反应是用LangChain的SequentialChain或者RouterChain拼接几个LLM调用。我试过也帮客户重构过——这种模式在POC阶段很香5分钟就能让AI读完合同PDF提取甲方名称、签约日期、违约金条款再生成摘要。但一旦进入真实业务立刻崩盘。原因很简单链式调用把“流程”藏在了函数调用栈里而业务流程的本质是状态迁移。举个具体例子某跨境电商的退货审核流程。理想链式流程可能是OCR识别运单号→查物流状态→查订单是否超7天→查商品是否支持无理由退货→生成审核结论。但现实是OCR识别失败模糊照片、物流接口超时第三方服务抖动、订单系统返回“数据异常”上游数据库主从延迟、客户申诉称“我昨天才下单”时间戳解析错误。这时候链式调用要么整个失败重来浪费算力要么硬编码一堆try-except代码迅速变成意大利面条更糟的是没人知道当前卡在哪个环节、失败原因是什么、重试几次了。LangGraph的破局点在于强制你把“状态”显式定义出来。我们先定义一个OrderReviewStatefrom typing import TypedDict, List, Optional, Dict, Any from langgraph.graph import StateGraph, END class OrderReviewState(TypedDict): order_id: str raw_image_data: bytes # 原始运单照片 ocr_result: Optional[Dict[str, str]] # OCR识别结果 logistics_status: Optional[str] # 物流状态 order_days: Optional[int] # 订单创建天数 return_eligible: Optional[bool] # 是否符合退货条件 audit_conclusion: Optional[str] # 最终结论 error_count: int # 当前错误次数 retry_history: List[Dict[str, Any]] # 重试历史记录看到没error_count和retry_history不是附加功能而是状态的一部分。每个节点Node函数接收整个OrderReviewState只修改它关心的字段然后返回更新后的state。比如OCR节点def ocr_node(state: OrderReviewState) - OrderReviewState: try: # 调用OCR服务 result call_ocr_api(state[raw_image_data]) return {**state, ocr_result: result, error_count: 0} except Exception as e: # 错误不抛出只记录状态 new_history state[retry_history] [{ step: ocr, error: str(e), timestamp: datetime.now().isoformat() }] return { **state, error_count: state[error_count] 1, retry_history: new_history }关键来了边Edge的判定逻辑不再写在节点里而是独立定义。比如OCR失败后是否重试重试几次要不要降级到人工这些全在图的连接逻辑里def should_retry_ocr(state: OrderReviewState) - str: if state[error_count] 3: return ocr_node # 重试OCR elif state[error_count] 3: return manual_review_node # 转人工 else: return fail_node # 终止流程 # 构建图 workflow StateGraph(OrderReviewState) workflow.add_node(ocr_node, ocr_node) workflow.add_node(logistics_check_node, logistics_check_node) workflow.add_node(manual_review_node, manual_review_node) workflow.add_node(fail_node, fail_node) workflow.set_entry_point(ocr_node) workflow.add_conditional_edges( ocr_node, should_retry_ocr, { ocr_node: ocr_node, # 循环重试 manual_review_node: manual_review_node, fail_node: fail_node } )这种设计带来三个硬性好处第一可观测性——你可以随时dump出任意时刻的state看到error_count2、retry_history里有两次超时记录立刻定位问题第二可测试性——写单元测试时直接构造一个state输入ocr_node断言输出的error_count是否正确完全不依赖外部API第三可演进性——业务方说“OCR失败3次后先发短信提醒客户拍照清晰点再转人工”你只需要新增一个sms_notify_node改一行should_retry_ocr的返回逻辑不用动OCR节点本身。这就像给业务流程装上了黑匣子和可编程控制器而不是靠人肉盯着日志猜故障。3. 核心细节解析State Schema设计、工具调用封装与错误熔断策略3.1 State Schema别把它当成DTO它是业务流程的宪法很多团队把State简单理解为“传参容器”结果越做越乱。我见过最离谱的案例一个金融风控Agent的State里塞了47个字段其中23个是临时变量temp_flag_1,intermediate_result_2还有5个字段命名是英文缩写加拼音cust_info_zh导致新成员入职两周都看不懂流程图。正确的做法是State Schema必须和业务领域模型严格对齐且遵循单一职责原则。回到退货审核的例子OrderReviewState的每个字段都应该能在业务文档里找到明确出处order_id: 来自ERP系统订单主键不可为空类型为strraw_image_data: 客户上传的原始图片二进制流大小限制5MB由前端校验ocr_result: OCR服务返回的结构化数据包含tracking_number运单号、ship_date发货日期等类型为Optional[Dict]logistics_status: 物流公司API返回的状态码如DELIVERED,IN_TRANSIT不是字符串描述避免歧义audit_conclusion: 最终结论枚举值{APPROVED, REJECTED, PENDING_MANUAL}禁止自由文本特别注意error_count和retry_history的设计。error_count不是简单的计数器它必须和具体的失败节点绑定。如果logistics_check_node也失败不能共用同一个error_count否则OCR失败3次后转人工物流查询失败1次就直接终止——这显然不合理。所以更严谨的State应该是class OrderReviewState(TypedDict): # ... 其他字段 ocr_error_count: int # 仅OCR错误计数 logistics_error_count: int # 仅物流错误计数 # 或者更进一步用字典 # error_counts: Dict[str, int] # key为节点名value为错误次数retry_history也不能存原始Exception对象序列化失败而应该标准化为{ step: ocr_node, # 节点名 attempt: 1, # 第几次尝试 error_type: TIMEOUT, # 预定义错误类型 error_message: OCR service timeout after 30s, timestamp: 2024-06-15T14:22:33.123Z }这样做的好处是后续可以基于error_type做聚合分析比如发现80%失败是TIMEOUT说明要扩容OCR服务也可以基于step做定向告警logistics_check_node错误率突增立刻通知运维。3.2 工具调用封装让LLM只做决策不碰脏活Agent SDK的核心价值之一是把“调用工具”这件事从LLM提示词里剥离出来。很多新手直接在system prompt里写“你是一个退货审核助手可以调用OCR工具识别运单号调用物流API查询状态……”结果LLM经常胡乱调用工具或者参数填错比如把order_id传给OCR接口。LangGraph的解法是工具调用必须由专门的Tool Node完成且输入输出严格契约化。我们定义一个标准工具调用协议from pydantic import BaseModel, Field from typing import Any, Dict, Optional class ToolCall(BaseModel): tool_name: str # 工具名必须注册在tool_registry中 arguments: Dict[str, Any] # 参数必须符合tool_schema timeout: int 30 # 超时秒数强制约束 class ToolResult(BaseModel): success: bool data: Optional[Any] None error: Optional[str] None metadata: Dict[str, Any] Field(default_factorydict)然后封装OCR工具class OcrTool: def __init__(self, api_url: str, api_key: str): self.api_url api_url self.api_key api_key def schema(self) - Dict[str, Any]: # 返回JSON Schema供LLM理解参数要求 return { type: object, properties: { image_base64: {type: string, description: 图片base64编码}, language: {type: string, enum: [zh, en], default: zh} }, required: [image_base64] } def invoke(self, image_base64: str, language: str zh) - ToolResult: try: response requests.post( f{self.api_url}/ocr, json{image: image_base64, lang: language}, headers{Authorization: fBearer {self.api_key}}, timeoutself.timeout ) if response.status_code 200: return ToolResult(successTrue, dataresponse.json()) else: return ToolResult(successFalse, errorfHTTP {response.status_code}) except requests.Timeout: return ToolResult(successFalse, errorTIMEOUT) except Exception as e: return ToolResult(successFalse, errorstr(e)) # 注册到全局工具库 tool_registry { ocr_tool: OcrTool(https://api.ocr.example.com, sk-xxx) }关键点在于LLM的提示词里永远不出现具体的API地址、参数名、错误码。它只看到你可用的工具 - ocr_tool: 识别图片中的文字输入需包含image_base64字段 - logistics_api: 查询物流状态输入需包含tracking_number字段而实际调用由tool_node完成def tool_node(state: OrderReviewState) - OrderReviewState: tool_call state.get(pending_tool_call) # LLM输出的工具调用请求 if not tool_call or tool_name not in tool_call: return {**state, tool_result: None} tool tool_registry.get(tool_call[tool_name]) if not tool: return {**state, tool_result: ToolResult(successFalse, errorUnknown tool)} try: result tool.invoke(**tool_call[arguments]) return {**state, tool_result: result} except Exception as e: return {**state, tool_result: ToolResult(successFalse, errorstr(e))}这样LLM彻底变成了“大脑”只负责思考“下一步该调用什么工具”而“怎么调用”“超时怎么办”“错误怎么解析”全部交给工程代码。我实测过同样的OCR任务在封装工具前后LLM调用错误率从37%降到1.2%因为LLM再也不用记image_base64这个字段名了——它只要说“调用ocr_tool”工具Node自动从state里提取raw_image_data并转成base64。3.3 错误熔断与降级业务连续性的最后一道保险再健壮的系统也会失败。真正的生产级Agent必须有熔断Circuit Breaker和降级Fallback机制。LangGraph本身不提供熔断但我们可以用tenacity库无缝集成from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type class RobustOcrTool(OcrTool): retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max10), retryretry_if_exception_type((requests.Timeout, requests.ConnectionError)) ) def invoke(self, image_base64: str, language: str zh) - ToolResult: # 原有逻辑不变但自动重试 return super().invoke(image_base64, language)但这只是技术层重试。业务层熔断更关键。比如物流API连续5分钟超时率90%就应该主动熔断不再调用直接走降级逻辑查本地缓存、返回“物流信息暂不可用”。我们在logistics_check_node里加入熔断器from circuitbreaker import CircuitBreaker, CircuitBreakerError logistics_breaker CircuitBreaker( failure_threshold5, # 连续5次失败触发熔断 recovery_timeout60, # 熔断60秒后尝试恢复 expected_exceptionrequests.RequestException ) def logistics_check_node(state: OrderReviewState) - OrderReviewState: try: with logistics_breaker: # 调用物流API result call_logistics_api(state[ocr_result][tracking_number]) return {**state, logistics_status: result[status]} except CircuitBreakerError: # 熔断状态走降级 return {**state, logistics_status: CIRCUIT_BREAKER_OPEN} except Exception as e: # 其他错误计入熔断器 raise e降级策略必须提前设计好。比如物流不可用时我们可能查订单创建时间若24小时返回“物流信息尚未同步请稍后再试”若24小时查历史相似订单的平均配送时长预估状态最终所有降级路径都要记录到state[fallback_used] True方便后续审计。提示熔断阈值不能拍脑袋定。我建议上线前做混沌工程测试用chaospy随机注入网络延迟、服务宕机观察熔断器触发时机是否合理。曾有个客户把failure_threshold设为1结果物流API偶发超时概率0.1%导致整个退货流程天天熔断——这比不熔断还糟。4. 实操全流程从零搭建一个可交付的退货审核Agent4.1 环境准备与依赖管理为什么必须用Poetry而不是pip很多教程教用pip install langgraph langchain-openai这在本地开发没问题但一上生产就踩坑。我吃过亏某次紧急上线运维用pip install -r requirements.txt部署结果langchain-core版本冲突StateGraph的add_conditional_edges方法签名变了整个流程跑不通。根源在于pip不锁子依赖版本。解决方案是Poetry——它生成的poetry.lock文件精确锁定每一层依赖的哈希值。初始化项目# 创建虚拟环境并激活 poetry init -n poetry env use 3.11 poetry add langgraph langchain-openai openai python-dotenv tenacity circuitbreaker poetry add --group dev pytest black isort mypypyproject.toml关键配置[tool.poetry.dependencies] python ^3.11 langgraph ^0.1.0 langchain-openai ^0.1.0 openai ^1.13.0 tenacity ^8.2.3 circuitbreaker ^1.2.6 # 强制指定子依赖版本避免冲突 [tool.poetry.dependencies.langchain-core] version ^0.1.0 allows-prereleases true [build-system] requires [poetry-core] build-backend poetry.core.masonry.api为什么不用requirements.txt因为pip freeze requirements.txt会把所有依赖包括dev依赖都写进去而Poetry的poetry export -f requirements.txt --without-hashes requirements.txt只导出生产依赖且--without-hashes保证兼容性。更重要的是Poetry的poetry install会自动创建隔离环境poetry shell进入后which python指向的就是项目专属Python解释器彻底告别/usr/bin/python和~/anaconda3/bin/python混用的灾难。4.2 State Schema与节点实现手把手写第一个可运行节点我们从最简单的OCR节点开始。创建src/agents/return_review/nodes.pyimport base64 import logging from typing import Dict, Any, Optional from pydantic import BaseModel from langgraph.graph import StateGraph from src.agents.return_review.state import OrderReviewState from src.tools.ocr import OcrTool logger logging.getLogger(__name__) class OcrInput(BaseModel): image_data: bytes language: str zh def ocr_node(state: OrderReviewState) - OrderReviewState: OCR节点识别运单图片提取关键信息 输入state.raw_image_data 输出state.ocr_result, state.ocr_error_count if not state.get(raw_image_data): logger.warning(OCR node skipped: no raw_image_data in state) return {**state, ocr_error_count: state.get(ocr_error_count, 0)} try: # 将bytes转base64 image_b64 base64.b64encode(state[raw_image_data]).decode(utf-8) # 调用OCR工具 ocr_tool OcrTool( api_urlhttps://api.ocr.example.com, api_keysk-xxx ) result ocr_tool.invoke(image_b64, languagezh) if result.success: logger.info(fOCR success for order {state[order_id]}) return { **state, ocr_result: result.data, ocr_error_count: 0 } else: logger.error(fOCR failed for order {state[order_id]}: {result.error}) return { **state, ocr_error_count: state.get(ocr_error_count, 0) 1, retry_history: state.get(retry_history, []) [{ step: ocr_node, error: result.error, timestamp: 2024-06-15T14:22:33.123Z }] } except Exception as e: logger.exception(fUnexpected error in ocr_node for order {state[order_id]}) return { **state, ocr_error_count: state.get(ocr_error_count, 0) 1, retry_history: state.get(retry_history, []) [{ step: ocr_node, error: fUNEXPECTED: {str(e)}, timestamp: 2024-06-15T14:22:33.123Z }] }注意几个实操细节logger使用模块名__name__便于日志按模块过滤所有异常都捕获并转化为结构化错误记录绝不让Exception穿透到Graph层retry_history里的时间戳用固定字符串上线后替换为datetime.now().isoformat()方便单元测试断言4.3 图构建与编译如何让StateGraph真正“活”起来创建src/agents/return_review/workflow.pyfrom langgraph.graph import StateGraph, END from src.agents.return_review.state import OrderReviewState from src.agents.return_review.nodes import ocr_node, logistics_check_node, manual_review_node, fail_node from src.agents.return_review.edges import should_retry_ocr, should_proceed_to_logistics def build_return_review_workflow() - StateGraph: 构建退货审核工作流图 返回可执行的CompiledGraph对象 workflow StateGraph(OrderReviewState) # 添加节点 workflow.add_node(ocr_node, ocr_node) workflow.add_node(logistics_check_node, logistics_check_node) workflow.add_node(manual_review_node, manual_review_node) workflow.add_node(fail_node, fail_node) # 设置入口点 workflow.set_entry_point(ocr_node) # 添加条件边 workflow.add_conditional_edges( ocr_node, should_retry_ocr, { retry: ocr_node, manual: manual_review_node, fail: fail_node } ) workflow.add_conditional_edges( logistics_check_node, should_proceed_to_logistics, { approve: final_approve_node, reject: final_reject_node, manual: manual_review_node } ) # 添加普通边无条件 workflow.add_edge(final_approve_node, END) workflow.add_edge(final_reject_node, END) workflow.add_edge(manual_review_node, END) # 编译图 compiled workflow.compile() # 添加中间件记录每个节点执行耗时 from langgraph.checkpoint.memory import MemorySaver compiled compiled.with_config( checkpointerMemorySaver(), callbacks[{ on_chain_start: lambda x: logger.info(fStart {x.name}), on_chain_end: lambda x: logger.info(fEnd {x.name}, duration: {x.end_time - x.start_time}) }] ) return compiled # 导出可调用的工作流实例 return_review_workflow build_return_review_workflow()关键点compiled workflow.compile()才是真正的魔法。它把节点、边、状态转换规则编译成一个可执行的CompiledGraph对象。这个对象可以同步调用result return_review_workflow.invoke({order_id: ORD123, raw_image_data: b...} )异步调用result await return_review_workflow.ainvoke(...)流式调用async for chunk in return_review_workflow.astream(...): print(chunk)检查点恢复return_review_workflow.get_state(config)获取当前state快照注意MemorySaver是内存检查点仅用于开发测试。生产环境必须用PostgresSaver或RedisSaver否则重启后所有进行中的流程就丢了。我见过客户用MemorySaver上线结果服务器半夜重启200个待审核订单全部消失——这比流程失败还致命。4.4 部署与监控如何让Agent在K8s里稳定跑一年本地跑通只是第一步。生产部署必须解决三件事资源隔离、流量控制、可观测性。资源隔离用Dockerfile精准控制环境FROM python:3.11-slim # 复制poetry.lock和pyproject.toml利用layer cache加速构建 COPY pyproject.toml poetry.lock ./ RUN pip install poetry \ poetry config virtualenvs.create false \ poetry install --no-dev --no-root # 复制源码 COPY src/ /app/src/ WORKDIR /app # 设置启动命令 CMD [poetry, run, python, -m, src.main]流量控制用FastAPI做API网关限流防刷from fastapi import FastAPI, HTTPException, Depends from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded from src.agents.return_review.workflow import return_review_workflow limiter Limiter(key_funcget_remote_address) app FastAPI() app.state.limiter limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) app.post(/api/v1/review) limiter.limit(100/minute) # 每分钟最多100次 async def review_order( request: ReviewRequest, background_tasks: BackgroundTasks ): try: # 构造state state { order_id: request.order_id, raw_image_data: request.image_data.encode(utf-8) # 前端传base64 } # 异步执行工作流 result await return_review_workflow.ainvoke(state) return {status: success, result: result} except Exception as e: logger.error(fWorkflow execution failed: {e}) raise HTTPException(status_code500, detailInternal server error)可观测性集成Prometheus指标from prometheus_client import Counter, Histogram, Gauge import time # 定义指标 WORKFLOW_START_COUNTER Counter(workflow_start_total, Total workflow starts, [workflow_name]) WORKFLOW_DURATION_HISTOGRAM Histogram(workflow_duration_seconds, Workflow execution duration, [workflow_name]) WORKFLOW_ERROR_COUNTER Counter(workflow_error_total, Total workflow errors, [workflow_name, error_type]) app.post(/api/v1/review) async def review_order(...): WORKFLOW_START_COUNTER.labels(workflow_namereturn_review).inc() start_time time.time() try: result await return_review_workflow.ainvoke(state) duration time.time() - start_time WORKFLOW_DURATION_HISTOGRAM.labels(workflow_namereturn_review).observe(duration) return result except Exception as e: duration time.time() - start_time WORKFLOW_DURATION_HISTOGRAM.labels(workflow_namereturn_review).observe(duration) WORKFLOW_ERROR_COUNTER.labels( workflow_namereturn_review, error_typetype(e).__name__ ).inc() raise这样运维就可以在Grafana里看到退货审核流程的P95耗时是否超过3秒错误率是否突增哪个错误类型占比最高——这才是真正的生产就绪。5. 常见问题与避坑指南那些只有踩过才懂的坑5.1 “State更新不生效”Python字典的浅拷贝陷阱最常被问的问题“我明明在node里写了state[key] new_value但下一个node里还是旧值” 这是因为LangGraph默认传递的是state的引用但某些操作如{**state, key: new}会创建新字典而state[key] new是原地修改。但问题在于如果state里嵌套了list或dict浅拷贝会共享引用。反例def bad_node(state: OrderReviewState) - OrderReviewState: # state[retry_history] 是一个list state[retry_history].append({step: bad}) # 修改原list return state # 返回原state对象 def good_node(state: OrderReviewState) - OrderReviewState: # 正确深拷贝或新建list new_history state.get(retry_history, []) [{step: good}] return {**state, retry_history: new_history}验证方法在node开头打印id(state[retry_history])如果多个node里ID相同说明是同一对象。解决方案始终用copy.deepcopy(state)或确保所有嵌套结构都重建。5.2 “LLM死循环调用工具”提示词里的隐藏炸弹LLM有时会陷入无限调用同一个工具。比如OCR节点返回{tracking_number: SF123456789}但LLM又调用OCR工具查这个运单号形成死循环。根源在提示词里写了“如果需要更多信息可调用工具”。正确做法是在System Prompt里明确禁止重复调用。优化后的提示词你是一个退货审核助手目标是根据客户上传的运单图片判断是否批准退货。 你可用的工具 - ocr_tool: 仅在state中没有ocr_result时调用一次 - logistics_api: 仅在ocr_result存在且包含tracking_number时调用一次 禁止重复调用同一工具禁止在已有结果时再次调用。更保险的做法在tool_node里加去重逻辑def tool_node(state: OrderReviewState) - OrderReviewState: tool_call state.get(pending_tool_call) if not tool_call: return state # 检查是否已调用过此工具 called_tools state.get(called_tools, []) if tool_call[tool_name] in called_tools: return {**state, tool_result: ToolResult(successFalse, errorTOOL_ALREADY_CALLED)} # 执行调用... new_called called_tools [tool_call[tool_name]] return {**state, called_tools: new_called, tool_result: result}5.3 “本地测试通过线上报错”环境差异的终极排查清单时区问题本地datetime.now()是东八区K8s Pod默认UTC。state[timestamp]在本地是2024-06-15T14:22:3308:00线上是2024-06-15T06:22:3300:00导致时间比较逻辑失效。解决方案所有时间戳统一用UTC显示时再转本地。文件路径本地开发用./config.yamlDocker里路径是/app/config.yaml。用pathlib.Path(__file__).parent.parent / config.yaml代替硬编码路径。Secret管理API Key不能写死在代码里。用python-dotenv加载.env文件Docker部署时用--env-file或K8s Secret挂载。依赖版本漂移Poetry lock文件必须提交到Git且CI/CD流程中必须poetry install --no-dev禁用pip install -r requirements.txt。5.4 “流程卡死不结束”END节点的隐形陷阱workflow.add_edge(node_a, END)看起来很简单但END不是普通节点。如果某个node返回的state里error_count 10但你忘了在edge逻辑里处理这个case流程就会卡在那个node永远不走到END。解决方案所有条件边必须覆盖所有可能分支。# 错误漏掉error_count超限的情况 def should_proceed(state: OrderReviewState) - str: if state[ocr_result]: return logistics_node # 忘了处理 ocr_error_count 3 的情况 # 正确穷举所有可能 def should_proceed(state: OrderReviewState) - str: if state[ocr_result]: return logistics_node elif state.get(ocr_error_count, 0) 3: return manual_review_node else: return ocr_node # 继续重试5.5 “性能瓶颈不在LLM在IO”异步调用的正确姿势很多人以为瓶颈在Open