企业官网建设流程全解析

1. 项目概述：当LangGraph真正“开口说话”——不是调用API，而是让图结构自己驱动LLM推理

你有没有试过这样一种状态：画完一张漂亮的LangGraph流程图，节点定义清晰、边逻辑严密、循环条件也写得滴水不漏，可一到执行环节，整个图就卡在某个LLMNode上，半天没响应，或者返回一堆格式错乱的JSON？我去年带三个实习生做智能客服编排系统时，就反复撞上这堵墙——图是活的，但LLM像被绑住了嘴。后来才发现，问题根本不在图结构本身，而在于我们一直把LLM当成“被动应答器”，却忘了LangGraph的设计哲学：它要的不是一个能回答问题的模型，而是一个能参与图中状态流转、能主动触发分支、能根据中间结果动态改写后续路径的“协作者”。这篇讲的，就是怎么让LangGraph真正和LLM建立起双向呼吸式的连接。核心不是“怎么调用”，而是“怎么共谋”。关键词很直白：LangGraph、LLM集成、状态驱动、工具调用桥接、流式响应对齐。它适合三类人：正在用LangGraph搭复杂Agent但总在LLM节点处掉链子的开发者；想把已有LangChain链式逻辑平滑迁移到图结构中的工程师；还有那些已经能写基础ReAct Agent、但卡在“如何让多步决策真正闭环”的技术负责人。这不是LangChain文档的复述，而是我在真实交付项目里，把LangGraph和Qwen2.5-72B、Claude-3.5-Sonnet、以及本地部署的Phi-3-mini全跑通后，撕下来的七张调试日志纸和四次重写提示词模板的总结。

2. 整体设计思路拆解：为什么不能直接llm.invoke()？图结构下的LLM必须“有状态、可中断、能反馈”

很多人第一次尝试LangGraph+LLM，会本能地写一个最简节点：

def llm_node(state: dict) -> dict: response = llm.invoke(state["messages"]) return {"messages": [response]}

然后发现：图跑不起来，或者跑起来但完全不按预期分支。这不是代码bug，而是范式错位。LangGraph不是函数式编程，它的核心是状态机（State Machine），而LLM在其中的角色，必须适配这个状态机的三大刚性约束：

2.1 约束一：LLM调用必须是“可中断”的，而非“一锤定音”

传统llm.invoke()是阻塞式调用，直到模型吐完所有token才返回。但在LangGraph里，一个节点可能需要在收到部分流式响应时就决定是否跳转到tools节点，或者在检测到"need_more_info"标记时立刻进入ask_user分支。LangGraph的interrupt_before和interrupt_after机制，要求LLM调用本身支持分段响应与状态快照。实测下来，直接用invoke会导致整个图在该节点无限等待，因为LangGraph无法在模型输出中途插入中断点。解决方案是必须使用llm.stream()配合自定义stream_parser，把原始流式输出按语义块切分（比如按\n\n、<tool>标签、或正则匹配"Action:"），每切出一块，就更新一次state，并触发一次图的状态检查。我试过用langchain_core.runnables里的RunnableGenerator包装stream，但最终发现手写一个轻量级StreamingLLMNode更可控——它内部维护一个buffer，只在buffer积累到完整语义单元（如一个JSON对象、一个工具调用块）时才提交state更新。

2.2 约束二：LLM必须“理解图上下文”，而非仅看当前消息

LangChain的ChatPromptTemplate默认只拼接messages，但LangGraph的state是动态演化的。比如在客服场景中，state里除了messages，还有user_profile: {"tier": "premium", "last_issue": "billing"}、session_context: {"language": "zh", "timezone": "Asia/Shanghai"}。如果LLM节点只读state["messages"]，它就永远不知道用户是VIP还是普通用户，自然无法触发escalate_to_human分支。所以，真正的集成不是“把LLM塞进图”，而是“让LLM成为图状态的原生解析器”。我的做法是在每个LLM节点前，强制注入一个state_enricher节点，它把所有非消息字段（user_profile,session_context,tool_results）序列化为一段结构化提示词前缀，再拼接到messages[-1].content开头。例如：

【用户画像】VIP用户，上次投诉账单问题；【会话上下文】中文服务，东八区时间；【可用工具】get_order_status, refund_request, escalate_to_human；【当前任务】判断是否需人工介入...

这样，LLM的输入不再是孤立的对话，而是带着全图状态的“作战地图”。测试数据表明，这种显式状态注入，让分支准确率从68%提升到92%，尤其在多条件嵌套判断中效果显著。

2.3 约束三：LLM输出必须“可解析、可验证、可回滚”，而非自由文本

LangGraph依赖LLM输出来驱动conditional_edges。如果LLM返回"I think we should check the order status"，图根本没法解析该走哪条边。必须强制LLM输出机器可读的结构化指令。这里有两个主流方案：

• 方案A（推荐）：用JSON Schema约束输出。通过PydanticOutputParser定义严格schema，如{"action": "get_order_status", "params": {"order_id": "12345"}}，并在system prompt里明确要求“只输出合法JSON，不加任何解释文字”。实测发现，Qwen2.5-72B在schema约束下，JSON合规率99.3%，而GPT-4o只有94.1%（它总爱在JSON外加个Here is the result:）。
• 方案B：用特殊标记分隔。要求LLM在输出中用<action>、</action>包裹动作块，用<reason>说明理由。好处是兼容性好，所有模型都支持；缺点是解析器要写正则，且容易被模型“误标”。我在金融风控项目里用过此方案，但必须加一层post_processor校验：若未匹配到<action>，则自动触发fallback_to_safe_action节点。

提示：不要迷信“模型越强，输出越规范”。我在对比测试中发现，Phi-3-mini在严格JSON模式下比Llama3-70B更稳定——小模型参数少，反而更“听话”。选型时，稳定性优先于参数量。

3. 核心细节解析与实操要点：从Prompt工程到流式对齐，七个必须死磕的细节

把LLM接入LangGraph，90%的坑不在代码，而在细节。以下是我在四个生产项目里踩出来的七条铁律，每一条都对应着一次线上故障的回滚记录。

3.1 细节一：System Prompt必须包含“图角色说明书”，而非通用指令

很多人的system prompt还停留在"You are a helpful AI assistant"。这对LangGraph是灾难。LLM需要知道自己此刻是“图中的决策节点”，它的输出将直接触发边跳转。我的标准模板是：

You are a state-aware decision node in a LangGraph workflow. Your role is NOT to answer user questions directly, but to output ONLY one of the following JSON objects based on the current state and conversation history: { "next": "node_name", "action": "tool_name", "params": { ... }, "reason": "brief explanation for this choice" } Rules: - NEVER output any text outside the JSON object. - If you need more information from the user, set "next": "ask_user". - If the user request violates policy, set "next": "reject_request". - All "params" must match the tool's exact input schema.

这个prompt的关键，在于把LLM的“身份认知”从“助手”切换为“图节点操作员”。上线后，conditional_edges的误跳转率从31%降到2.4%。

3.2 细节二：Messages必须做“角色归一化”，否则模型会混淆谁在说话

LangGraph的state里messages通常是[HumanMessage, AIMessage, HumanMessage...]，但不同LLM对role字段的敏感度天差地别。Qwen系列严格区分user/assistant/system，而Claude只认user/assistant，Llama3则对system角色有特殊权重。如果直接把LangChain的HumanMessage传给Claude，它会把HumanMessage当成assistant的回复，导致逻辑全乱。解决方案是统一预处理：

def normalize_messages(messages: list) -> list: normalized = [] for msg in messages: if isinstance(msg, HumanMessage): normalized.append({"role": "user", "content": msg.content}) elif isinstance(msg, AIMessage): normalized.append({"role": "assistant", "content": msg.content}) elif isinstance(msg, SystemMessage): # Claude不支持system，合并到首条user message if normalized and normalized[0]["role"] == "user": normalized[0]["content"] = f"SYSTEM: {msg.content}\n{normalized[0]['content']}" else: normalized.append({"role": "user", "content": f"SYSTEM: {msg.content}"}) return normalized

这个函数在每个LLM节点入口调用，确保输入格式100%匹配目标模型的tokenizer期望。实测避免了73%的“角色错位”类幻觉。

3.3 细节三：Tool调用必须走“双通道验证”，不能信LLM的一面之词

LLM说要调用get_order_status，你就真去调？大错。LangGraph的tools节点必须自带验证层。我的标准结构是：

1. 通道一（前置校验）：在tool_node入口，用pydantic.BaseModel校验LLM输出的params是否符合工具签名。例如get_order_status要求order_id: str且长度为10位数字，校验失败则直接跳invalid_params分支。
2. 通道二（后置断言）：工具执行后，返回结果必须包含assertions: List[str]字段，如["order_status != 'cancelled'", "refund_eligible == True"]。tool_result_handler节点会逐条执行这些断言，任一失败即触发retry_with_correction。

这套双保险让我在电商项目里，把因参数错误导致的工具调用失败率从18%压到0.3%。记住：LLM是策士，不是执行官；工具节点才是真正的守门人。

3.4 细节四：流式响应必须绑定“语义块ID”，否则图状态会错乱

用llm.stream()时，如果只是简单地把每个chunk追加到state["messages"]，会出现“半截响应被当完整指令”的问题。比如LLM流式输出：

Chunk1: {"action": "get_order_status" Chunk2: , "params": {"order_id": "12345"}}

如果图在Chunk1后就检查state，会得到一个非法JSON，直接崩溃。解决方案是给每个语义块分配唯一ID，并在stream parser中累积：

class StreamingLLMNode: def __init__(self): self.buffer = "" self.block_id = 0 def parse_chunk(self, chunk: str) -> Optional[dict]: self.buffer += chunk # 尝试解析完整JSON块 try: obj = json.loads(self.buffer) # 成功解析，返回带ID的块 result = {"block_id": self.block_id, "data": obj} self.buffer = "" # 清空buffer self.block_id += 1 return result except json.JSONDecodeError: # 未完成，继续累积 return None

只有parse_chunk返回非None时，才更新state并触发图检查。这保证了每次状态更新都是原子性的。

3.5 细节五：Conditional Edges的判定函数，必须用“白名单+正则”双校验

LangGraph的end_conditional_edge常写成：

def route_to_next(state): last_msg = state["messages"][-1] if "get_order_status" in last_msg.content: return "tool_node" elif "human" in last_msg.content.lower(): return "escalate_node" else: return "llm_node"

这种写法极其脆弱。LLM只要说一句"I'll get the order status for you"，就会误入tool_node。正确做法是：

1. 先用正则提取LLM输出中的"action": "xxx"；
2. 再查白名单VALID_ACTIONS = {"get_order_status", "refund_request", "escalate_to_human"}；
3. 白名单外的动作，一律路由到safe_fallback。

import re def route_to_next(state): last_msg = state["messages"][-1].content # 提取action值 match = re.search(r'"action"\s*:\s*"([^"]+)"', last_msg) if not match: return "safe_fallback" action = match.group(1) return action if action in VALID_ACTIONS else "safe_fallback"

这个改动让边缘路由的准确率从79%跃升至99.8%。

3.6 细节六：错误处理必须分“LLM层”和“图层”，不能混为一谈

LLM调用失败（如超时、429）和图逻辑失败（如state缺失字段）必须隔离处理。我设计了两层错误处理器：

• LLM层错误：由llm_node捕获，记录error_type: "llm_timeout"，然后return {"error": "LLM_UNAVAILABLE", "retry_count": state.get("retry_count", 0) + 1}，并路由到retry_node（带指数退避）。
• 图层错误：由graph.checkpointer的on_error钩子捕获，记录error_type: "state_corruption"，触发restore_from_last_checkpoint。

两者日志分开打，告警规则也不同：LLM层错误发企业微信，图层错误直接电话告警。上线三个月，图层错误为0，LLM层错误平均每天0.7次，全部自动恢复。

3.7 细节七：本地模型必须做“Tokenizer对齐”，否则输入会被截断

用Ollama或vLLM部署本地LLM时，常遇到Input length exceeds model max length。这不是模型问题，而是LangGraph传入的messages长度计算方式与模型tokenizer不一致。LangChain的count_tokens用的是tiktoken，而Qwen用transformers.AutoTokenizer，计数差20%-30%。我的补丁是：

from transformers import AutoTokenizer tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2.5-72B") def count_tokens_for_qwen(text: str) -> int: return len(tokenizer.encode(text, add_special_tokens=False)) # 在llm_node中，用此函数检查length，超长则truncate并log warning

这个补丁让本地Qwen2.5的OOM率从12%降到0.1%。

4. 实操过程与核心环节实现：从零搭建一个“多轮订单诊断”LangGraph Agent

现在，我们用一个真实场景——“用户投诉订单未发货，需多轮诊断并自动处理”——来走一遍完整实现。这不是玩具Demo，而是我上个月交付给某跨境电商客户的最小可行版本（MVP），已稳定运行23天。

4.1 步骤一：定义图状态（State）——不是随意加字段，而是按数据血缘设计

LangGraph的状态设计，本质是定义数据在图中流动的“血管”。我拒绝用dict，而是用TypedDict强制类型：

from typing import TypedDict, List, Optional, Dict, Any from langchain_core.messages import BaseMessage class OrderDiagnosisState(TypedDict): messages: List[BaseMessage] # 对话历史，必有 user_id: str # 用户ID，来自初始state，永不变更 order_id: Optional[str] # 订单ID，首次交互后填充 diagnosis_steps: List[str] # 已执行的诊断步骤，如["check_inventory", "verify_payment"] tool_results: Dict[str, Any] # 工具调用结果缓存，键为tool_name escalation_level: int # 升级等级，0=自助，1=人工，2=主管 retry_count: int # LLM调用重试次数，防雪崩

关键点：order_id设为Optional[str]，因为第一轮用户只说“我的订单没发货”，还没提供ID；diagnosis_steps用List而非Set，因为诊断必须有序（先查库存，再验支付，最后看物流）；tool_results用Dict缓存，避免重复调用同一工具。这个state设计，让后续所有节点都能精准知道“当前走到哪一步，缺什么信息”。

4.2 步骤二：构建核心节点——每个节点只做一件事，且有明确退出契约

节点A：extract_order_id_node（信息抽取）

职责：从用户首轮消息中提取order_id。不用LLM，用正则+规则：

import re def extract_order_id_node(state: OrderDiagnosisState) -> dict: user_msg = state["messages"][-1].content # 多模式匹配 patterns = [ r'订单号[:：\s]*([A-Z0-9]{10,15})', r'Order ID[:：\s]*([A-Z0-9]{10,15})', r'#(\d{10,15})' ] for p in patterns: match = re.search(p, user_msg) if match: return {"order_id": match.group(1).upper()} # 未匹配到，触发追问 return {"messages": [AIMessage(content="请提供您的12位订单号，通常在订单确认邮件中。")]} # 边缘路由 def route_after_extract(state: OrderDiagnosisState): return "diagnose_node" if state["order_id"] else "ask_order_id_node"

注意：这里没用LLM，因为正则100%准确，且快。LLM只用在真正需要推理的地方。

节点B：diagnose_node（核心LLM决策节点）

这是全文重点，完整代码：

from langchain_core.output_parsers import JsonOutputParser from langchain_core.pydantic_v1 import BaseModel, Field from langchain_openai import ChatOpenAI from langchain_community.chat_models import ChatOllama class DiagnosisAction(BaseModel): next: str = Field(description="下一个节点名，如 'check_inventory', 'escalate_to_human'") action: Optional[str] = Field(description="要调用的工具名，如 'get_order_status'") params: Optional[dict] = Field(description="工具参数") reason: str = Field(description="选择此路径的理由，不超过20字") parser = JsonOutputParser(pydantic_object=DiagnosisAction) # 根据环境选择LLM if os.getenv("ENV") == "prod": llm = ChatOllama( model="qwen2.5:72b", temperature=0.1, num_predict=512, stop=["<|eot_id|>"] ) else: llm = ChatOpenAI(model="gpt-4o", temperature=0.1) def diagnose_node(state: OrderDiagnosisState) -> dict: # 1. 状态富化 enriched_prompt = f""" 【用户画像】VIP用户，近30天下单5次； 【当前订单】{state['order_id']}； 【已执行步骤】{', '.join(state['diagnosis_steps'])}； 【工具结果】{json.dumps(state['tool_results'], ensure_ascii=False)}； 【可用工具】check_inventory, verify_payment, get_shipping_info, escalate_to_human； 【决策规则】若库存不足且支付成功，升级；若物流无更新超48h，升级；否则自助处理。 """ # 2. 构建消息 messages = normalize_messages(state["messages"]) # 应用3.2节的归一化 # 注入system prompt messages.insert(0, {"role": "system", "content": SYSTEM_PROMPT_DIAGNOSE}) # 注入富化提示 messages[-1]["content"] = enriched_prompt + "\n" + messages[-1]["content"] # 3. 调用LLM（带流式处理） try: response = llm.invoke(messages) # 解析JSON parsed = parser.parse(response.content) # 更新诊断步骤 if parsed.action: new_steps = state["diagnosis_steps"] + [parsed.action] else: new_steps = state["diagnosis_steps"] return { "diagnosis_steps": new_steps, "messages": [AIMessage(content=response.content)], "tool_results": state["tool_results"] # 暂不更新，等tool_node返回 } except Exception as e: # LLM层错误处理 return { "messages": [AIMessage(content="系统繁忙，请稍后再试。")], "retry_count": state.get("retry_count", 0) + 1 } # 边缘路由（应用3.5节的双校验） def route_diagnosis(state: OrderDiagnosisState) -> str: last_msg = state["messages"][-1].content match = re.search(r'"next"\s*:\s*"([^"]+)"', last_msg) if not match: return "safe_fallback" next_node = match.group(1) # 白名单校验 VALID_NODES = {"check_inventory", "verify_payment", "get_shipping_info", "escalate_to_human", "safe_fallback"} return next_node if next_node in VALID_NODES else "safe_fallback"

节点C：check_inventory_node（工具节点）

def check_inventory_node(state: OrderDiagnosisState) -> dict: # 1. 前置校验 if not state["order_id"]: return {"messages": [AIMessage(content="订单ID缺失，无法查询库存。")]} # 2. 调用真实API（此处简化为mock） inventory_status = mock_check_inventory(state["order_id"]) # 返回 {"in_stock": False, "warehouse": "SH"} # 3. 后置断言 assert inventory_status["in_stock"] is False or inventory_status["warehouse"] is not None, "库存API返回异常" # 4. 缓存结果 new_results = state["tool_results"].copy() new_results["check_inventory"] = inventory_status return { "tool_results": new_results, "messages": [AIMessage(content=f"库存检查完成：{inventory_status}")], "diagnosis_steps": state["diagnosis_steps"] + ["check_inventory"] }

4.3 步骤三：组装图（Graph）——用StateGraph而非CompiledGraph

很多人用CompiledGraph，但生产环境我坚持用StateGraph，因为可调试性：

from langgraph.graph import StateGraph, START, END workflow = StateGraph(OrderDiagnosisState) # 添加节点 workflow.add_node("extract_order_id_node", extract_order_id_node) workflow.add_node("ask_order_id_node", lambda s: {"messages": [AIMessage(content="请提供订单号。")]}) workflow.add_node("diagnose_node", diagnose_node) workflow.add_node("check_inventory_node", check_inventory_node) workflow.add_node("verify_payment_node", verify_payment_node) workflow.add_node("get_shipping_info_node", get_shipping_info_node) workflow.add_node("escalate_to_human_node", escalate_to_human_node) workflow.add_node("safe_fallback_node", safe_fallback_node) # 添加边 workflow.add_edge(START, "extract_order_id_node") workflow.add_conditional_edges("extract_order_id_node", route_after_extract) workflow.add_conditional_edges("diagnose_node", route_diagnosis) workflow.add_edge("check_inventory_node", "diagnose_node") workflow.add_edge("verify_payment_node", "diagnose_node") workflow.add_edge("get_shipping_info_node", "diagnose_node") workflow.add_edge("escalate_to_human_node", END) workflow.add_edge("safe_fallback_node", "diagnose_node") # 设置检查点（关键！） from langgraph.checkpoint.sqlite import SqliteSaver checkpointer = SqliteSaver.from_conn_string(":memory:") # 生产用PostgreSQL # 编译 app = workflow.compile(checkpointer=checkpointer) # 测试输入 initial_state = { "messages": [HumanMessage(content="我的订单ABCD12345678还没发货！")], "user_id": "u_98765", "escalation_level": 0, "retry_count": 0, "diagnosis_steps": [], "tool_results": {} } # 执行 for output in app.stream(initial_state, stream_mode="values"): print("State update:", output)

4.4 步骤四：流式响应对齐——让前端看到“思考过程”

用户不关心图怎么跑，只关心“它在想什么”。所以，app.stream()的输出必须映射到前端可消费的事件：

async def stream_graph_response(user_input: str): initial_state = {...} # 同上 async for event in app.astream(initial_state, stream_mode="values"): # 提取AI的流式消息 if "messages" in event and event["messages"]: last_msg = event["messages"][-1] if isinstance(last_msg, AIMessage): # 发送SSE事件 yield f"data: {json.dumps({'type': 'ai_message', 'content': last_msg.content})}\n\n" # 提取工具调用 if "tool_results" in event: for tool_name, result in event["tool_results"].items(): yield f"data: {json.dumps({'type': 'tool_call', 'tool': tool_name, 'result': str(result)})}\n\n" # 最终状态 yield f"data: {json.dumps({'type': 'done', 'final_state': event})}\n\n"

前端用EventSource监听，就能看到：

AI: 正在检查库存... Tool: check_inventory -> {"in_stock": false, "warehouse": "SH"} AI: 库存不足，正在验证支付...

这就是真正的“可解释AI”。

5. 常见问题与排查技巧实录：七类高频故障的根因与秒级定位法

在交付现场，我总结了一套“30秒故障定位法”：看日志、查state、验token。以下是七类问题的速查表，附真实日志片段。

5.1 问题一：图卡死在llm_node，CPU 100%，无日志输出

现象：app.stream()调用后，进程卡住，ps aux | grep python显示高CPU，但tail -f logs/app.log无新日志。
根因：LLM流式响应未关闭，llm.stream()返回的generator未被消费完，导致线程阻塞。
定位：lsof -i :11434（Ollama端口）看连接数，若>100，基本确定。
解决：在llm_node中加超时保护：

import signal def timeout_handler(signum, frame): raise TimeoutError("LLM stream timeout") signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(30) # 30秒超时 try: for chunk in llm.stream(messages): # 处理chunk finally: signal.alarm(0) # 取消定时器

5.2 问题二：conditional_edges随机跳转，同输入多次运行结果不同

现象：输入完全相同，第一次走check_inventory，第二次走escalate_to_human。
根因：LLM的temperature未设为0，导致输出非确定性。
定位：打印LLM的response.content，对比两次输出，看JSON是否一致。
解决：所有生产环境LLM初始化，temperature=0.0，top_p=1.0，禁用采样。

5.3 问题三：tool_results字段为空，但工具明明执行成功了

现象：check_inventory_node日志显示"库存检查完成"，但diagnose_node收到的state["tool_results"]是空字典。
根因：节点返回的dict未包含"tool_results"键，LangGraph不会自动merge。
定位：在diagnose_node入口加print("Received state keys:", list(state.keys()))。
解决：所有节点返回dict，必须显式包含所有要更新的state字段，哪怕值不变。return {"tool_results": new_results}，不能只写{"new_results": ...}。

5.4 问题四：本地Qwen模型报IndexError: index out of range，但API调用正常

现象：用ChatOllama调用本地Qwen，llm.invoke()抛IndexError，但curl http://localhost:11434/api/chat返回正常。
根因：Ollama的num_predict参数与LangChain的max_tokens冲突，Ollama默认num_predict=128，LangChain可能请求更多。
定位：ollama show qwen2.5:72b --modelfile看默认配置。
解决：初始化ChatOllama时，显式设num_predict=512，并与llm.invoke(..., max_tokens=512)保持一致。

5.5 问题五：messages里出现SystemMessage，导致Claude返回{"error": "Invalid role"}

现象：用Claude时，llm.invoke()返回400错误，message为Invalid role 'system'。
根因：Claude不支持system角色，但LangChain的SystemMessage被直接传入。
定位：print([m.type for m in state["messages"]])，看是否有system。
解决：应用3.2节的normalize_messages，把SystemMessage内容合并到首条user消息。

5.6 问题六：图执行后，state里messages长度暴增，内存OOM

现象：运行几轮后，state["messages"]有200+条，进程被OOM Killer干掉。
根因：messages未做截断，LangGraph默认全量保存。
定位：print(len(state["messages"]))在每个节点入口。
解决：在llm_node中加截断逻辑：

def truncate_messages(messages: List[BaseMessage], max_len: int = 20) -> List[BaseMessage]: if len(messages) <= max_len: return messages # 保留system（如果有）、最近max_len-1条，加一条摘要 recent = messages[-(max_len-1):] summary = f"【对话摘要】已进行{len(messages)-len(recent)}轮交互，用户核心诉求：{messages[0].content[:30]}..." return [SystemMessage(content=summary)] + recent

5.7 问题七：SqliteSaver报Database is locked，并发请求失败

现象：压测时，多用户同时请求，app.stream()抛OperationalError: database is locked。
根因：SQLite不支持高并发写，checkpointer的save操作阻塞。
定位：strace -p $(pgrep -f "sqlite")看系统调用。
解决：生产环境必须换PostgresSaver，或用AsyncPostgresSaver。SQLite只用于开发。

我的个人经验是：LangGraph的调试，80%时间花在“看state长什么样”。所以，我在每个节点入口加了一行logger.debug(f"Node X state keys: {list(state.keys())}, messages_len: {len(state.get('messages', []))}")。这行日志，救了我三次通宵。

6. 工具链与环境配置：一份可直接pip install的生产级依赖清单

别再用pip install langgraph了，那个包太旧。以下是我在Kubernetes集群里跑的精确版本组合，经过23天压力测试（QPS 127，P99延迟<850ms）：

# requirements.txt langchain-core==0.3.12 langchain-community==0.3.12 langchain-openai==0.2.12 langgraph==0.2.52 langchain-text-splitters==0.3.3 pydantic==2.9.2 sqlalchemy==2.0.34 psycopg2-binary==2.9.9 # Postgres saver必需 tiktoken==0.7.0 transformers==4.44.2 accelerate==0.33.0 bitsandbytes==0.43.3 # 量化必需 ollama==0.3.4 openai==1.47.0

关键配置项（.env）：

# LLM配置 LLM_PROVIDER=ollama OLLAMA_MODEL=qwen2.5:72b OLLAMA_BASE_URL=http://ollama-service:11434 # 图配置 LANGGRAPH_CHECKPOINTER=postgres POSTGRES_URL=postgresql://user:pass@postgres:5432/langgraph_db # 安全配置 LANGCHAIN_TRACING_V2=true LANGCHAIN_PROJECT=order-diagnosis-prod LANGCHAIN_ENDPOINT=https://api.smith.langchain.com

Dockerfile精简版：

FROM python:3.11-slim #