企业官网建设流程全解析

1. 项目概述：为什么“从零搭建智能体”这件事，现在比任何时候都更值得你亲手做一遍

LangGraph 这个词最近在技术社区里出现的频率，已经快赶上当年 Docker 刚火起来时的“容器化”了。但和 Docker 不同的是，LangGraph 不是帮你打包应用的工具，而是帮你把大模型真正“用活”的操作系统——它让 LLM 不再是那个只能聊天、写诗、续写故事的“高级文本生成器”，而是一个能记住上下文、能调用工具、能自主决策、能反复试错、甚至能自我反思的可编程智能体（Programmable Agent）。我从去年底开始在三个不同业务线落地 LangGraph，从客服工单自动分派系统，到内部知识库的多跳问答引擎，再到合规文档的自动化审查流水线，最深的体会是：真正卡住团队落地进度的，从来不是模型能力，而是缺乏一套清晰、可控、可调试、可追踪的 Agent 执行框架。LangGraph 的 StateGraph 就是为解决这个问题而生的。它不像传统 workflow 引擎那样抽象成“节点+连线”，而是把 Agent 的每一次思考、每一次工具调用、每一次状态变更，都映射成一个明确的 Python 函数和一个可序列化的状态字典。这意味着你不需要去猜模型“到底在想什么”，而是可以像调试一个 Flask 路由一样，打断点、看变量、改逻辑、重放执行流。标题里说的“从零搭建”，指的不是从零训练模型，而是从零开始构建一个具备记忆、规划、工具调用、错误恢复能力的最小可行 Agent 系统。它适合三类人：刚学完 LangChain 想进阶的开发者、正在评估 Agent 框架选型的技术负责人、以及需要快速验证某个业务场景是否适合 Agent 化的产品同学。这篇文章不讲概念堆砌，不画虚线框图，只讲我在真实项目里敲出来的每一行关键代码、踩过的每一个坑、以及为什么非得这么写不可。

2. 核心设计思路拆解：StateGraph 不是“另一个 workflow”，它是 Agent 的状态机操作系统

2.1 为什么不用现成的 workflow 框架？比如 Airflow、Prefect 或者 Camunda？

这是我在第一个项目评审会上被问最多的问题。答案很直接：Airflow 是为批处理任务设计的，Prefect 是为数据管道设计的，Camunda 是为企业级 BPM 设计的——它们全都不理解“LLM 的不确定性”这个核心前提。举个例子：你在 Airflow 里定义一个 task，它要么成功，要么失败，失败就重试或告警。但 Agent 的一次“工具调用”可能返回格式错误的 JSON、可能超时、可能返回完全无关的信息、甚至可能因为模型幻觉而给出一个看似合理实则危险的结论。这时候，你不能简单地“重试”，而需要判断：“是参数错了？还是工具本身不可用？还是模型理解偏了？要不要换一个工具？要不要把上一步的结果再喂给模型让它重新规划？”——这种动态决策链，是任何静态 workflow 引擎都无法表达的。LangGraph 的 StateGraph 本质是一个带状态的有向图（Stateful Directed Graph），它的每个节点（node）是一个纯函数，接收当前 state，返回更新后的 state；每条边（edge）是一个条件函数，接收当前 state，返回下一个节点的名字。这个设计的精妙之处在于：state 是唯一的真相源（single source of truth），所有节点都只读写这个 state，没有隐式共享变量，没有跨节点的副作用。我在做客服工单系统时，state 里存着{"ticket_id": "T-2024-0876", "current_status": "awaiting_analysis", "analysis_history": [...], "available_tools": ["search_knowledge_base", "query_customer_db"]}。当analyze_ticket节点运行后，它不改变全局变量，只是返回一个新的 dict，比如{"current_status": "needs_customer_data", "customer_id": "C-98765"}。下一个节点fetch_customer_data就能精准地拿到这个customer_id去查库。这种设计让整个流程变得极其透明，也极其容易测试——你完全可以把 state 当作一个 fixture，传给任意节点函数，看它输出什么，完全脱离整个图的上下文。

2.2 StateGraph 和 LangChain 的 RunnableSequence / RunnableParallel 有什么本质区别？

RunnableSequence 看起来很像一个 workflow：chain = prompt | llm | parser。但它是一个线性、无状态、无分支的管道。一旦定义好，输入进来，就按固定顺序走完，中间无法根据 LLM 的输出内容决定下一步走哪条路。而 StateGraph 的核心价值，恰恰在于它的条件驱动（conditional routing）。比如，在一个电商客服 Agent 里，用户说“我的订单还没发货，我要投诉”。LLM 的plan节点可能输出：{"action": "check_order_status", "order_id": "E-12345"}。这时，边的条件函数会检查state["action"] == "check_order_status"，然后路由到check_order_status节点。如果 LLM 输出的是{"action": "escalate_to_manager", "reason": "delivery_delay_over_7_days"}，边的条件函数就会路由到escalate_to_manager节点。这种基于 LLM 动态输出内容来决定流程走向的能力，是 RunnableSequence 根本不具备的。你可以把它理解为：RunnableSequence 是“预设剧本”，StateGraph 是“即兴演出”，而导演（也就是你的条件函数）随时盯着演员（LLM）的台词，决定下一幕该谁上场。这也是为什么 LangGraph 的官方示例里，几乎所有的add_conditional_edges都是围绕state["next"]或state["decision"]这样的 key 来写的——它把控制权，从硬编码的流程，交还给了模型的推理结果。

2.3 为什么强调“从零”？为什么不直接用 LangGraph 官方的create_react_agent或create_openai_functions_agent？

官方提供的create_*_agent是极好的学习起点，但它们是高度封装的“黑盒”。它们内部已经帮你定义好了标准的 ReAct 模板、工具调用的 JSON Schema、错误重试逻辑、甚至默认的max_iterations。这就像给你一辆已经调校好的赛车，你可以上手就开，但如果你想知道为什么过弯时要收油、为什么进站要换软胎、为什么引擎在 8000 转时会断油——你就必须自己动手组装一台发动机。在我们第二个项目（内部知识库问答）中，官方 Agent 在处理“对比 A 和 B 两个方案的优缺点，并结合我们公司去年的营收数据给出建议”这类复杂多跳问题时，经常在第三步就陷入死循环，因为它默认的max_iterations=10太小，而重试逻辑又过于简单（只是原样重发）。我们最终不得不拆开create_react_agent的源码，发现它底层用的StateGraph其实只有 4 个节点：call_model,tool_node,should_continue,end。于是我们自己 fork 了一份，把should_continue改成了一个能分析state["tool_calls"]历史、判断是否已获取足够信息、并能主动触发summarize_findings节点的智能函数。这个过程，就是“从零搭建”的真正含义：不是重复造轮子，而是亲手拧紧每一颗螺丝，确保它完全符合你业务场景的扭矩要求。

3. 核心细节解析与实操要点：State、Node、Edge 三要素的实战定义法则

3.1 State 的设计哲学：它不是“数据容器”，而是 Agent 的“记忆与意图声明书”

很多初学者一上来就想把所有东西都塞进 state：用户原始输入、LLM 的所有中间输出、工具返回的原始 JSON、甚至日志时间戳。这是个巨大的误区。State 的设计，必须遵循“最小完备性”原则：它只应该包含两类信息——Agent 当前的明确意图（intent）和完成该意图所必需的、最新的、结构化的上下文（context）。我给自己定了一条铁律：State 里的每一个 key，都必须能在下一轮节点执行中，被至少一个节点的函数签名直接消费。举个反例：如果你在 state 里存了raw_user_input: "帮我查一下订单 E-12345 的状态"，但没有任何一个节点的函数签名里有raw_user_input: str这个参数，那这个字段就是冗余的，它只会增加序列化开销、拖慢调试速度、并在出错时制造干扰信息。正确的做法是，在第一个parse_user_query节点里，就把这句话解析成结构化的 intent，比如{"intent": "check_order_status", "order_id": "E-12345", "required_info": ["shipping_date", "current_location"]}。后续所有节点，都只跟这个结构化的 intent 打交道。我在鼎新 workflow 项目里看到过一个典型反面案例：他们的 state 里有一个debug_info字段，里面塞满了各种中间变量的字符串 repr。结果当 state 变得巨大时，LangGraph 的get_state_historyAPI 响应时间从 200ms 暴涨到 3s，根本没法做实时调试。后来我们把它重构为：只在 state 里存一个debug_mode: boolflag，真正的 debug 信息，由一个独立的log_debug_info节点，通过print()或写入 Redis 的方式异步记录。State 保持轻量，才是高性能、可扩展 Agent 的基石。

3.2 Node 的编写规范：纯函数 + 明确副作用隔离，是可测试性的唯一保障

一个合格的 LangGraph node，必须是一个纯函数（pure function）：给定相同的 state 输入，它必须总是返回相同的新 state 输出，且不产生任何外部副作用（比如不直接修改数据库、不发 HTTP 请求、不写文件）。所有“真实世界”的交互，都必须封装在单独的、可 mock 的工具函数里。这是为了保证 node 的可测试性（testability）和可重放性（replayability）。我习惯把 node 分成两类：orchestration nodes（编排节点）和tool-invocation nodes（工具调用节点）。Orchestration nodes 只做三件事：解析 state、决定调用哪个工具、构造工具调用参数。Tool-invocation nodes 则只做一件事：安全地调用那个工具函数，并把结果结构化地塞回 state。来看一个真实的check_order_statusorchestration node：

def check_order_status(state: State) -> dict: """Orchestration node: decides to call the order status tool and prepares its args.""" # 1. 从 state 中提取必要信息 order_id = state.get("order_id") if not order_id: raise ValueError("order_id is missing from state") # 2. 构造工具调用参数，注意：这里不调用工具！ return { "tool_name": "get_order_status", "tool_args": {"order_id": order_id}, "next": "call_tool" # 显式声明下一步 }

而对应的 tool-invocation node 是：

def call_tool(state: State) -> dict: """Tool-invocation node: safely calls the external tool.""" tool_name = state.get("tool_name") tool_args = state.get("tool_args", {}) # 3. 这里才真正调用工具，且做了完整的异常捕获和降级 try: result = TOOLS[tool_name](**tool_args) return {"tool_result": result, "next": "process_tool_result"} except ToolNotFoundError: return {"error": f"Tool {tool_name} not found", "next": "handle_tool_error"} except Exception as e: logger.error(f"Tool {tool_name} failed: {e}") return {"error": str(e), "next": "handle_tool_error"}

这种分离，让我可以在单元测试里，轻松地 mockTOOLS["get_order_status"]，然后只测试check_order_status的逻辑是否正确地构造了参数，或者只测试call_tool是否正确地处理了各种异常。如果把这两步混在一个函数里，测试就变成了集成测试，速度慢、不稳定、难以定位问题。

3.3 Edge 的条件函数：别写if/else，用“状态谓词”表达业务逻辑

add_conditional_edges的第二个参数，是一个函数，它接收 state，返回下一个节点的名字。新手常犯的错误是，在这个函数里写一堆if state["x"] == "y": return "node_a" elif ...。这会让条件逻辑变得臃肿、难以维护、且无法复用。更好的做法是，把每个业务规则，抽象成一个独立的、命名清晰的状态谓词（state predicate）函数。这些函数只做一件事：回答一个布尔问题。比如：

def should_call_tool(state: State) -> bool: """Predicate: Is there a pending tool call ready to be executed?""" return "tool_name" in state and "tool_args" in state def should_process_result(state: State) -> bool: """Predicate: Did the last tool call succeed and return useful data?""" return "tool_result" in state and not state.get("error") def should_handle_error(state: State) -> bool: """Predicate: Did the last step encounter an error that needs handling?""" return "error" in state

然后，在add_conditional_edges里，用一个清晰的字典来映射：

graph.add_conditional_edges( "check_order_status", lambda s: ( "call_tool" if should_call_tool(s) else "handle_missing_order_id" if not s.get("order_id") else "end" ), { "call_tool": "call_tool", "handle_missing_order_id": "handle_missing_order_id", "end": END, } )

提示：把谓词函数单独抽出来，最大的好处是它们可以被单元测试覆盖。你可以写assert should_call_tool({"tool_name": "x", "tool_args": {}}) == True，这比测试一个嵌套了 5 层if/else的 lambda 表达式要可靠得多。而且，当业务规则变化时，你只需要修改谓词函数，而不用动图的拓扑结构。

4. 实操过程与核心环节实现：从pip install到第一个可交互 Agent 的完整路径

4.1 环境准备与依赖管理：为什么推荐uv而不是pip或conda？

LangGraph 的依赖树相当复杂，它底层依赖langchain-core,langchain-community,langsmith,pydantic,httpx,tenacity等十几个包，其中很多包对 Python 版本、typing模块的兼容性要求极其苛刻。我试过用pip install langgraph，结果在 Python 3.11 环境下，langchain-core的BaseModel和pydanticv2 的RootModel冲突，导致State类初始化就报错。也试过conda install -c conda-forge langgraph，但 conda-forge 的包版本滞后，缺少StateGraph的最新interrupt_before特性。最终，我们团队统一迁移到了uv—— 一个由 Rust 编写的、号称“比 pip 快 100 倍”的新一代 Python 包管理器。它的优势在于：精确的依赖解析（exact dependency resolution）和隔离的虚拟环境创建（isolated venv creation）。uv不会像pip那样，因为一个包的setup.py里写了"requests>=2.25.0"就去安装最新的requests 2.32.0，而是会根据pyproject.toml里锁死的requirements.txt，安装那个经过充分测试的、确定能工作的版本。我们的标准初始化命令是：

# 1. 创建一个干净、隔离的 uv venv uv venv .venv --python 3.11 # 2. 激活它（Linux/Mac） source .venv/bin/activate # 3. 使用 uv 的 lock 文件进行精确安装（我们维护了一个 langgraph-lock.txt） uv pip install -r langgraph-lock.txt # 4. 验证核心组件可用 python -c "from langgraph.graph import StateGraph; print('✅ StateGraph imported')" python -c "from langgraph.checkpoint.memory import MemorySaver; print('✅ MemorySaver imported')"

langgraph-lock.txt的内容，是我们团队在 CI/CD 流水线里，用uv pip compile pyproject.toml --output-file langgraph-lock.txt生成的，它包含了所有依赖及其精确的哈希值。这样，无论是在开发机、测试服务器还是生产 K8s Pod 里，uv pip install -r langgraph-lock.txt安装出来的环境，都是 100% 一致的。这避免了“在我机器上是好的”这类经典问题。

4.2 定义 State：用 Pydantic v2 的BaseModel构建强类型契约

LangGraph 的State并不是一个魔法对象，它就是一个普通的 Pythondict。但为了获得 IDE 的自动补全、运行时的类型检查、以及清晰的文档，我强烈建议用 Pydantic v2 的BaseModel来定义它。这不是可选项，而是必选项。下面是我们电商客服 Agent 的State定义：

from typing import List, Optional, Dict, Any from pydantic import BaseModel, Field class TicketState(BaseModel): """The single source of truth for the agent's current state.""" # --- Core Intent & Context --- ticket_id: str = Field(..., description="Unique identifier for the support ticket") user_query: str = Field(..., description="The original, unmodified user question") intent: str = Field(..., description="The parsed high-level intent, e.g., 'check_order_status'") # --- Execution Flow Control --- next: str = Field(default="start", description="The name of the next node to execute") iteration_count: int = Field(default=0, description="How many times has the main loop run?") max_iterations: int = Field(default=15, description="Hard limit to prevent infinite loops") # --- Tool Interaction Data --- tool_name: Optional[str] = Field(default=None, description="Name of the tool to be called next") tool_args: Optional[Dict[str, Any]] = Field(default_factory=dict, description="Arguments for the next tool call") tool_result: Optional[Dict[str, Any]] = Field(default=None, description="The raw result from the last tool call") error: Optional[str] = Field(default=None, description="Error message from the last failed step") # --- Structured Output & Final Answer --- final_answer: Optional[str] = Field(default=None, description="The final, polished answer to send to the user") analysis_summary: Optional[str] = Field(default=None, description="A concise summary of findings for human review") # --- Debug & Audit Trail --- debug_mode: bool = Field(default=False, description="Enable verbose logging for this execution") history: List[Dict[str, Any]] = Field(default_factory=list, description="A log of all state transitions for debugging") def model_dump(self, *args, **kwargs) -> dict: """Override to ensure we only serialize the fields we want.""" # We exclude internal fields like 'history' from being serialized to checkpoint storage # to keep the checkpoint size small. exclude_fields = {"history"} return super().model_dump(*args, **kwargs, exclude=exclude_fields)

注意：model_dump方法的重写是关键。history字段虽然对调试极其有用，但它会随着每次 state 更新而不断增长，如果不排除，checkpoint 的大小会指数级膨胀。我们只在内存中保留它，用于print(state.history)，而序列化到 Redis 或 SQLite 时，自动过滤掉。

4.3 构建第一个 StateGraph：一个能自我修复的“订单状态查询”Agent

现在，我们把前面定义的TicketState、check_order_status、call_tool等节点，组装成一个完整的StateGraph。这个 Agent 的目标很简单：用户输入一个订单号，它能查出状态，并在查不到时，主动提示用户检查订单号格式。整个图只有 5 个节点，但已经具备了 Agent 的核心能力：规划、工具调用、错误处理、自我修复。

from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver # 1. 初始化图，指定 State 类型 workflow = StateGraph(TicketState) # 2. 添加节点（注意：节点函数名就是节点名） workflow.add_node("start", start_node) # 解析用户输入，设置初始 intent workflow.add_node("check_order_status", check_order_status) workflow.add_node("call_tool", call_tool) workflow.add_node("process_tool_result", process_tool_result) workflow.add_node("handle_error", handle_error) # 3. 添加边：START -> start workflow.add_edge(START, "start") # 4. 添加条件边：从 start 节点出发，根据 intent 决定下一步 workflow.add_conditional_edges( "start", lambda s: s.intent, { "check_order_status": "check_order_status", "unknown": "handle_error", # 如果解析失败，直接进错误处理 } ) # 5. 添加条件边：从 check_order_status 出发，决定是调用工具还是报错 workflow.add_conditional_edges( "check_order_status", lambda s: ( "call_tool" if "tool_name" in s and s.tool_name == "get_order_status" else "handle_error" if not s.get("order_id") else "end" ), { "call_tool": "call_tool", "handle_error": "handle_error", "end": END, } ) # 6. 添加普通边：call_tool 总是走到 process_tool_result workflow.add_edge("call_tool", "process_tool_result") # 7. 添加条件边：process_tool_result 根据结果决定是结束还是继续 workflow.add_conditional_edges( "process_tool_result", lambda s: "final_answer" in s and s.final_answer is not None, { True: END, False: "handle_error", # 如果没生成答案，说明处理失败 } ) # 8. 添加普通边：handle_error 总是走到 end（也可以设计成重试） workflow.add_edge("handle_error", END) # 9. 设置内存检查点，用于保存中间状态（支持中断和恢复） memory = MemorySaver() app = workflow.compile(checkpointer=memory)

4.4 启动并交互：用app.invoke()运行你的第一个 Agent

编译完成后，app就是一个可调用的对象。你可以用app.invoke()来运行一次完整的流程。下面是一个完整的、可直接复制粘贴运行的交互示例：

# 模拟一次用户提问 initial_state = TicketState( ticket_id="T-2024-001", user_query="我的订单 E-12345 还没发货，请查一下状态。", intent="check_order_status", order_id="E-12345", debug_mode=True, ) # 运行 Agent！ result = app.invoke( initial_state, config={"configurable": {"thread_id": "test-thread-001"}}, # 必须提供 thread_id 用于 checkpoint ) print("=== 最终结果 ===") print(f"最终答案: {result.final_answer}") print(f"分析摘要: {result.analysis_summary}") print(f"执行历史: {len(result.history)} 步") # 查看详细的执行轨迹（这就是 StateGraph 的魔力） print("\n=== 详细执行轨迹 ===") for i, step in enumerate(result.history): print(f"Step {i}: {step['node']} -> {step.get('next', 'END')}") if "tool_name" in step: print(f" 工具调用: {step['tool_name']}({step['tool_args']})") if "tool_result" in step: print(f" 工具结果: {step['tool_result'].get('status', 'N/A')}")

当你第一次运行这段代码时，你会看到控制台打印出类似这样的输出：

Step 0: start -> check_order_status Step 1: check_order_status -> call_tool Step 2: call_tool -> process_tool_result Step 3: process_tool_result -> END 工具调用: get_order_status({'order_id': 'E-12345'}) 工具结果: shipped === 最终结果 === 最终答案: 您的订单 E-12345 已于 2024-05-20 发货，预计 5 月 25 日送达。 分析摘要: 订单状态为 'shipped'，物流单号为 'SF123456789CN'。

实操心得：app.invoke()的config参数里的thread_id是强制的。这是因为 LangGraph 的检查点（checkpoint）机制，是按thread_id来区分不同会话的。如果你不提供，它会抛出ValueError: Missing configurable: thread_id。你可以把它理解为 Web 应用里的 session ID。在生产环境中，这个thread_id通常来自用户的 JWT token 或数据库里的会话 ID。

5. 常见问题与排查技巧实录：那些官方文档里不会写的“血泪教训”

5.1 问题速查表：高频报错与根因分析

5.2 “中断与恢复”功能的正确打开方式：不是所有场景都适合interrupt_before

LangGraph 的interrupt_before是一个强大特性，它允许你在某个节点执行前，暂停整个流程，把控制权交还给用户（比如，让用户确认一个高风险操作）。但很多教程把它用错了。最常见的错误是：workflow.add_node("approve_payment", approve_payment); workflow.add_edge("calculate_total", "approve_payment"); workflow.add_edge("approve_payment", "execute_payment")，然后设置interrupt_before="approve_payment"。这看起来很合理，但问题在于：approve_payment是一个 node，而interrupt_before的作用对象是node 的执行入口。这意味着，当流程走到approve_payment节点时，它会暂停，等待你调用app.update_state(...)来提供一个state，然后才真正执行approve_payment函数。但approve_payment函数本身，很可能需要state["payment_amount"]和state["user_approval"]这两个字段。而update_state时，你只提供了{"user_approval": "yes"}，却漏掉了payment_amount，导致approve_payment执行时报错。正确的做法是：interrupt_before的节点，应该是一个纯粹的“决策点”，它本身不消费任何 state，只负责把当前 state 的关键信息，以友好的方式呈现给用户，并等待用户输入一个简单的确认信号。我们把它叫做human_in_the_loop节点：

def human_in_the_loop(state: TicketState) -> dict: """A pure decision point. It does NOT consume any state, only presents it.""" # 这里不读取 state 的任何字段！只把 state 本身作为上下文传递出去 return {"next": "await_human_approval"} # 然后，我们设置中断点在这个节点之前 workflow.add_node("human_in_the_loop", human_in_the_loop) workflow.add_edge("calculate_total", "human_in_the_loop") workflow.add_edge("human_in_the_loop", "execute_payment") workflow.add_edge("await_human_approval", "execute_payment") # 这个节点是用户手动触发的 # 关键：中断点设在 'human_in_the_loop' 之前，而不是之后 workflow.add_edge(START, "human_in_the_loop") # 错误！这会导致一启动就中断 # 正确： workflow.add_conditional_edges( "calculate_total", lambda s: "await_human_approval" if s.payment_amount > 1000 else "execute_payment", { "await_human_approval": "human_in_the_loop", "execute_payment": "execute_payment", } )

5.3 性能瓶颈排查：当你的 Agent 从 2 秒变 20 秒，问题一定出在这里

在我们第三个项目（交通预测 LLM）中，Agent 的平均响应时间从上线初期的 1.8 秒，缓慢爬升到了 18 秒。监控显示，CPU 和内存都很健康，网络延迟也正常。最终，我们用cProfile对app.invoke()进行了性能剖析，发现 92% 的时间，花在了langgraph.checkpoint.memory.MemorySaver.put()这个方法上。原因很隐蔽：我们在TicketState的history字段里，不仅存了{"node": "x", "next": "y"}，还存了{"tool_result": {"huge_nested_dict": {...}}}。MemorySaver默认使用pickle序列化整个 state，而pickle对大型嵌套字典的序列化效率极低。解决方案有两个：

1. 最推荐：禁用history的序列化。正如前面model_dump()方法所示，我们在序列化时，显式地exclude={"history"}。history只存在于内存中，用于print()调试，不进入持久化层。

2. 次选：更换检查点后端。MemorySaver只适合开发和测试。生产环境，必须换成SqliteSaver或PostgresSaver。它们会把 state 拆分成多个字段存储，tool_result这种大字段，可以单独存为TEXT类型，用json.dumps()存储，效率远高于pickle。

from langgraph.checkpoint.sqlite import SqliteSaver # 使用 SQLite 作为检查点后端 saver = SqliteSaver.from_conn_string("./checkpoints.db") app = workflow.compile(checkpointer=saver)

实操心得：永远不要在state里存二进制数据（如图片 base64）、大段日志文本、或未经压缩的 JSON。如果业务确实需要，把它们存到对象存储（如 S3）或数据库里，然后在state里只存一个s3_key或db_id。State 的设计哲学，永远是“最小完备”，而不是“最大信息”。

6. 从“能跑”到“能用”：生产环境部署的四个关键加固点

6.1 输入清洗与防注入：LLM 是你的大脑，不是你的防火墙

很多人以为，只要用了 LangGraph，就天然具备了安全性。这是致命的误解。LangGraph 本身不提供任何输入过滤。如果用户输入"; rm -rf /"，而你的parse_user_query节点又恰好用eval()去解析，那后果不堪设想。我们必须在start节点之前，就建立一道坚固的输入清洗网关。我们采用的是“三层过滤”策略：

1. 字符级白名单：只允许 ASCII 字母、数字、常见标点（. , ! ? ; : ' " ( ) [ ] { }）和中文字符。所有其他字符（如\x00,<script>,--）一律替换为空格。
2. 长度限制：单次输入严格限制在 2000 字符以内。超过部分截断，并在state中标记input_truncated: True，供后续节点生成警告。
3. LLM 辅助检测：用一个轻量级的、本地部署的tinyllm模型，对清洗后的输入进行快速分类：“正常咨询”、“恶意指令”、“垃圾信息”。只有分类为“正常咨询”的，才进入主StateGraph流程。

def sanitize_input(user_input: str) -> str: # 1. 白名单