LangGraph生产级错误处理：RateLimitError与AuthError的四层防御体系

1. 这不是“加个try-catch”就能解决的事

LangChain 和 LangGraph 在生产环境里崩得悄无声息——你收到一条告警，日志里只有一行RateLimitError: 429 Too Many Requests，或者更糟，AuthError: Invalid API key，而整个Agent工作流已经卡死在某个节点上，下游服务等了三分钟才超时。这不是开发阶段那种“重跑一下就过了”的小毛病，这是会直接导致用户对话中断、订单流程失败、客服机器人失联的系统级风险。我去年在给一家金融客户做智能投顾Agent时，就因为没把错误处理当回事，上线第三天凌晨两点，API限流触发后整个对话树直接挂起，用户发来的“帮我查下上月基金收益”石沉大海，后台监控显示37个会话同时卡在retrieve_stock_data节点，没人知道是该重试、降级、还是跳过。后来我们花了整整两周时间重构错误路径，才把平均故障恢复时间从8分钟压到22秒。LangChain 的Runnable是链式执行，LangGraph 的StateGraph是状态驱动，它们的错误传播机制和传统 Web 服务完全不同：一个节点抛异常，不等于整个流程终止，而是取决于你有没有定义interrupt、fallback、retry_policy，以及是否在add_node时显式声明了error_handler。很多人学 LangGraph 教程时只盯着send(node_name, state)怎么传数据，却忽略了send本身不会捕获异常——它只是把消息推到队列，真正的错误发生在节点函数内部执行时。而RateLimitError和AuthError这两类错误，恰恰是最容易被忽略的“非业务错误”：它们不反映模型逻辑缺陷，却暴露了基础设施层的脆弱性。你不能指望 LLM 自己判断“我现在被限流了，先睡5秒再试”，这必须由框架层兜底。所以这篇不是讲怎么写个except RateLimitError:，而是拆解在真实高并发、多租户、混合模型调用（OpenAI + 本地Llama + 第三方RAG引擎）的生产场景下，如何让 LangChain/LangGraph 具备像 Spring Boot 的@Retryable或 Kubernetes 的 Pod 重启策略那样的韧性。适合正在用 LangGraph 搭建客服Agent、金融风控Agent、或企业知识库问答系统的工程师，也适合刚学完langchain入门教程、正准备把 demo 推到生产环境的开发者——别等线上出事才翻文档，那会发现官方文档里关于retry的参数说明只有两行，而实际要填的坑有十七个。

2. 错误类型本质与传播路径深度拆解

2.1 RateLimitError：不是“请求太快”，而是“配额耗尽”的信号灯

RateLimitError 表面看是 HTTP 429 状态码，但它的底层含义远比“你发请求太猛”复杂。以 OpenAI 为例，其限流是三级嵌套结构：每分钟请求数（RPM）、每分钟Token数（TPM）、账户总配额（Monthly Quota）。LangChain 的ChatOpenAI默认配置中，max_retries=2只针对网络瞬断（如 DNS 失败、连接超时），对 429 是无效的——因为 429 是服务端明确拒绝，不是临时不可达。我实测过：当 RPM 触顶时，OpenAI 返回的Retry-Afterheader 是 60 秒，但如果你用默认retry_backoff_factor=1，第一次重试会在 1 秒后，第二次在 2 秒后，完全无视服务端建议。更麻烦的是，LangGraph 的StateGraph在节点执行中抛出 RateLimitError，会直接中断当前stream()调用，state 停在出错节点，后续send()操作根本不会触发。这不是 bug，是设计使然：LangGraph 假设你已为每个节点定义了容错边界。所以关键不是“怎么重试”，而是“在哪重试”。比如你在retrieve_knowledge节点调用向量数据库，同时又在generate_response节点调用 LLM，这两个节点的限流来源完全不同——前者受 Milvus 或 PGVector 连接池限制，后者受 OpenAI 配额限制。混在一起用全局重试策略，会导致知识检索失败时错误地重试 LLM 调用，浪费 Token。正确做法是分层拦截：在RunnableLambda封装的节点函数内，用tenacity库做精细化重试，而不是依赖 LangChain 内置的max_retries。

2.2 AuthError：认证失效的连锁反应比想象中更致命

AuthError 看似简单——API Key 过期或权限不足。但在 LangGraph 的 Agent 架构中，它会引发雪崩式中断。举个真实案例：某 SaaS 客户的 Agent 使用了三个模型服务：OpenAI（主LLM）、Cohere（备用摘要）、自建 Llama3（本地RAG）。所有服务共用一个密钥管理服务（KMS），当 KMS 因网络抖动返回空密钥时，LangChain 的ChatOpenAI初始化会静默失败，但StateGraph.add_node("llm_call", llm.invoke)仍能注册成功——因为llm.invoke是延迟绑定的。直到第一个用户请求到达，llm_call节点执行时才抛出AuthError，此时StateGraph的stream()已启动，state 中的messages列表已存入用户输入，但后续所有节点都无法执行。更隐蔽的问题是：LangGraph 默认不记录节点初始化错误，你只能在stream()的on_chain_start回调里捕获，但这时 state 已污染。解决方案不是“检查密钥再启动”，而是采用Lazy Initialization + Circuit Breaker模式：把 LLM 实例化封装进一个带熔断器的工厂函数，首次调用时校验密钥有效性并缓存结果，后续调用直接复用；一旦连续3次AuthError，熔断器打开，自动切换到备用模型，并触发告警。这要求你放弃llm = ChatOpenAI()的直觉写法，改用llm_factory = lambda: get_llm_by_priority(["openai", "cohere", "llama3"])。

2.3 错误传播的三大陷阱：LangChain 与 LangGraph 的根本差异

LangChain 的错误传播是线性的：chain.invoke(input)→ 某个Runnable抛异常 → 整个链终止。LangGraph 则是状态驱动的，错误传播路径取决于图结构：

• 无条件边（conditional edge）未定义错误分支：比如graph.add_conditional_edges("llm_call", route_to_tool)，但route_to_tool函数在 LLM 返回格式错误时抛出ValueError，这个异常不会被conditional_edges捕获，而是直接向上抛给stream()，导致整个图崩溃。
• 节点间状态传递隐式依赖：send("tool_executor", state)时，如果state中tool_calls字段为空（因前序节点llm_call因RateLimitError未执行），tool_executor节点会因AttributeError崩溃，但这个错误和原始限流无关，属于状态污染。
• interrupt机制被误用：很多教程教用graph.add_edge("__interrupt", "llm_call")实现人工干预，但__interrupt是特殊节点名，若在错误处理中手动send("__interrupt", state)，会绕过所有条件边逻辑，直接跳转，造成状态不一致。

我画过一张生产环境错误传播拓扑图（文字版）：

用户请求 → StateGraph.stream() ↓ [llm_call] ——RateLimitError→ [retry_llm] ——成功→ [route_to_tool] ↓ ↓ AuthError ValueError（格式错误） ↓ ↓ [failover_to_backup] [handle_format_error] ↓ ↓ [log_and_alert] ←——————— [update_state_with_suggestion]

关键在于：每个箭头都必须是显式定义的边，不能依赖隐式异常传播。LangGraph 不是 try-catch 的容器，而是错误路由的编排器。

3. 生产级错误处理四层架构设计

3.1 第一层：节点级防御——用 tenacity 实现语义化重试

不要用 LangChain 内置的max_retries，它太粗粒度。必须为每个节点定制重试策略，核心是区分“可重试错误”和“不可重试错误”。以llm_call节点为例：

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type from langchain_core.exceptions import RateLimitError, TimeoutError # 针对 RateLimitError 的专用重试：遵守 Retry-After，指数退避 @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10), # 最小等待4秒，最大10秒 retry=retry_if_exception_type(RateLimitError), reraise=True ) def robust_llm_invoke(llm, messages): try: return llm.invoke(messages) except RateLimitError as e: # 解析 OpenAI 的 Retry-After header，覆盖 tenacity 的 wait if hasattr(e, 'response') and e.response and e.response.headers.get('Retry-After'): retry_after = int(e.response.headers['Retry-After']) time.sleep(retry_after) # 强制等待服务端指定时间 raise e # 封装成 Runnable llm_node = RunnableLambda(lambda state: robust_llm_invoke(state["llm"], state["messages"]))

这里的关键细节：

• wait_exponential的min=4是硬性要求：OpenAI 的Retry-After最小值是1秒，但实际中1秒重试大概率再次429，4秒是实测平衡点；
• reraise=True确保最终失败时异常透出，供上层路由处理；
• retry_if_exception_type严格限定只重试RateLimitError，避免把AuthError也重试（重试无效密钥毫无意义）。

对比tool_call节点，重试策略完全不同：

# 工具调用重试：关注网络瞬断，而非限流 @retry( stop=stop_after_attempt(2), wait=wait_fixed(0.5), # 固定0.5秒，工具API通常响应快 retry=retry_if_exception_type((TimeoutError, ConnectionError)), reraise=True ) def robust_tool_invoke(tool, input): return tool.invoke(input)

提示：tenacity的before_sleep回调可用于记录每次重试，但切忌在回调里修改state——节点函数的state参数是副本，修改不影响图状态。

3.2 第二层：图级路由——用 conditional_edges 构建错误决策树

LangGraph 的灵魂在于add_conditional_edges。错误处理不能靠 try-catch，而要靠状态驱动的路由。核心思想：把错误类型编码进 state，让图自己决定下一步。

首先，定义错误状态字段：

class AgentState(TypedDict): messages: list[BaseMessage] error: Optional[str] # 新增：存储错误类型，如 "rate_limit", "auth" error_count: int # 新增：同一错误连续发生次数 last_retry_time: Optional[float] # 新增：上次重试时间戳，用于防抖

然后，在每个可能出错的节点后，添加错误路由：

def should_retry_or_fail(state: AgentState) -> str: """根据错误类型和计数决定路由""" if not state.get("error"): return "continue" # 无错误，走正常流程 error_type = state["error"] count = state["error_count"] if error_type == "rate_limit": if count < 3: return "retry_llm" # 重试最多3次 else: return "failover_to_backup" # 切换备用模型 elif error_type == "auth": if count == 1: return "refresh_api_key" # 首次认证失败，尝试刷新 else: return "alert_and_terminate" # 多次失败，告警并终止 else: return "handle_unknown_error" # 绑定到图 graph.add_conditional_edges( "llm_call", should_retry_or_fail, { "continue": "route_to_tool", "retry_llm": "llm_call", # 循环回自身，实现重试 "failover_to_backup": "backup_llm_call", "refresh_api_key": "key_refresher", "alert_and_terminate": "log_alert", } )

这个设计的精妙之处在于：

• retry_llm边指向自身，形成循环，但通过state["error_count"]递增控制次数，避免无限循环；
• refresh_api_key是独立节点，负责调用 KMS 接口获取新密钥，成功后重置error_count；
• 所有错误分支都显式定义，没有“默认 fallback”，杜绝意外跳转。

3.3 第三层：服务级熔断——集成 circuitbreaker 库防雪崩

当某个下游服务（如向量数据库）持续失败，必须熔断，否则会拖垮整个 Agent。LangGraph 本身不提供熔断，需引入circuitbreaker库：

from circuitbreaker import CircuitBreaker, CircuitBreakerError # 为向量检索创建熔断器 vector_search_breaker = CircuitBreaker( failure_threshold=5, # 连续5次失败开启熔断 recovery_timeout=60, # 熔断后60秒尝试恢复 expected_exception=ConnectionError # 只对连接错误熔断 ) @vector_search_breaker def safe_vector_search(query: str) -> list[Document]: return vectorstore.similarity_search(query) # 在节点中使用 def vector_search_node(state: AgentState): try: results = safe_vector_search(state["messages"][-1].content) return {"documents": results} except CircuitBreakerError: # 熔断器开启时抛出此异常 return {"documents": [], "error": "vector_service_down"} except Exception as e: # 其他异常，如解析错误 return {"documents": [], "error": "vector_search_failed"}

熔断器开启后，所有safe_vector_search调用立即返回CircuitBreakerError，不发起真实请求。这比重试更激进，但对基础设施故障是必要的。注意：CircuitBreaker的recovery_timeout必须大于你的重试间隔，否则刚恢复就又被打垮。

3.4 第四层：可观测性闭环——错误日志、指标、告警三位一体

生产环境错误处理的终点不是“修复”，而是“预防”。必须建立可观测性闭环：

• 结构化日志：用structlog替代logging，在每个节点入口/出口记录state关键字段：

  import structlog logger = structlog.get_logger() def llm_call_node(state: AgentState): logger.info("llm_call_start", user_id=state.get("user_id"), model=state["llm"].model_name, message_len=len(state["messages"][-1].content)) # ... 执行逻辑 ... logger.info("llm_call_success", token_usage=response.response_metadata.get("token_usage", {})) return {...}

• Prometheus 指标：暴露关键指标：

  • langgraph_node_errors_total{node="llm_call",error_type="rate_limit"}：按节点和错误类型统计
  • langgraph_state_transitions_total{from="llm_call",to="retry_llm"}：路由跳转次数
  • circuitbreaker_state{service="vector_search",state="open"}：熔断器状态

• 告警规则（Prometheus Alertmanager）：

  - alert: LangGraphRateLimitSpikes expr: rate(langgraph_node_errors_total{error_type="rate_limit"}[5m]) > 10 for: 2m labels: severity: warning annotations: summary: "RateLimitError spike on {{ $labels.node }}" description: "More than 10 rate limit errors per minute for 2 minutes" - alert: LangGraphAuthErrorPersistent expr: count by (node) (langgraph_node_errors_total{error_type="auth"}[1h]) > 5 for: 10m labels: severity: critical annotations: summary: "Persistent AuthError on {{ $labels.node }}" description: "AuthError occurred more than 5 times in last hour"

这套体系的价值在于：当RateLimitError频繁出现时，你不仅能立刻告警，还能通过 Grafana 查看是哪个租户的请求导致（user_id标签），进而联系客户调整配额，而不是被动救火。

4. 实操：从零搭建一个抗错的 LangGraph Agent

4.1 环境准备与依赖锁定

生产环境严禁pip install langgraph这种模糊版本。必须锁定精确版本并验证兼容性：

# requirements.txt langchain==0.1.16 langgraph==0.1.12 tenacity==8.2.3 circuitbreaker==1.4.0 structlog==23.3.0 prometheus-client==0.17.1 # 注意：langchain 0.1.16 与 langgraph 0.1.12 是经过生产验证的组合 # 高于此版本的 langgraph 0.2.x 引入了 async stream，但错误处理 API 有 breaking change

注意：langgraph0.1.x 和 0.2.x 的StateGraph初始化方式不同。0.1.x 用StateGraph(StateClass)，0.2.x 用StateGraph().add_node(...)。本文基于 0.1.12，因其在金融客户环境中稳定运行超6个月。

4.2 定义鲁棒的 State 类型

from typing import List, Optional, Dict, Any from langchain_core.messages import BaseMessage from langchain_core.pydantic_v1 import BaseModel, Field class AgentState(BaseModel): """生产环境强化版 State""" messages: List[BaseMessage] = Field(default_factory=list) # 错误上下文 error: Optional[str] = None error_count: int = 0 last_retry_time: Optional[float] = None # 服务状态 llm_status: str = "active" # "active", "degraded", "down" vector_status: str = "active" # 诊断信息 trace_id: str = "" # 用于链路追踪 user_id: str = "" # 用于租户隔离 class Config: arbitrary_types_allowed = True

关键点：

• Field(default_factory=list)确保messages总是列表，避免None导致AttributeError；
• llm_status和vector_status字段用于熔断器状态同步，避免重复查询；
• trace_id和user_id是可观测性基石，必须从请求头注入。

4.3 构建带错误处理的完整图

from langgraph.graph import StateGraph from langgraph.checkpoint.memory import MemorySaver # 初始化检查点（生产环境必须用 Redis，此处简化） checkpointer = MemorySaver() # 创建图 graph = StateGraph(AgentState) # 定义节点 def entry_node(state: AgentState) -> dict: """入口节点：清洗输入，注入 trace_id""" # 从请求上下文提取 trace_id（实际中从 FastAPI request.state 获取） state.trace_id = "trace_" + str(int(time.time() * 1000000)) return {"trace_id": state.trace_id} def llm_call_node(state: AgentState) -> dict: """带重试和错误捕获的 LLM 调用""" try: # 使用前面定义的 robust_llm_invoke response = robust_llm_invoke(state.llm, state.messages) return { "messages": [response], "error": None, "error_count": 0, "last_retry_time": None } except RateLimitError as e: logger.warning("RateLimitError in llm_call", trace_id=state.trace_id, error=str(e)) return { "error": "rate_limit", "error_count": state.error_count + 1, "last_retry_time": time.time() } except AuthError as e: logger.error("AuthError in llm_call", trace_id=state.trace_id, error=str(e)) return { "error": "auth", "error_count": state.error_count + 1, "last_retry_time": time.time() } def backup_llm_call_node(state: AgentState) -> dict: """备用 LLM 调用，仅在主 LLM 熔断时触发""" # 此处调用 Cohere 或本地 Llama pass def log_alert_node(state: AgentState) -> dict: """告警节点：发送 Slack/Webhook，并返回友好错误消息""" # 发送告警 send_slack_alert(f"CRITICAL: AuthError persistent on {state.trace_id}") # 返回用户可见的错误消息 return { "messages": [ AIMessage(content="抱歉，服务暂时不可用，请稍后再试。") ] } # 添加节点 graph.add_node("entry", entry_node) graph.add_node("llm_call", llm_call_node) graph.add_node("backup_llm_call", backup_llm_call_node) graph.add_node("log_alert", log_alert_node) # 添加边 graph.set_entry_point("entry") graph.add_edge("entry", "llm_call") # 错误路由边 def route_after_llm(state: AgentState) -> str: if state.error == "rate_limit" and state.error_count < 3: return "llm_call" # 重试 elif state.error == "rate_limit": return "backup_llm_call" # 切换备用 elif state.error == "auth": return "log_alert" # 认证失败直接告警 else: return "__end__" # 其他错误终止 graph.add_conditional_edges( "llm_call", route_after_llm, { "llm_call": "llm_call", # 重试边 "backup_llm_call": "backup_llm_call", "log_alert": "log_alert", "__end__": "__end__" } ) graph.add_edge("backup_llm_call", "__end__") graph.add_edge("log_alert", "__end__") # 编译图 app = graph.compile(checkpointer=checkpointer)

4.4 部署时的检查清单

上线前必须逐项核验：

检查项	说明	验证方法
重试策略生效	RateLimitError是否按Retry-After等待	用pytestmock OpenAI 返回 429 和Retry-After: 5，检查time.sleep调用
熔断器触发	向量库连续5次失败后，第6次是否立即返回CircuitBreakerError	用unittest.mockpatchvectorstore.similarity_search，模拟5次ConnectionError
状态字段完整性	error_count是否在重试时正确递增	在llm_call_node中打印state.error_count，观察连续请求变化
日志结构化	structlog输出是否包含trace_id和user_id	查看日志文件，grep"trace_id"
指标暴露	/metrics端点是否返回langgraph_node_errors_total	curl http://localhost:8000/metrics | grep langgraph_node_errors_total

实操心得：我们曾因忘记在llm_call_node的返回字典中重置error_count=0，导致一次成功后error_count仍为1，下次错误时直接跳过重试进入backup_llm_call。这个 bug 在压测时才暴露——因为压测流量大，错误频发，而日常测试只测单次流程。

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

5.1 “重试了3次还是429，retry_after 没生效？”

现象：robust_llm_invoke函数中e.response.headers.get('Retry-After')返回None，导致time.sleep(retry_after)报错。

根因：OpenAI 的 429 响应中，Retry-Afterheader 并非总是存在。官方文档说明：“When the rate limit is exceeded, the API returns a 429 status code with a Retry-After header indicating how long to wait before retrying. However, this header may be omitted in some cases.” 实测发现，当 TPM（Token Per Minute）超限时，Retry-After常为空；而 RPM（Request Per Minute）超限时，该 header 存在。

解决方案：增加 fallback 逻辑：

def robust_llm_invoke(llm, messages): try: return llm.invoke(messages) except RateLimitError as e: retry_after = 0 if hasattr(e, 'response') and e.response and e.response.headers.get('Retry-After'): retry_after = int(e.response.headers['Retry-After']) else: # Fallback: 根据错误消息推测 if "TPM" in str(e): retry_after = 60 # TPM 超限，保守等待60秒 else: retry_after = 10 # RPM 超限但无 header，等待10秒 time.sleep(max(4, retry_after)) # 至少等待4秒 raise e

5.2 “AuthError 后切换备用模型，但备用模型也报 AuthError，陷入死循环”

现象：backup_llm_call节点同样抛出AuthError，由于route_after_llm函数未处理备用节点的错误，图直接崩溃。

根因：错误路由只定义在llm_call节点后，backup_llm_call是终端节点，没有自己的错误路由。

解决方案：为备用节点也添加条件边，且设置更严格的终止策略：

# 在 backup_llm_call 后添加路由 def route_after_backup(state: AgentState) -> str: if state.error == "auth": return "final_failure" # 终极失败节点 else: return "__end__" graph.add_conditional_edges( "backup_llm_call", route_after_backup, { "final_failure": "final_failure", "__end__": "__end__" } ) def final_failure_node(state: AgentState) -> dict: # 记录终极失败，返回兜底消息 logger.critical("All LLM services failed", trace_id=state.trace_id) return { "messages": [AIMessage(content="系统繁忙，请稍后再试。")] } graph.add_node("final_failure", final_failure_node) graph.add_edge("final_failure", "__end__")

5.3 “StateGraph.stream() 返回空生成器，什么日志都没有”

现象：调用app.stream({"messages": [HumanMessage(content="hi")]})后，for 循环直接结束，无任何输出，也无错误日志。

根因：StateGraph的stream()方法在遇到未处理的异常时，会静默消耗生成器，而不是抛出异常。常见于entry_node中state.trace_id = ...时，state是TypedDict实例，不支持属性赋值。

排查技巧：

• 在entry_node开头加print(f"Entry node called with state: {type(state)}")；
• 检查State类型是否继承自TypedDict（不可变）还是BaseModel（可变）；
• 用try...except Exception as e: print(e); raise包裹节点函数，强制暴露错误。

修正：确保AgentState是BaseModel子类（如 4.2 节所示），或改用字典更新：

def entry_node(state: dict) -> dict: # 如果 state 是 dict，直接更新 state["trace_id"] = "trace_" + str(int(time.time() * 1000000)) return state

5.4 “熔断器开了，但日志里还在疯狂打 ConnectionError”

现象：vector_search_breaker熔断后，logger.info仍在高频记录ConnectionError。

根因：circuitbreaker的CircuitBreakerError是运行时异常，但logger.info在except块外执行。正确结构应为：

@vector_search_breaker def safe_vector_search(query: str): return vectorstore.similarity_search(query) def vector_search_node(state: AgentState): try: results = safe_vector_search(state["query"]) return {"documents": results} except CircuitBreakerError: logger.warning("Circuit breaker OPEN for vector search", trace_id=state.trace_id) return {"documents": [], "error": "vector_service_down"} except Exception as e: logger.error("Vector search failed", trace_id=state.trace_id, error=str(e)) return {"documents": [], "error": "vector_search_failed"}

注意：CircuitBreakerError必须在@vector_search_breaker装饰的函数内部抛出，外部try-except才能捕获。如果safe_vector_search函数内还有其他try-except吞掉了异常，熔断器无法感知失败。

5.5 “Prometheus 指标里 langgraph_node_errors_total 为0，但我知道有错误”

现象：应用日志显示RateLimitError，但 Prometheus 查询langgraph_node_errors_total返回空。

根因：指标计数器未在错误发生时 increment。必须在每个节点的except块中显式调用：

from prometheus_client import Counter ERROR_COUNTER = Counter( 'langgraph_node_errors_total', 'Total number of errors in LangGraph nodes', ['node', 'error_type'] ) def llm_call_node(state: AgentState) -> dict: try: # ... 正常逻辑 except RateLimitError as e: ERROR_COUNTER.labels(node="llm_call", error_type="rate_limit").inc() # ... 其余逻辑

速查表：LangGraph 错误处理黄金法则

场景	正确做法	错误做法	后果
RateLimitError	在节点内用tenacity重试，wait_exponential(min=4)	依赖 LangChainmax_retries=2	重试间隔太短，持续429
AuthError	单独路由到refresh_api_key节点，失败后告警	在llm_call内time.sleep(1)后重试	浪费资源，密钥无效
状态污染	所有节点返回dict更新state，不直接修改state对象	state.error = "xxx"直接赋值	StateGraph状态不一致
熔断器	CircuitBreaker包裹具体服务调用，except CircuitBreakerError单独处理	try-except捕获所有异常统一处理	熔断器失效，雪崩
日志	structlog记录trace_id和user_id，INFO级别记录成功，WARNING/ERROR记录失败	print()或logging.info()无结构	无法关联请求，排查困难

6. 我在金融客户项目中的血泪经验

最后分享一个真实教训：我们最初为投顾 Agent 设计了“三级降级”策略——主 LLM 失败 → 备用 LLM → 规则引擎兜底。听起来很完美，但上线后发现规则引擎的响应时间高达 800ms（因要查 12 张数据库表），而用户平均等待阈值是 1.2 秒。当主 LLM 因限流失败，切换到规则引擎后，整体 P95 延迟从 450ms 暴涨到 1100ms，大量用户流失。我们以为是规则引擎慢，花了一周优化 SQL，结果收效甚微。后来用py-spy采样发现，真正瓶颈是StateGraph在send("rule_engine", state)后，等待rule_engine节点返回时，stream()的协程被阻塞，而其他会话的请求也在排队。根本解法不是优化规则引擎，而是异步化降级路径：把规则引擎调用放到asyncio.to_thread()中，避免阻塞事件循环。代码改动很小：

import asyncio async def rule_engine_node(state: AgentState) -> dict: # 同步规则引擎调用，放入线程池 result = await asyncio.to_thread(sync_rule_engine_invoke, state["messages"][-1].content) return {"messages": [AIMessage(content=result)]}

这一改，P95 延迟回到 520ms。这件事让我明白：LangGraph