【第三部分：第一个 Agent 应用】12. 使用状态机开发一个可控 Agent

​上一篇介绍主流 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 通常并不是二选一，而是：外层使用确定性的 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

这些字段分别回答：

相比上一篇只维护：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 最好理解成一个生命周期

这一部分不需要再用长数据流表示，可以压缩成三个阶段：

典型场景包括：等待用户补充需求；等待审批；等待外部系统回调；服务升级与任务恢复；长任务阶段性执行。

十一、State 和 Memory 不要混在一起

引入 Checkpoint 后，很容易把 State 与后续文章要介绍的 Memory 混淆。

例如：

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 同样能够恢复和重放。可以简单比较：

因此关键不是：LangGraph 还是 Harness？而是：任务本身是流程型，还是探索型？

十三、企业 Agent 更值得采用“两层控制”

综合前面的代码和设计，一个企业 Agent 更实用的结构通常是两层。

外层 Workflow 负责：哪些路径合法、什么时候暂停、什么时候结束。

内层 Agent 负责：节点内部具体怎样解决问题。

确定性的骨架 + 概率性的智能

十四、什么时候值得引入状态机

状态机也不是越多越好。

如果任务只有：查询天气 → 计算 → 回答。完全没有必要引入 StateGraph。

但当出现：分支 + 循环 + Checkpoint + Human Approval + Resume。状态机的价值就开始明显。

十五、小结

上一篇解决的是：主流框架替我们封装了哪些 Agent 能力？这一篇进一步通过实际代码回答：怎样用状态机开发一个真正可控的 Agent？整个需求分析 Agent 最终可以概括成四层：

第一层：State 保存任务当前事实。

第二层：Node 让程序和 LLM 分别完成适合自己的工作。

第三层：Edge 把模型的不确定判断转换成明确的流程路由。

第四层：Checkpoint / Interrupt / Resume 让任务能够暂停、等待人工并可靠恢复。

最终形成的不是纯 Workflow，也不是完全自由的 Agent：状态机负责控制任务如何流转，Agent 负责解决节点内部无法完全写死的智能问题。也就是：确定性的骨架 + 概率性的智能。

对于需求分析 Agent 来说，模型可以理解自然语言、生成需求和评价质量；状态机则决定什么时候补充信息、什么时候修改、最多修改几次以及什么时候必须交给人确认。这也是 Agent 从 Demo 走向真正业务系统的一次重要变化：生产级 Agent 的目标不是拥有无限自主权，而是在明确、可恢复、可审计的边界内完成任务。

上一篇回顾：

【第三部分：第一个 Agent 应用】11.Spring AI、LangChain4j、LangGraph 到底帮我们封装了什么？-腾讯云开发者社区-腾讯云

下一篇将进入：为 Agent 增加记忆能力 。

因为 State Machine 解决的是：这一次任务执行到哪里。而 Memory 要解决的是：跨会话、跨任务以后，Agent 应该记住什么，以及什么时候应该忘掉什么。

​