普通的 LLM 调用通常是一条直线：输入提示词，等待模型返回结果，然后结束。但研究、审核、工具调用和多智能体协作往往不是线性流程。任务可能需要反复修改，也可能根据中间结果切换执行路径，还要在多次调用之间保存进度。LangGraph 的核心价值，就是把这类工作流表达成一张包含状态、节点、边和循环的图。
从调用链转向状态图
在 LangGraph 中，可以把一次智能体运行理解为四个组成部分：
- State：节点共享的数据，例如用户目标、消息、检索结果、草稿和重试次数。
- Node：执行具体工作的函数或智能体，例如研究员、写作者、审核员。
- Edge：定义节点执行顺序。
- Conditional edge：读取当前状态，并决定接下来进入哪个节点。
与固定调用链相比，状态图更适合表达“审核不通过就重新研究”这样的循环。循环本身并不复杂，真正需要认真设计的是退出条件：如果没有最大迭代次数、质量阈值或超时机制，工作流可能永远运行下去，并持续消耗模型调用额度。
状态也不应该变成一个无限增长的字典。工程上通常需要明确区分：
- 只在本轮执行中使用的临时状态；
- 需要通过检查点恢复的线程状态；
- 应长期保存在数据库或向量存储中的用户记忆。
检查点能够恢复图的执行状态，但它不等同于完整的长期记忆系统。
一个可以运行的循环工作流
下面可以这样实践：使用三个确定性 Python 节点模拟研究员、写作者和审核员。示例不调用外部模型，因此不需要 API Key，可以直接观察共享状态、条件路由和循环如何工作。
安装 LangGraph：
python -m pip install -U langgraph
创建 agent_graph.py：
import operator
from typing import Annotated, Literal, TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
class AgentState(TypedDict, total=False):
topic: str
facts: Annotated[list[str], operator.add]
draft: str
revision: int
next: str
def researcher(state: AgentState) -> dict:
round_no = state.get("revision", 0) + 1
fact = (
f"第 {round_no} 轮研究：{state['topic']} 需要明确状态、路由和退出条件。"
)
return {"facts": [fact]}
def writer(state: AgentState) -> dict:
evidence = "\n".join(f"- {fact}" for fact in state["facts"])
draft = (
f"主题：{state['topic']}\n\n"
f"已有材料：\n{evidence}\n\n"
"结论：应把复杂任务拆成可检查、可恢复的图节点。"
)
return {"draft": draft}
def reviewer(state: AgentState) -> dict:
# 至少积累两条研究材料，否则回到 researcher。
if len(state["facts"]) < 2:
return {
"revision": state.get("revision", 0) + 1,
"next": "research",
}
return {"next": "finish"}
def review_route(state: AgentState) -> Literal["research", "finish"]:
return "research" if state["next"] == "research" else "finish"
builder = StateGraph(AgentState)
builder.add_node("researcher", researcher)
builder.add_node("writer", writer)
builder.add_node("reviewer", reviewer)
builder.add_edge(START, "researcher")
builder.add_edge("researcher", "writer")
builder.add_edge("writer", "reviewer")
builder.add_conditional_edges(
"reviewer",
review_route,
{
"research": "researcher",
"finish": END,
},
)
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "demo-thread-001"}}
result = graph.invoke(
{
"topic": "用 LangGraph 构建可靠的智能体工作流",
"facts": [],
"revision": 0,
},
config=config,
)
print(result["draft"])
print(f"研究轮数：{len(result['facts'])}")
snapshot = graph.get_state(config)
print(f"检查点中的下一节点：{snapshot.next}")
运行：
python agent_graph.py
这里有两个容易忽略的细节。
facts 使用了 Annotated[list[str], operator.add]。这表示节点返回新的事实时，LangGraph 会把它追加到现有列表，而不是覆盖整个字段。对于消息历史、工具结果和审计事件，这类 reducer 很有用；但如果列表没有裁剪策略，状态会持续膨胀。
另一个细节是 thread_id。检查点以线程标识区分不同会话。相同的 thread_id 可以定位同一条执行线程，不同用户或任务必须使用不同标识，否则状态可能串到错误的会话中。
接入真实模型时，保留图的控制权
上面的节点只是为了让示例无需外部服务即可运行。接入真实 LLM 时，可以把 writer 替换成模型调用，同时保留审核节点和条件边。建议让模型负责生成内容，让普通 Python 代码负责次数限制、权限检查和路由决策。
可以采用这样的提示词结构：
你是工作流中的写作节点。
主题：{topic}
研究材料：
{facts}
只输出一份技术草稿，不要决定下一步执行哪个节点。
路由和重试由工作流控制器处理。
这种职责划分有两个好处：一是路由规则可以测试，二是模型无法仅凭自然语言输出绕过关键控制逻辑。对于高风险工具，例如付款、删除数据或发布内容，还应在工具节点外增加显式授权和参数校验。
所谓多智能体，也不一定意味着同时运行多个模型。更实用的做法通常是按角色拆节点：研究节点检索材料，写作节点生成草稿，审核节点输出结构化结论，调度节点根据结论选择路径。各节点可以使用不同模型，也可以共享同一个模型但使用不同提示词和工具权限。
记忆、循环和并发的工程边界
内存检查点适合本地开发和测试，但进程退出后数据会丢失。生产环境应选择持久化检查点实现，并确认它与所用 LangGraph 版本兼容。部署前还需要处理以下问题：
- 为循环设置最大次数、超时和总 token 预算；
- 对工具调用增加幂等键，避免恢复执行后重复扣款或重复写入；
- 限制状态体积，必要时摘要旧消息或把大对象存到外部存储；
- 使用不可预测且隔离良好的会话标识，并执行访问控制；
- 记录节点输入、输出、耗时和路由原因，但不要把密钥与敏感数据写入日志；
- 为条件边编写单元测试，覆盖成功、重试、失败和人工介入路径。
LangGraph 适合那些确实需要状态、分支、恢复和循环的流程。若业务只有一次提示词调用，普通函数会更简单；当工作流开始出现审核回路、多角色协作、工具重试和会话恢复时，状态图才真正体现价值。采用时可以先把一个最不稳定的链路改造成图，明确状态字段和终止条件，再逐步加入持久化、可观测性与真实模型调用。