企业官网建设流程全解析

1. 项目概述：这不是写个脚本，而是让业务逻辑自己跑起来

“Python + Agent SDK：把业务场景转化为自动化工作流的完整流程”——这句话里藏着一个被很多人忽略的关键动词：“转化”。不是“用Python写个自动化脚本”，也不是“调用几个API凑成一个链路”，而是把真实业务中那些需要人反复判断、跨系统切换、查文档翻记录、在Excel里拖拽公式、再发邮件确认的“活儿”，原样搬进代码里，让它具备感知、决策、执行、纠错、重试、归档的完整闭环能力。我做过银行对公信贷审批辅助系统，也搭过电商售后自动定责引擎，还重构过制造业设备报修工单分派流程，所有这些项目的起点，都不是“我们要做个AI应用”，而是业务方拍着桌子说：“这个环节每天要花3个人2小时，错一次就要赔5000块，你们能不能让它别出错？”

核心关键词Python在这里不是语法糖或胶水语言，而是整个工作流的底盘——它提供足够灵活的类型系统支撑多源数据融合（比如把ERP里的JSON、CRM里的CSV、邮件正文里的表格片段统一喂给Agent），有成熟的异步IO生态应对高并发任务调度（比如同时处理200个客户投诉工单），更重要的是，它的包管理机制（pip+venv）和模块化设计，让不同业务线的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的破局点在于强制你把“状态”显式定义出来。我们先定义一个OrderReviewState：

from 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_count=2、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系统订单主键，不可为空，类型为str
• raw_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_factory=dict)

然后封装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": f"Bearer {self.api_key}"}, timeout=self.timeout ) if response.status_code == 200: return ToolResult(success=True, data=response.json()) else: return ToolResult(success=False, error=f"HTTP {response.status_code}") except requests.Timeout: return ToolResult(success=False, error="TIMEOUT") except Exception as e: return ToolResult(success=False, error=str(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(success=False, error="Unknown tool")} try: result = tool.invoke(**tool_call["arguments"]) return {**state, "tool_result": result} except Exception as e: return {**state, "tool_result": ToolResult(success=False, error=str(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( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10), retry=retry_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_threshold=5, # 连续5次失败触发熔断 recovery_timeout=60, # 熔断60秒后尝试恢复 expected_exception=requests.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. 实操全流程：从零搭建一个可交付的退货审核Agent

4.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 mypy

pyproject.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.py：

import 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_url="https://api.ocr.example.com", api_key="sk-xxx" ) result = ocr_tool.invoke(image_b64, language="zh") if result.success: logger.info(f"OCR success for order {state['order_id']}") return { **state, "ocr_result": result.data, "ocr_error_count": 0 } else: logger.error(f"OCR 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(f"Unexpected 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": f"UNEXPECTED: {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.py：

from 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( checkpointer=MemorySaver(), callbacks=[{ "on_chain_start": lambda x: logger.info(f"Start {x.name}"), "on_chain_end": lambda x: logger.info(f"End {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_func=get_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(f"Workflow execution failed: {e}") raise HTTPException(status_code=500, detail="Internal 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_name="return_review").inc() start_time = time.time() try: result = await return_review_workflow.ainvoke(state) duration = time.time() - start_time WORKFLOW_DURATION_HISTOGRAM.labels(workflow_name="return_review").observe(duration) return result except Exception as e: duration = time.time() - start_time WORKFLOW_DURATION_HISTOGRAM.labels(workflow_name="return_review").observe(duration) WORKFLOW_ERROR_COUNTER.labels( workflow_name="return_review", error_type=type(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(success=False, error="TOOL_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:33+08:00，线上是2024-06-15T06:22:33+00:00，导致时间比较逻辑失效。解决方案：所有时间戳统一用UTC，显示时再转本地。
• 文件路径：本地开发用./config.yaml，Docker里路径是/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