AI AGENT ENGINEERING

LangGraph：让大模型从“会回答”走向“会执行”

用状态、节点与边，构建可恢复、可观测、可干预的 AI 工作流

核心观点： 大模型负责理解与推理，工具负责行动，LangGraph 负责把状态、步骤、分支和恢复机制组织成一个可长期运行的系统。

聊天模型擅长生成答案，但真正的业务系统往往要求它连续完成多步任务：读取上下文、选择工具、检查结果、根据反馈调整路径，并在故障后继续运行。问题的难点因此不再只是提示词写得好不好，而是如何让整个执行过程可靠、透明且可控制。

LangGraph 的价值正在这里显现。它不替代模型，也不限制开发者使用哪一种模型或工具，而是提供一套面向有状态工作流的编排方式。开发者可以像画流程图一样组织智能应用，同时保留大模型在关键节点上的判断能力。

为什么单次模型调用不够

一次“输入提示词、得到结果”的调用很适合问答，却很难直接支撑长时间、多步骤的任务。当系统开始承担客服、数据分析、研发辅助或旅行规划等职责时，四类工程问题会迅速暴露出来。

• 状态容易丢失：任务一旦跨越多轮对话或多个处理步骤，模型必须知道已经完成了什么、当前处于哪里、下一步需要什么。
• 决策难以追踪：如果所有逻辑都藏在一次长提示中，开发者很难判断模型为什么选择某个工具，错误又发生在哪一步。
• 流程无法干预：高风险操作通常需要人工确认；完全自动化的黑盒流程既不安全，也不符合真实业务的审批习惯。
• 长任务缺少恢复能力：网络波动、模型限流、工具超时都可能中断执行。没有检查点和持久化，系统只能从头再来。

可以把一个智能服务想象成团队：大模型是负责判断的成员，搜索、数据库和业务 API 是工具，LangGraph 则同时承担项目经理和流程系统的角色。它记录现场、安排下一步、处理分支，并为持久化、人工介入、调试和部署提供结构基础。

工作流与智能体：先划清控制边界

“工作流”和“智能体”经常被混为一谈，但两者最重要的区别在于流程由谁决定。工作流的路径由代码预先定义，强调可预测性与一致性；智能体把更多选择权交给大模型，让模型根据环境和中间结果动态决定下一步。

工作流适合报销审批、工单处理、内容流水线等步骤清晰的任务。它像一条经过设计的生产线，每个环节都有明确输入和输出。

智能体适合开放式研究、复杂故障排查、跨工具规划等无法提前枚举全部路径的任务。它可以观察结果、重新规划，并多次调用工具。

工程选择： 生产系统通常不是二选一。更稳妥的做法是让代码控制主干流程，只在确实需要判断的节点交给大模型决策，形成“确定性骨架 + 智能分支”的混合结构。

用图表达复杂流程

LangGraph 使用图来描述执行过程。图并不神秘：节点表示处理步骤，边表示步骤之间的连接，状态则是在整个流程中流动和更新的数据。快递网络是理解这套模型的好类比。

State：贯穿全程的共享状态

State 就像包裹信息卡，记录单号、起点、目的地、配送状态和流转历史。每个节点都可以读取它，并返回自己负责的那部分更新。状态应当结构化定义，因为清晰的数据契约能让节点职责、类型检查和调试边界都更明确。

Node：只做一件事的处理单元

节点本质上是函数。它接收当前状态，执行一次明确操作，然后返回状态更新。节点之间不直接共享局部变量，而是通过 State 协作。把“理解需求、调用模型、执行搜索、评估结果、生成答案”拆成不同节点，可以让每一步独立测试，也让错误定位变得直接。

Edge：把步骤连接成可执行路径

固定边表示无条件从 A 进入 B；条件边会读取状态，根据规则或模型判断选择不同分支。START 和 END 是特殊节点，分别标记流程入口与出口。只要把节点、边和状态组合起来，一个业务流程就拥有了清晰的执行拓扑。

START -> receive -> sort -> route
                         |-- priority --> express -> deliver -> END
                         `-- standard --> ground  -> deliver -> END

状态更新不是简单赋值

一个状态字段可能被多个节点连续更新。默认覆盖并不总是正确，例如对话消息、处理历史和累计里程更适合追加或累加。LangGraph 通过 reducer 定义新值与旧值如何合并。

from operator import add
from typing import Annotated
from typing_extensions import TypedDict

class PackageState(TypedDict):
    package_id: str
    destination: str
    priority: str
    status: str
    history: Annotated[list[str], add]
    total_distance: Annotated[int, add]

这里的 history 会把各节点返回的列表依次拼接，total_distance 会累计每一段里程；而未声明 reducer 的字段通常按新值覆盖旧值。把合并规则写进类型定义，比让每个节点手动读取、修改再写回更一致，也更适合并行或循环流程。

从零搭起一个可执行图

构建 Graph API 的顺序很稳定：先定义状态，再实现节点，然后添加固定边或条件边，最后编译图。这里的“编译”不是把 Python 翻译成机器码，而是在运行时组装图结构并执行基本校验，例如检查节点连接是否有效。

from langgraph.graph import StateGraph, START, END

def receive(state: PackageState):
    return {"status": "已揽收", "history": ["完成揽收"]}

def sort_package(state: PackageState):
    return {"status": "已分拣", "history": ["完成分拣"]}

def deliver(state: PackageState):
    return {
        "status": "已签收",
        "history": [f"送达 {state['destination']}"]
    }

builder = StateGraph(PackageState)
builder.add_node("receive", receive)
builder.add_node("sort", sort_package)
builder.add_node("deliver", deliver)

builder.add_edge(START, "receive")
builder.add_edge("receive", "sort")
builder.add_edge("sort", "deliver")
builder.add_edge("deliver", END)

graph = builder.compile()
result = graph.invoke({
    "package_id": "P001",
    "destination": "上海",
    "priority": "普通",
    "status": "待揽收",
    "history": [],
    "total_distance": 0,
})

固定流程可以直接使用 add_edge；需要动态选择时，则通过 add_conditional_edges 绑定一个路由函数。路由函数读取状态并返回目标分支的标识，图据此选择下一节点。开发者仍然掌握允许出现的路径，大模型或业务规则只负责在这些路径中做选择。

执行时，invoke 适合一次性拿到最终状态，stream 则会逐步产出节点更新，便于把中间进度实时推送给前端。对于耗时较长的代理任务，流式输出不仅改善体验，也能帮助开发者观察图到底运行到了哪里。

让模型学会使用搜索工具

一旦把搜索、数据库查询或外部 API 作为工具绑定给模型，工作流就从固定处理链升级为会行动的智能代理。模型节点负责理解问题并生成工具调用，工具节点负责执行调用并返回 ToolMessage，随后模型根据工具结果继续推理。

from langchain.chat_models import init_chat_model
from langchain_tavily import TavilySearch
from langgraph.graph import StateGraph, START
from langgraph.graph import MessagesState
from langgraph.prebuilt import ToolNode, tools_condition

search = TavilySearch(max_results=4)
model = init_chat_model("gpt-4o-mini").bind_tools([search])

def call_model(state: MessagesState):
    reply = model.invoke(state["messages"])
    return {"messages": [reply]}

builder = StateGraph(MessagesState)
builder.add_node("model", call_model)
builder.add_node("tools", ToolNode([search]))
builder.add_edge(START, "model")
builder.add_conditional_edges("model", tools_condition)
builder.add_edge("tools", "model")
agent = builder.compile()

这段图形成一个闭环：模型若判断不需要搜索，流程直接结束；若生成 tool_calls，则进入工具节点；工具结果回到模型节点，模型重新判断，直到得到最终答案。工具消息必须携带与调用对应的 tool_call_id，否则模型无法把执行结果与之前的请求正确配对。

设计提醒： 工具是否可调用、参数如何校验、最多循环多少次，都应由系统约束。不要把所有控制权都交给模型；边界越清晰，代理越可靠。

从线性 RAG 到代理式 RAG

传统 RAG 通常按照“问题 -> 检索 -> 拼接上下文 -> 生成答案”的直线运行。它容易理解，但无论问题是否需要知识库、检索结果是否相关，流程都会继续向前。一旦初次检索偏离主题，最终答案也会被低质量上下文拖累。

代理式 RAG 把这条直线改造成带反馈的图。系统先判断是否需要检索；检索后评估文档与问题的相关性；相关则基于资料生成答案，不相关则重写查询，再次检索。模型不只是生成答案，还参与路由、评估和纠错。

用户问题
   |
决策节点 ---- 已知答案 ----------------> 直接回答
   |
需要检索
   v
检索节点 -> 相关性评估 ---- 相关 ------> 生成答案 -> END
                         |
                         `-- 不相关 --> 重写问题 --+
                                                   |
                                                   +--> 再次检索

一个清晰的代理式 RAG 可以拆成四个核心节点：

• 决策节点：判断直接回答还是生成知识库检索请求。
• 检索节点：执行向量检索或其他知识库查询，并把结果写入消息状态。
• 问题优化节点：当资料不相关时，结合原问题推断真实意图并生成更合适的查询。
• 答案生成节点：只使用经过评估的资料组织回答，降低无依据生成的风险。

图中最关键的是两条条件边：一条判断是否需要工具，另一条判断检索质量。相关性评估可以要求模型输出结构化的 yes/no 结果，再映射到“生成答案”或“重写问题”两个目标节点。实际项目还应增加最大重试次数、空结果处理和超时降级，避免图在低质量检索下无限循环。

把状态设计成清晰的 API 契约

随着图变复杂，状态不应退化成一个什么都往里塞的字典。输入、内部状态和输出承担不同职责，把它们分开能够限制数据暴露，也让服务接口更稳定。

需要重置时，用 Overwrite 绕过 reducer

reducer 适合累积数据，但有时业务明确要求整体替换。例如用户开始新会话、错误恢复后清理旧状态，或发现缓存内容已损坏。Overwrite 可以跳过原有合并规则，直接写入新值。它是一种显式的例外机制，应只在确实需要重置语义时使用。

输入模式、内部模式与输出模式分离

class InputState(TypedDict):
    question: str

class OutputState(TypedDict):
    answer: str

class OverallState(InputState, OutputState):
    retrieved_docs: list[str]
    retry_count: int

builder = StateGraph(
    OverallState,
    input_schema=InputState,
    output_schema=OutputState,
)

调用方只提交 question，也只收到 answer；检索文档、重试次数和其他中间数据留在图内部。这种设计尤其适合 API、微服务和数据管道，因为对外契约不会随着内部实现细节一起膨胀。

私有状态只在需要的节点之间流动

某些中间数据对处理有用，却不应进入最终输出，例如数据库原始记录、认证令牌、调试详情或包含敏感字段的临时结果。可以为特定节点定义更窄的输入输出类型，让数据只在必要的处理链中传递，最终节点只看到清理后的公共状态。

走向生产环境，真正要补齐什么

图结构解决了流程表达问题，但生产级智能服务还需要一组围绕执行过程的工程能力。LangGraph 的优势在于这些能力都能围绕同一份状态和拓扑自然展开。

• 持久化与检查点：在关键节点保存状态，使长任务在进程重启或工具失败后能够继续，而不是从头执行。
• 人工介入：在付款、发送消息、修改数据等高风险动作前暂停图，让用户审阅、修改状态或批准继续。
• 可观测性：记录每个节点的输入、输出、耗时、工具调用和路由结果，并结合 LangSmith 等工具查看完整追踪。
• 容错策略：为外部工具设置超时、重试和降级路径；将可能重复执行的节点设计为幂等，避免恢复后产生重复副作用。
• 边界控制：限制工具集合、参数范围、循环次数和总预算，让智能决策始终运行在业务允许的空间内。
• 可扩展部署：把对话状态、任务线程和运行记录放入可共享的存储，使多实例服务仍能一致地恢复和继续任务。

一套更稳妥的落地方法

面对一个新的智能应用，可以按下面的顺序设计，而不是直接从提示词和模型选择开始：

1. 先写清输入与最终输出，确定系统对外承诺的数据契约。
2. 列出任务中必须保存的事实，把它们定义为结构化 State，并为累积字段选择 reducer。
3. 把每个动作拆成单一职责节点，优先使用普通代码完成可确定的逻辑。
4. 用固定边搭出主干，只把分类、路由、评估和开放式规划交给模型。
5. 为循环设置退出条件和次数上限，为外部调用增加超时、重试与降级。
6. 最后再加入检查点、人工确认、追踪和流式输出，并用真实失败场景验证恢复能力。

这种顺序能避免“先做出一个会演示的代理，再补工程能力”的常见困境。一个可靠的智能系统，往往不是模型调用最多的系统，而是状态最清楚、节点最克制、路径最可解释的系统。

结语

LangGraph 带来的核心变化，是把大模型应用从一段隐含流程的调用代码，提升为一个显式、有状态、可检查的执行系统。State 保存任务现场，Node 封装单一动作，Edge 描述确定路径与动态分支，持久化和人工介入则让系统能够面对真实世界的不确定性。

当应用只需要一次问答时，简单调用已经足够；当它需要记住、规划、使用工具、反复评估并可靠完成任务时，图式编排才真正体现价值。理解这条边界，也就理解了 LangGraph 最适合解决的问题。