小节二：状态流转——READY、THINKING、TOOL_CALLING 到底为什么要显式设计？

State 不只是“有哪些字段”。

更重要的是，这些状态之间怎么流转。

在继续列状态前，我们先区分两个层次：

Run 生命周期：任务整体是否还在运行
步骤阶段：这一轮正在请求模型、执行工具，还是处理结果

为了先把状态机讲清楚，本节示例暂时把两者放进同一个 AgentStatus，所以会看到 THINKING、TOOL_CALLING、OBSERVING 和 COMPLETED 出现在同一个枚举里。到了配套工程中，RunStatus 主要保存 READY、RUNNING、WAITING_APPROVAL、COMPLETED、FAILED 等生命周期状态；模型调用和工具执行的细节则进入 Step、Tool History 与 Trace。两种写法并不冲突，前者方便理解状态流转，后者更适合服务化查询。

如果我们只保存：

status = "running"

信息太少。

因为“running”可能包含完全不同的阶段：

• 正在调用模型；
• 正在执行 Tool；
• 正在等待 Tool Result；
• 正在等待人工确认；
• 正在生成最终答案。

这些阶段的允许动作完全不同。

所以我们进一步设计：

READY
THINKING
TOOL_CALLING
OBSERVING
WAITING_APPROVAL
COMPLETED
FAILED

把这些名字连成一条可运行的路径，会发现 THINKING → TOOL_CALLING → OBSERVING 是会反复经过的主干；完成、审批暂停和不可恢复失败是不同的分支。

尤其留意 WAITING_APPROVAL：模型提出高风险动作后先停下，得到批准才回到工具执行。一次可恢复的 Tool Error 则会作为 Observation 进入下一轮，并不必然让整个任务变成 FAILED。

READY：任务已经创建，但还没开始

刚创建一个 Run 时：

status = READY

这表示：

• Goal 已经存在；
• State 已初始化；
• 还没开始第一轮模型调用。

它很适合做：

创建 Run
 ↓
READY
 ↓
开始执行

THINKING：当前正在请求模型做决策

进入：

THINKING

表示当前 Loop 正在调用 LLM。

这个状态可以用于：

• 前端展示“正在分析”；
• Trace 记录；
• 超时控制；
• 防止重复发起同一轮决策。

TOOL_CALLING：模型已经决定要执行 Tool

模型返回：

{
  "name": "read_file",
  "arguments": {
    "path": "app.py"
  }
}

这时状态切换：

THINKING
 ↓
TOOL_CALLING

这里 Harness 可以插入：

参数校验
权限校验
Middleware
Human Approval

注意，Tool Call 产生，不等于 Tool 已执行。

所以 TOOL_CALLING 是非常有意义的中间状态。

OBSERVING：工具执行完成，正在吸收结果

Tool 执行后得到：

FILE_NOT_FOUND

或：

文件内容……

此时进入：

OBSERVING

Harness 会：

• 记录 Tool Result；
• 更新 messages；
• 更新 tool_history；
• 记录错误；
• 决定是否继续下一轮。

然后再回到：

THINKING

这就是循环。

WAITING_APPROVAL：动态系统里非常重要的暂停状态

假设 Agent 想：

delete_file("config.py")

Harness 判断，这是高风险操作，需要人工确认。

这时不能：

FAILED

因为任务没有失败。

也不能：

THINKING

因为当前不应该继续模型决策。

所以需要：

WAITING_APPROVAL

这意味着：

• Run 仍然存在；
• 当前动作已经确定；
• 系统暂停；
• 等用户批准后恢复。

这是后面 Human in the Loop 的基础。

COMPLETED 与 FAILED

COMPLETED 表示任务已经正常结束。

通常还会保存：

final_answer
completed_at

FAILED 表示系统确认任务无法继续。

例如：

• 超过最大步数；
• 不可恢复错误；
• 模型服务长期失败；
• 必需资源不存在；
• 安全规则强制终止。

注意，Tool 一次失败，不一定让整个 Agent FAILED。

这是我们上一节讲过的。

Tool Error 可能只是 Observation。

只有确定无法继续，才进入 FAILED。

用 Enum 管理状态

比字符串更稳的方式是 Enum。

from enum import Enum


class AgentStatus(str, Enum):
    READY = "READY"
    THINKING = "THINKING"
    TOOL_CALLING = "TOOL_CALLING"
    OBSERVING = "OBSERVING"
    WAITING_APPROVAL = "WAITING_APPROVAL"
    COMPLETED = "COMPLETED"
    FAILED = "FAILED"

然后 State：

from dataclasses import dataclass, field


@dataclass
class AgentState:
    goal: str
    messages: list[dict] = field(default_factory=list)
    step: int = 0
    tool_history: list[dict] = field(default_factory=list)
    status: AgentStatus = AgentStatus.READY

这样就避免：

status = "COMPELTED"

这种拼写错误。

状态流转也应该限制

不是所有状态都能任意切换。

例如：

COMPLETED

以后不应该再回到：

THINKING

WAITING_APPROVAL 也不应该自动跳到 COMPLETED。

我们可以定义：

ALLOWED_TRANSITIONS = {
    AgentStatus.READY: {
        AgentStatus.THINKING,
        AgentStatus.FAILED,
    },
    AgentStatus.THINKING: {
        AgentStatus.TOOL_CALLING,
        AgentStatus.COMPLETED,
        AgentStatus.FAILED,
    },
    AgentStatus.TOOL_CALLING: {
        AgentStatus.OBSERVING,
        AgentStatus.WAITING_APPROVAL,
        AgentStatus.FAILED,
    },
    AgentStatus.OBSERVING: {
        AgentStatus.THINKING,
        AgentStatus.FAILED,
    },
    AgentStatus.WAITING_APPROVAL: {
        AgentStatus.TOOL_CALLING,
        AgentStatus.FAILED,
    },
    AgentStatus.COMPLETED: set(),
    AgentStatus.FAILED: set(),
}

完整可运行：

from dataclasses import dataclass
from enum import Enum


class AgentStatus(str, Enum):
    READY = "READY"
    THINKING = "THINKING"
    TOOL_CALLING = "TOOL_CALLING"
    OBSERVING = "OBSERVING"
    WAITING_APPROVAL = "WAITING_APPROVAL"
    COMPLETED = "COMPLETED"
    FAILED = "FAILED"


ALLOWED_TRANSITIONS = {
    AgentStatus.READY: {
        AgentStatus.THINKING,
        AgentStatus.FAILED,
    },
    AgentStatus.THINKING: {
        AgentStatus.TOOL_CALLING,
        AgentStatus.COMPLETED,
        AgentStatus.FAILED,
    },
    AgentStatus.TOOL_CALLING: {
        AgentStatus.OBSERVING,
        AgentStatus.WAITING_APPROVAL,
        AgentStatus.FAILED,
    },
    AgentStatus.OBSERVING: {
        AgentStatus.THINKING,
        AgentStatus.FAILED,
    },
    AgentStatus.WAITING_APPROVAL: {
        AgentStatus.TOOL_CALLING,
        AgentStatus.FAILED,
    },
    AgentStatus.COMPLETED: set(),
    AgentStatus.FAILED: set(),
}


@dataclass
class AgentState:
    goal: str
    status: AgentStatus = AgentStatus.READY

    def transition_to(self, new_status: AgentStatus) -> None:
        allowed = ALLOWED_TRANSITIONS[self.status]

        if new_status not in allowed:
            raise ValueError(
                f"非法状态流转：{self.status} -> {new_status}"
            )

        self.status = new_status


if __name__ == "__main__":
    state = AgentState(
        goal="检查项目"
    )

    print("初始：", state.status)

    state.transition_to(AgentStatus.THINKING)
    print("模型决策：", state.status)

    state.transition_to(AgentStatus.TOOL_CALLING)
    print("准备执行工具：", state.status)

    state.transition_to(AgentStatus.OBSERVING)
    print("处理工具结果：", state.status)

    state.transition_to(AgentStatus.THINKING)
    print("继续决策：", state.status)

这段代码看起来比直接赋值麻烦。

但后面你会发现，这种“麻烦”非常值钱。

因为它让整个 Agent 的生命周期开始可控。

状态机真正落地时，关键不只是能画出箭头，还要让代码拒绝“不该存在”的箭头。下面这张允许与拒绝对照图，正对应前面的 ALLOWED_TRANSITIONS 和 transition_to()。

例如已完成的 Run 不能重新跳进 THINKING，等待审批的动作也不能绕过执行阶段直接宣称完成。把这些约束集中在一个入口，后续做暂停和恢复时才不会各处各改一套状态。