上一篇介绍主流 Agent Framework 时，我们已经看到，简单 Agent 的核心仍然是 Model → Action → Observation → Model。模型根据当前上下文判断下一步，调用 Tool 获取结果，再结合新的 Observation 继续决策。
这种自由循环非常适合搜索、研究和 Coding 等开放式任务，但进入需求分析、合同审核、审批、财务操作等企业场景后，系统关注的问题开始发生变化：Agent 当前执行到哪里？哪些步骤不能跳过？一个节点最多允许重复多少次？服务重启以后怎样继续？什么时候必须让人确认？
例如一个需求分析 Agent，需要完成需求提取、完整性检查、生成、质量评价、修改和人工确认。如果全部继续写在一个巨大的 while + if + else 中，代码很快就会变得难以维护。
因此，复杂业务 Agent 更适合采用一个基本原则：状态机负责控制任务如何流转，Agent 负责解决节点内部无法完全写死的智能问题。
这一篇就围绕一个“需求分析 Agent”，看看如何通过 State、Node、Edge、Checkpoint、Interrupt 和 Human-in-the-Loop，真正开发一个更加可控的 Agent。
一、为什么生产系统不能只依赖自由 Agent Loop
上一篇手写的最小 Agent，本质上是：
while not finished: response = model(messages) if response.tool_calls: result = execute_tool(response.tool_calls) messages.append(result) else: finished = True
这种方式最大的优势是灵活：模型可以决定调用哪个 Tool、是否继续检索以及什么时候结束。问题也恰恰来自这种自由度。
| 维度 | 自由 Agent Loop | 状态机 Agent |
| 下一步由谁决定 | 主要由模型动态判断 | State 和 Edge 限定 |
| 执行路径 | 开放、动态 | 显式、可预测 |
| 循环方式 | Agent 自主继续 | 条件和次数明确 |
| 中断恢复 | 需要额外实现 | 天然适合 Checkpoint |
| 人工审批 | 需要嵌入 Loop | 可以成为正式 Node |
| 适合场景 | Coding、研究、搜索 | 需求、合同、审批、业务流程 |
现实中的生产 Agent 通常并不是二选一，而是：外层使用确定性的 State Machine 控制业务边界，节点内部继续使用 LLM / Agent 处理不确定问题。
二、先定义 State：把任务状态从 Messages 中拿出来
LangGraph 当前的核心抽象就是 State、Node 和 Edge：State 表示应用当前快照，Node 读取并更新 State，Edge 根据当前状态决定下一步执行什么。对于需求分析 Agent，我们先定义真正影响流程的数据：
from typing import TypedDict class RequirementState(TypedDict): raw_requirement: str extracted_data: dict missing_fields: list[str] requirements: list[dict] quality_score: int revision_count: int human_decision: str
这些字段分别回答：
| State 字段 | 表达的状态 |
raw_requirement |
用户原始需求 |
extracted_data |
已提取的结构化信息 |
missing_fields |
仍然缺少什么 |
requirements |
当前生成的需求条目 |
quality_score |
当前质量得分 |
revision_count |
已经自动修改多少次 |
human_decision |
人工最终决策 |
相比上一篇只维护：messages = [...]。这里最大的变化不是多定义了几个变量，而是：任务的业务状态被显式建模了。即使模型上下文重新组装，系统仍然知道任务执行到了哪里。
三、Node：让 LLM 只负责节点内部的智能问题
有了 State 以后，把需求分析过程拆成几个职责明确的 Node：
extract check generate review revise human_approval
每个 Node 接收当前 State，只返回自己产生的更新。LangGraph 的 Node 本质上就是这种读取当前 State、执行计算并返回部分 State Update 的函数。例如需求提取节点：
def extract_requirement(state: RequirementState): extracted = llm_extract( state["raw_requirement"] ) return { "extracted_data": extracted }
llm_extract() 代表我们封装好的模型调用，这里不展开具体 Provider API。完整性检查则更适合优先使用确定性代码：
REQUIRED_FIELDS = [ "role", "goal", "input", "output" ] def check_completeness(state: RequirementState): data = state["extracted_data"] missing = [ field for field in REQUIRED_FIELDS if not data.get(field) ] return { "missing_fields": missing }
这里已经体现出一个非常重要的分工：
LLM 负责从自然语言中理解：用户角色是什么？目标是什么？输入输出是什么？
程序 负责检查：这些必填字段到底有没有。
生产 Agent 不应该为了体现“智能”而把这种确定性判断也重新交给模型。
四、Conditional Edge：模型可以判断，但程序决定怎样跳转
状态机真正开始发挥价值，是在条件分支出现以后。检查完需求完整性，我们需要决定：
- 信息完整 → 继续生成；
- 信息不足 → 请求用户补充。
可以定义一个非常简单的路由函数：
def route_completeness(state: RequirementState): if state["missing_fields"]: return "ask_user" return "generate"
再把它注册成 Conditional Edge：
builder.add_conditional_edges( "check", route_completeness, { "ask_user": "ask_user", "generate": "generate" } )
LangGraph 当前的 add_conditional_edges() 就是根据一个 routing function 的结果动态选择下一节点。这与自由 Agent Loop 有一个非常重要的区别。
自由 Agent 可能是：“模型，你觉得下一步该怎么办？”
状态机则变成：“系统只允许你在这几个合法路径中继续。”
因此：模型可以参与判断，但最终允许走哪些路径，由程序定义。
五、需求生成和质量检查：把智能留在真正需要智能的地方
需求生成显然更适合 LLM：
def generate_requirement(state: RequirementState): requirements = llm_generate( state["extracted_data"] ) return { "requirements": requirements }
质量评价也可以使用模型，例如从完整性、清晰度、可测试性和一致性几个维度打分：
def review_quality(state: RequirementState): score = llm_review( state["requirements"] ) return { "quality_score": score }
但“评分以后怎么办”，不应该继续让模型自由决定。例如：
MAX_REVISIONS = 3 def route_quality(state: RequirementState): if state["quality_score"] >= 80: return "human_approval" if state["revision_count"] >= MAX_REVISIONS: return "human_approval" return "revise"
这样即使模型连续给出：72 → 76 → 79 → 79。系统也不会无限进入：“再优化一次看看。”第三次仍未达标，就转人工。这体现了状态机 Agent 最核心的工程思想：模型决定怎么修改，Runtime 决定最多允许修改多少次。
六、把这些 Node 真正组装成一个 State Graph
有了 State、Node 和路由函数以后，整个 Agent 的核心结构已经可以写出来：
from langgraph.graph import ( StateGraph, START, END ) builder = StateGraph(RequirementState) builder.add_node( "extract", extract_requirement ) builder.add_node( "check", check_completeness ) builder.add_node( "ask_user", ask_user ) builder.add_node( "generate", generate_requirement ) builder.add_node( "review", review_quality ) builder.add_node( "revise", revise_requirement ) builder.add_node( "human_approval", human_approval ) builder.add_edge( START, "extract" ) builder.add_edge( "extract", "check" )
再把几个关键分支接起来：
builder.add_conditional_edges( "check", route_completeness, { "ask_user": "ask_user", "generate": "generate" } ) builder.add_edge( "generate", "review" ) builder.add_conditional_edges( "review", route_quality, { "revise": "revise", "human_approval": "human_approval" } ) builder.add_edge( "revise", "review" ) builder.add_edge( "human_approval", END )
到这里，文章标题里的“使用状态机开发 Agent”已经真正落到了代码上。状态图不再只是架构图，而是直接成为应用运行逻辑。LangGraph 的 StateGraph 也正是按照“定义 State → 添加 Nodes → 添加 Edges → compile”的方式构建和运行 Graph。
七、状态机不是固定 Workflow：Node 内部依然可以运行 Agent
使用 StateGraph 并不意味着 Agent 被退化成普通 BPM Workflow。例如 generate Node 内部完全可以继续执行：
读取当前 State + RAG 检索业务知识 + 调用需求模板 Skill + LLM 生成
甚至可以封装一个局部 Agent：
def generate_requirement(state): result = requirement_agent.run( goal="生成结构化需求", context=state["extracted_data"], tools=[ search_knowledge, load_requirement_template ] ) return { "requirements": result }
于是整体架构实际变成：外层 State Machine，内层 Agent Loop。
状态机决定：什么时候生成；生成以后去哪；是否允许重新生成；最多重新生成多少次。
Agent 决定：需要查什么资料；调什么 Tool；如何组织最终内容。
这就是“确定性的骨架 + 概率性的智能”。
八、Checkpoint：让 Agent 不再依赖一次请求活到底
到目前为止，这个 Graph 仍然可以一次跑完。但真实需求分析通常会遇到：用户几个小时以后才补充需求；审批人第二天才确认；服务升级；Worker 重启；长任务执行失败。这时就需要 Checkpoint。
LangGraph 支持在 compile Graph 时加入 checkpointer，用于保存 Graph State；官方当前示例使用 InMemorySaver 做内存级 Checkpoint，同时也明确指出生产环境应该使用 PostgreSQL、SQLite 等持久化 Saver。示例：
from langgraph.checkpoint.memory \ import InMemorySaver checkpointer = InMemorySaver() graph = builder.compile( checkpointer=checkpointer )
调用任务时给它一个稳定的 thread_id：
config = { "configurable": { "thread_id": "requirement-1001" } } graph.invoke( initial_state, config=config )
于是：requirement-1001。不再只是一次 HTTP Request，而是一个可以继续恢复的任务实例。生产环境里，InMemorySaver 只是示例，真正部署时应换成持久化 Checkpointer。
九、Interrupt + Resume：真正实现“等待用户”和“人工审批”
需求不完整时：请求用户补充。人工审核时：等待 Approve / Edit / Reject。这两类场景都不应该通过线程等待。LangGraph 当前提供 interrupt() 暂停节点执行，并依赖 Checkpointer 保存当前状态；后续可以通过 Command(resume=...) 使用相同 thread_id 恢复。例如人工审核节点：
from langgraph.types import interrupt def human_approval(state: RequirementState): decision = interrupt({ "type": "requirement_approval", "requirements": state["requirements"], "quality_score": state["quality_score"] }) return { "human_decision": decision }
执行到这里后，Graph 会暂停。审批人稍后提交：APPROVE。系统再恢复：
from langgraph.types import Command graph.invoke( Command(resume="APPROVE"), config=config )
如果审批结果是：EDIT。或者：REJECT。则可以继续通过 Conditional Edge 路由到 Review 或 Revise。需要特别注意：LangGraph 当前的 interrupt 在恢复时会从当前 Node 开头重新执行，因此 interrupt 之前的副作用应该保持幂等，或者拆到独立 Node 中。这个细节很重要，因为它直接关系到：邮件会不会发两次？数据库会不会插入两条记录？审批任务会不会重复创建？
十、Checkpoint、Interrupt、Resume 最好理解成一个生命周期
这一部分不需要再用长数据流表示，可以压缩成三个阶段：
| 阶段 | 作用 |
| Checkpoint | 保存当前 State |
| Interrupt | 暂停任务并释放执行资源 |
| Resume | 恢复 State 后继续执行 |
典型场景包括：等待用户补充需求；等待审批；等待外部系统回调；服务升级与任务恢复；长任务阶段性执行。 
十一、State 和 Memory 不要混在一起
引入 Checkpoint 后，很容易把 State 与后续文章要介绍的 Memory 混淆。
| State | Memory | |
| 范围 | 当前任务 | 跨任务、跨会话 |
| 解决的问题 | 运行到哪里 | 长期应该记住什么 |
| 示例 | 当前评分 85 | 用户偏好敏捷模板 |
| 生命周期 | Task 生命周期 | 可以长期保存 |
| 主要用途 | Checkpoint / Resume | 个性化、经验复用 |
例如：
quality_score = 85 revision_count = 2 human_decision = ""
属于 Task State。而：
用户习惯输出三级需求 用户偏好使用敏捷模板
属于长期 Memory。因此：Checkpoint 解决任务恢复，不等于 Agent 已经拥有长期记忆。
十二、自由 Harness 和 State Graph 是两条不同路线
不是所有 Agent 都应该画成状态图。Coding、研究、搜索任务往往路径非常开放，Agent 可能不断读取代码、运行测试、观察错误、重新修改。Pi 这类 minimal harness 更强调保持自由 Loop，再通过扩展能力增强；DeepSeek Harness 则通过 Session/Event Log 记录模型消息、Tool Call 和 Tool Result，使自由 Agent 同样能够恢复和重放。可以简单比较：
| State Graph | Harness / Event Log | |
| 核心 | 预定义合法路径 | 记录实际执行轨迹 |
| 自由度 | 相对较低 | 相对较高 |
| 适合 | 业务流程 | Coding、研究、探索 |
| 状态 | 显式业务 State | Session / Event |
| 恢复 | Checkpoint | Session Replay / Resume |
| 共同点 | 持久化、恢复、可观测 | 持久化、恢复、可观测 |
因此关键不是：LangGraph 还是 Harness？而是：任务本身是流程型，还是探索型？
十三、企业 Agent 更值得采用“两层控制”
综合前面的代码和设计，一个企业 Agent 更实用的结构通常是两层。
| 确定性控制层 | 智能执行层 |
| State | LLM |
| Node / Edge | RAG |
| Checkpoint | Tool Calling |
| Approval | Reasoning |
| Timeout / Retry | 内容分析 |
| Permission / Recovery | 内容生成 |
外层 Workflow 负责：哪些路径合法、什么时候暂停、什么时候结束。
内层 Agent 负责：节点内部具体怎样解决问题。
确定性的骨架 + 概率性的智能
十四、什么时候值得引入状态机
状态机也不是越多越好。
| 任务特征 | 推荐方式 |
| 搜索、研究、Coding | Agent Loop |
| 简单 Tool Calling | Agent Loop |
| 简单 RAG 问答 | Agent Loop / Workflow |
| 明确业务步骤 | State Machine |
| 多分支、多循环 | State Machine |
| 人工审批 | State Machine |
| 长任务与恢复 | Durable State Runtime |
| 复杂企业任务 | State Machine + Agent |
如果任务只有：查询天气 → 计算 → 回答。完全没有必要引入 StateGraph。
但当出现：分支 + 循环 + Checkpoint + Human Approval + Resume。状态机的价值就开始明显。
十五、小结
上一篇解决的是：主流框架替我们封装了哪些 Agent 能力？这一篇进一步通过实际代码回答：怎样用状态机开发一个真正可控的 Agent？整个需求分析 Agent 最终可以概括成四层：
第一层：State 保存任务当前事实。
第二层：Node 让程序和 LLM 分别完成适合自己的工作。
第三层：Edge 把模型的不确定判断转换成明确的流程路由。
第四层：Checkpoint / Interrupt / Resume 让任务能够暂停、等待人工并可靠恢复。
最终形成的不是纯 Workflow，也不是完全自由的 Agent：状态机负责控制任务如何流转，Agent 负责解决节点内部无法完全写死的智能问题。也就是：确定性的骨架 + 概率性的智能。
对于需求分析 Agent 来说，模型可以理解自然语言、生成需求和评价质量；状态机则决定什么时候补充信息、什么时候修改、最多修改几次以及什么时候必须交给人确认。这也是 Agent 从 Demo 走向真正业务系统的一次重要变化：生产级 Agent 的目标不是拥有无限自主权，而是在明确、可恢复、可审计的边界内完成任务。
上一篇回顾：
https://developer.aliyun.com/article/1758517?spm=a2c6h.13148508.setting.16.2a9a4f0eKImYFP
下一篇将进入：为 Agent 增加记忆能力 。
因为 State Machine 解决的是：这一次任务执行到哪里。而 Memory 要解决的是：跨会话、跨任务以后，Agent 应该记住什么，以及什么时候应该忘掉什么。
