8.2 状态机与工作流引擎

状态机是工作流编排的数学基础，通过有限个状态和转移规则来描述系统行为。本节介绍 FSM 原理、Claude Code 的工作流实现、OpenClaw 的 Lobster 引擎和 Python 状态机的完整实现，最后讨论把确定性封装为代码的两种做法：由工程师沉淀的 Interpreter Skills，以及由模型当场编写的程序化工具调用。

8.2.1 有限状态机的核心原理

有限状态机(Finite State Machine, FSM)是工作流引擎的数学基础。FSM 通过状态和转移来描述系统在不同条件下的行为演进。

在 Agent 编排中，FSM 定义了：

1. 有限个状态：工作流的每个阶段（如“待验证”、“执行中”、“完成”）

2. 输入符号：触发转移的事件（如“approve”、“error”、“timeout”）

3. 转移函数：δ(state, event) → new_state

4. 初始状态 和 接受状态

8.2.2 Claude Code 的工作流执行方式

在 Claude Code 中，FSM 通过智能体的 Task 系统实现，支持通过 Task 的 Workflow 类型来定义状态机。主要特点：

• 代码优先：通过 Python 代码定义状态和转移

• 灵活的条件：支持任意 Python 表达式作为转移条件

• 内置上下文：自动维护上下文变量和执行历史

• 错误处理：与智能体的错误恢复机制深度集成

8.2.3 OpenClaw Lobster 引擎概述

Lobster 是 OpenClaw 的确定性工作流引擎，其核心特点：

• 声明式定义：YAML 格式工作流定义，无需编码

• 确定性执行：相同输入保证相同的执行路径和输出

• 副作用暂停：在执行副作用前暂停，等待人工审批

• 自动恢复：支持从检查点恢复中断的执行

• 可审计：完整的执行历史和决策日志

Lobster 引擎的执行过程：

1. YAML解析 → 构建内部FSM表示
2. 初始化执行上下文
3. 循环执行:
   - 评估当前状态的出边条件
   - 选择满足条件的转移
   - 标记副作用(需审批)或执行纯计算
   - 状态转移
   - 检查是否到达终止状态

状态机的核心循环提供了通用的执行框架。为了使非技术人员也能定义和配置工作流，系统设计了一套标准的 YAML 语法，用于声明式地描述工作流的结构和行为。

8.2.4 YAML 工作流定义语法

基本结构

YAML 工作流定义的基本结构如下：

version: "1.0"
name: "工作流名称"
description: "工作流描述"

## 全局变量
variables:
  max_retries: 3
  timeout: 300

## 状态定义
states:
  start:
    type: initial

  validate:
    type: normal
    actions:
      - id: validation_check
        type: tool_call
        tool: validator
        params:
          data: "{{ context.input }}"

  execute:
    type: normal
    actions:
      - id: main_execution
        type: tool_call
        tool: executor
        side_effect: true  # 需要审批

  retry:
    type: normal
    actions:
      - id: retry_execution
        type: tool_call
        tool: executor

  error:
    type: final

  complete:
    type: final

## 转移定义
transitions:
  - from: start
    to: validate

  - from: validate
    to: execute
    condition: "{{ result.validation_check.success }}"

  - from: validate
    to: error
    condition: "{{ not result.validation_check.success }}"

  - from: execute
    to: complete
    condition: "{{ result.main_execution.success }}"

  - from: execute
    to: retry
    condition: "{{ attempt < variables.max_retries }}"

## 错误处理
error_handlers:
  - on_state: "*"
    action: log_error
    fallback_state: error

tool_call 动作描述的是工作流意图，不代表编排层可以直接绕过执行治理。实际执行必须进入运行时 / 工具执行器的权限、参数验证、审计和追踪路径，尤其是 side_effect: true 的动作。

状态类型详解

状态类型	含义	特点
initial	初始状态	工作流启动时的状态，有且仅有一个
normal	普通状态	执行 actions，评估转移条件
final	终止状态	工作流成功完成，可有多个
error	异常状态	工作流失败或异常中止
wait	等待状态	等待外部输入或异步结果
parallel	并行状态	同时执行多个分支

条件分支示例

条件分支的 YAML 定义示例如下：

states:
  classify:
    type: normal
    actions:
      - id: classify_task
        type: tool_call
        tool: classifier
        params:
          input: "{{ context.data }}"

transitions:
  # 基于分类结果的条件分支
  - from: classify
    to: path_a
    condition: "{{ result.classify_task.category == 'typeA' }}"

  - from: classify
    to: path_b
    condition: "{{ result.classify_task.category == 'typeB' }}"

  - from: classify
    to: path_c
    condition: "{{ result.classify_task.category == 'typeC' }}"

  # 默认分支
  - from: classify
    to: unknown
    condition: "{{ true }}"

循环与重试

循环与重试的 YAML 定义示例如下：

states:
  retry_loop:
    type: normal
    variables:
      attempt: 0
    actions:
      - id: attempt_operation
        type: tool_call
        tool: operation
        params:
          data: "{{ context.data }}"

transitions:
  # 成功则继续
  - from: retry_loop
    to: process_result
    condition: "{{ result.attempt_operation.success }}"

  # 失败且还有重试次数则重试
  - from: retry_loop
    to: retry_loop
    condition: "{{ not result.attempt_operation.success and state.attempt < variables.max_retries }}"
    on_transition:
      - action: increment_variable
        variable: state.attempt

  # 失败且无重试次数则进入错误处理
  - from: retry_loop
    to: handle_error
    condition: "{{ not result.attempt_operation.success and state.attempt >= variables.max_retries }}"

8.2.5 Python 状态机实现

状态机的 Python 实现包括数据结构定义、执行引擎逻辑、和具体使用示例。我们将分三部分展示：

第一部分：数据结构与枚举类型

首先定义表示状态、动作、转移的数据结构：

from typing import Dict, List, Callable, Any, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum
import json
from datetime import datetime

class ActionType(Enum):
    """Action类型"""
    TOOL_CALL = "tool_call"
    CONDITIONAL = "conditional"
    SET_VARIABLE = "set_variable"
    LOG = "log"
    PARALLEL = "parallel"

class StateType(Enum):
    """状态类型"""
    INITIAL = "initial"
    NORMAL = "normal"
    FINAL = "final"
    ERROR = "error"
    WAIT = "wait"
    PARALLEL = "parallel"

@dataclass
class Action:
    """Action定义"""
    action_id: str
    action_type: ActionType
    side_effect: bool = False
    params: Dict[str, Any] = field(default_factory=dict)
    result: Optional[Dict[str, Any]] = None
    executed: bool = False

@dataclass
class State:
    """状态定义"""
    state_id: str
    state_type: StateType
    actions: List[Action] = field(default_factory=list)
    variables: Dict[str, Any] = field(default_factory=dict)

@dataclass
class Transition:
    """转移定义"""
    from_state: str
    to_state: str
    condition: Optional[Callable[[Dict[str, Any]], bool]] = None
    on_transition: List[Callable] = field(default_factory=list)

设计说明：分离 ActionType 和 StateType 的枚举使代码更加类型安全。每个 Action 都包含“副作用”标记，这对于支持需要人工审批的操作至关重要。

第二部分：工作流执行引擎

执行引擎是状态机的核心，负责状态转移和动作执行的逻辑：

class WorkflowExecutor:
    """工作流执行引擎"""

    def __init__(self):
        self.states: Dict[str, State] = {}
        self.transitions: List[Transition] = []
        self.current_state: Optional[str] = None
        self.execution_history: List[Dict] = []
        self.context: Dict[str, Any] = {}
        self.execution_paused = False
        self.pending_approvals: Dict[str, Action] = {}

    def add_state(self, state: State) -> None:
        """添加状态"""
        self.states[state.state_id] = state

    def add_transition(self, transition: Transition) -> None:
        """添加转移"""
        self.transitions.append(transition)

    def initialize(self, initial_state: str, context: Dict[str, Any]) -> None:
        """初始化工作流"""
        if initial_state not in self.states:
            raise ValueError(f"初始状态{initial_state}不存在")
        self.current_state = initial_state
        self.context = context
        self.execution_history = []
        self.pending_approvals = {}
        self.execution_paused = False

    def evaluate_condition(self, condition: Callable) -> bool:
        """评估转移条件"""
        try:
            return condition(self.context)
        except Exception as e:
            print(f"条件评估失败: {e}")
            return False

    def find_next_state(self) -> Optional[str]:
        """根据当前状态和条件查找下一个状态"""
        applicable_transitions = [
            t for t in self.transitions if t.from_state == self.current_state
        ]
        for transition in applicable_transitions:
            if transition.condition is None or self.evaluate_condition(transition.condition):
                return transition.to_state
        return None

设计说明：find_next_state 采用了第一个满足条件的转移，这意味着转移的顺序很重要。在实际应用中可以添加优先级或更复杂的选择策略。

第三部分：动作执行与审批处理

这部分处理具体的动作执行，包括对副作用的暂停和人工审批：

from __future__ import annotations
from typing import Any, Dict


class WorkflowExecutor:
    # 省略 __init__、状态注册和转移选择方法；此处只展示执行/审批相关方法。

    def execute_action(self, action: Action) -> bool:
        """执行单个Action"""
        if action.executed:
            return True

        if action.action_type == ActionType.TOOL_CALL:
            if action.side_effect:
                self.pending_approvals[action.action_id] = action
                self.execution_paused = True
                print(f"[审批] Action {action.action_id} 需要批准")
                return False
            else:
                action.result = self._simulate_tool_call(action.params)
                action.executed = True
                return True

        elif action.action_type == ActionType.CONDITIONAL:
            result = self.evaluate_condition(action.params.get("condition"))
            action.result = {"value": result}
            action.executed = True
            return True

        elif action.action_type == ActionType.SET_VARIABLE:
            var_name = action.params.get("name")
            var_value = action.params.get("value")
            self.context[var_name] = var_value
            action.executed = True
            return True

        elif action.action_type == ActionType.LOG:
            message = action.params.get("message", "")
            print(f"[日志] {message}")
            action.executed = True
            return True
        return False

    def _simulate_tool_call(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """模拟工具调用"""
        return {"success": True, "output": f"Executed with {params}"}

    def execute_state(self) -> bool:
        """执行当前状态的所有actions"""
        if self.current_state is None:
            return False
        state = self.states[self.current_state]
        print(f"\n[进入状态] {self.current_state}")
        for action in state.actions:
            if not self.execute_action(action):
                if self.execution_paused:
                    return False
        return True

    def approve_action(self, action_id: str) -> bool:
        """批准等待的副作用Action"""
        if action_id not in self.pending_approvals:
            return False
        action = self.pending_approvals[action_id]
        action.result = self._simulate_tool_call(action.params)
        action.executed = True
        del self.pending_approvals[action_id]
        if not self.pending_approvals:
            self.execution_paused = False
        return True

    def step(self) -> bool:
        """执行一步:执行当前状态,查找下一个状态"""
        if self.current_state is None:
            return False
        if not self.execute_state():
            return False
        self.execution_history.append({
            "timestamp": datetime.now().isoformat(),
            "state": self.current_state,
            "context": dict(self.context),
        })
        next_state = self.find_next_state()
        if next_state is None:
            return False
        self.current_state = next_state
        if self.states[self.current_state].state_type == StateType.FINAL:
            return False
        return True

    def run(self) -> Dict[str, Any]:
        """执行工作流直到完成或错误"""
        max_iterations = 100
        iterations = 0
        while iterations < max_iterations:
            if self.execution_paused:
                break
            if not self.step():
                break
            iterations += 1
        return {
            "final_state": self.current_state,
            "context": self.context,
            "history": self.execution_history,
            "paused": self.execution_paused,
            "pending_approvals": list(self.pending_approvals.keys()),
        }

    def get_status(self) -> Dict[str, Any]:
        """获取当前工作流状态"""
        return {
            "current_state": self.current_state,
            "is_paused": self.execution_paused,
            "pending_approvals": list(self.pending_approvals.keys()),
            "context": self.context,
            "history_length": len(self.execution_history),
        }

设计说明：approve_action 方法允许外部系统（如用户或管理员）在工作流暂停时批准或拒绝操作。execute_action 必须先跳过已经执行的动作，否则审批后再次 run() 会把同一副作用动作重新挂起，无法继续转移到下一个状态。

第四部分：使用示例

以下是一个完整的审批工作流示例，展示如何构建和执行状态机。首先定义状态，然后定义转移，最后执行工作流：

# 构建审批工作流
executor = WorkflowExecutor()

# 定义状态
executor.add_state(State(
    state_id="submitted",
    state_type=StateType.INITIAL,
    actions=[Action(action_id="validate_input", action_type=ActionType.TOOL_CALL)],
))
executor.add_state(State(
    state_id="reviewing",
    state_type=StateType.NORMAL,
    actions=[Action(
        action_id="apply_changes",
        action_type=ActionType.TOOL_CALL,
        side_effect=True,  # 需要人工审批
        params={"target": "production"},
    )],
))
executor.add_state(State(state_id="approved", state_type=StateType.FINAL))
executor.add_state(State(state_id="rejected", state_type=StateType.FINAL))

# 定义转移
executor.add_transition(Transition(
    from_state="submitted",
    to_state="reviewing",
    condition=lambda ctx: ctx.get("valid", False),
))
executor.add_transition(Transition(
    from_state="submitted",
    to_state="rejected",
    condition=lambda ctx: not ctx.get("valid", False),
))
executor.add_transition(Transition(
    from_state="reviewing",
    to_state="approved",
    condition=lambda ctx: ctx.get("approved", False),
))

# 初始化并执行
executor.initialize("submitted", context={"valid": True})
result = executor.run()
print(result)
# {"final_state": "reviewing", "paused": True, "pending_approvals": ["apply_changes"], ...}

# 模拟人工审批
executor.context["approved"] = True
executor.approve_action("apply_changes")
result = executor.run()
print(result)
# {"final_state": "approved", "paused": False, ...}

8.2.6 错误处理与恢复

错误处理策略

工作流可能在多个阶段出现错误：

1. 验证错误：输入数据不合法

2. 执行错误：工具调用失败

3. 超时错误：任务执行超时

4. 审批拒绝：人工审批被拒

error_handlers:
  - on_state: validate
    error_type: validation_error
    action: notify_requester
    fallback_state: rejected

  - on_state: execute
    error_type: tool_failure
    action: retry_with_backoff
    max_retries: 3
    fallback_state: manual_intervention

  - on_state: "*"
    error_type: timeout
    action: kill_workflow
    fallback_state: error_state

检查点与恢复

通过保存执行状态，工作流可以从中断点恢复：

def save_checkpoint(self) -> str:
    """保存检查点"""
    checkpoint = {
        "current_state": self.current_state,
        "context": self.context,
        "execution_history": self.execution_history,
        "timestamp": datetime.now().isoformat(),
    }
    checkpoint_id = hash_checkpoint(checkpoint)
    # 保存到持久化存储
    return checkpoint_id

def restore_from_checkpoint(self, checkpoint_id: str) -> bool:
    """从检查点恢复"""
    checkpoint = load_checkpoint(checkpoint_id)
    if not checkpoint:
        return False

    self.current_state = checkpoint["current_state"]
    self.context = checkpoint["context"]
    self.execution_history = checkpoint["execution_history"]
    return True

8.2.7 代码即工作流：Deep Agents 的 Interpreter Skills

前面两节展示了两种把“确定性”注入工作流的路径：Claude Code 用代码定义的 Task 状态机（8.2.2），OpenClaw 用声明式 YAML 的 Lobster 引擎（8.2.3）。两者都预先固定了执行路径。但现代智能体 Harness 的主流设计恰恰相反——让模型基于当前上下文自行决定下一步。这就带来一个反复出现的工程问题：

如何在保留模型自主性的同时，确保某段关键流程一定按既定方式执行？

LangChain 在 Deep Agents 中给出的一个实验性答案是 Interpreter Skills（解释器技能）。它把 Agent Skills 标准（SKILL.md，开放规范的系统介绍见《智能体 AI 权威指南》4.4 节，本书 1.3、10.2 亦有涉及）与 解释器（Interpreter） 结合：解释器是一个与 Harness 协同运行的 TypeScript 运行时，智能体可在其中直接编写并执行代码，而 Harness 仍然控制这段代码能触及什么——文件、网络、工具、子智能体都需要显式授权，便于 Harness 在此处做允许、计量与审查。

Interpreter Skills 的形态是“指令 + 模块”两层：

• SKILL.md 通过 frontmatter 声明技能元数据，并用 metadata.module 指向一个代码模块，告诉模型何时该用这个行为；

• index.ts 导出工作流或辅助函数，定义这个行为如何执行。

---
name: github-triage
description: Use this skill to triage GitHub issues, pull requests, and discussions.
metadata:
  module: ./index.ts
---

当用户请求仓库分诊（triage）时使用本技能。
通过解释器导入模块并调用 `triage(repo, options)`。

当技能适用时，模型自行决定调用时机与入参，由解释器执行模块代码：

const { triage } = await import("@/skills/github-triage");

const result = await triage("langchain-ai/deepagents", {
  issues: true,
  prs: true,
  discussions: true,
});

result.toMarkdown();   // 返回结构化结果，并可渲染为模型友好的报告

一个编排型工作流的例子

以“仓库分诊”为例，triage() 内部是一条固定的流水线：

1. 按入参从 GitHub 拉取所有未关闭的 issue / PR / discussion；

2. 为每个条目 spawn 一个子智能体，生成压缩后的描述；

3. 将子智能体的结果投入一个队列；

4. 逐条消费队列，由子智能体判断该条目应归入已有聚类，还是新建聚类。

这正是 8.3 将要展开的多智能体编排，只是这里它被封装成一段可复用、可测试、可版本化的代码，而非散落在提示词里的步骤说明。也正因为运行在解释器里、而非外部脚本，模块能直接参与 Harness 循环：spawn 子智能体、调度与等待任务图、处理部分失败，并在全部完成后才把控制权交还模型。

为什么是“技能”，而不是脚本或工具？

封装形态	适合做什么	局限
工具（Tool）	跨越外部边界：取数、读写文件、建工单、调用分类器	把每个 parse / join / filter 都做成工具会膨胀动作面，模型要在更多小动作间做更多次模型中介的决策
脚本（Script）	一次性外部计算，通过命令行参数 / stdout 通信	无法自然参与 Harness 循环（spawn 子智能体、调度任务图、处理部分失败）
解释器技能（Interpreter Skill）	把“已知有效”的确定性子流程封装为代码，由模型决定何时调用	仍依赖模型正确判断“何时适用”；目前是 Deep Agents 特定的实验特性

把确定性放进代码模块，还顺带改善了长流程下的稳定性与可评估性：

• 缓解上下文焦虑：仓库分诊不是一个决策，而是数百个小决策的链条。若全靠提示词与上下文窗口来支撑，模型在接近工作上下文边缘时容易过度压缩流程、走捷径——即本书 8.3.4、2.2 讨论的“上下文焦虑”。把流程交给代码后，模型只需调用一次例程，由代码去创建数百个子任务、收集并分类，模型不再为每个中间步骤承担工作记忆负担。

• 更清晰的评估信号：纯提示词流程只能问“模型大体上遵循指令了吗”；而对解释器技能，可以问出更具体的问题——是否调用了预期的函数？是否传入了预期的入参？返回的结构是否符合预期？

工程启示：把“最佳已知例程”沉淀为库

至此，本节出现了三种为工作流注入确定性的路径，可对照理解：

方案	确定性来源	形态	代表系统
Task 工作流	代码定义的状态机	Python 代码	Claude Code（8.2.2）
Lobster 引擎	声明式 FSM	YAML 配置	OpenClaw（8.2.3）
Interpreter Skills	可复用代码模块	TS 模块 + SKILL.md	LangChain Deep Agents

Deep Agents 是本书 1.4 已确立的 Harness 参考系统之一。Interpreter Skills 的设计取向可以概括为一句话：“外层交给裁量，内层交给确定性（discretion on the outside, determinism on the inside）”——模型负责判断“要不要用、用在哪”，代码负责“具体怎么跑”。

注意：Interpreter Skills 是 LangChain 在 Deep Agents 中提出的实验性扩展（原文明确为 “We're experimenting with”），并非行业标准。它与 8.2.2／8.2.3 的状态机方案是并列的设计选择而非取代关系：当流程天然是有限状态迁移时，声明式 FSM 往往更直观；当流程更像“一段需要原子完成的子程序”时，代码模块更合适。

8.2.8 模型现写的确定性代码：程序化工具调用

前面三种方案里，确定性逻辑都是工程师事先写好的。另一种做法是让模型在任务中当场写代码。Anthropic 的 程序化工具调用（programmatic tool calling）把允许的工具暴露为沙箱容器里的 async Python 函数，模型写一段代码，用循环、条件和变量把多次工具调用串起来。代码运行期间不再重新采样模型，中间结果也不进入上下文，只有最终输出回到模型。Anthropic 在 Code execution with MCP 一文中把同样的思路用到 MCP 上：智能体按需读取工具定义、写代码调用工具，文中示例的 token 用量从约 15 万降到约 2000。Cloudflare 把这种做法称为“Code Mode”。

这种做法改变的是确定性代码由谁来写，没有改变由谁来把关。对 Harness 有三点影响：

1. 每次子调用仍然经过 Harness。代码调用工具时执行会暂停，API 返回一个 tool_use 块，由客户端执行工具后把结果交回，代码再继续运行。响应中的 caller 字段标明这次调用来自代码还是模型直接发起。权限检查、参数校验和审计对两种来源应一视同仁。caller 在续传协议里有必要的用途（识别待回传的程序化调用，这类结果只能回传文本），也可用于审计统计，但不能作为放宽检查的依据。

2. allowed_callers 不是安全边界。工具定义里的 allowed_callers 决定工具以哪种方式呈现给模型，并会与 tool_choice 做校验。官方文档明确说明它不是 API 层面的硬性拦截，客户端仍要准备处理对任何已定义工具的直接调用。

3. 中间结果的校验要放在工具执行层。中间结果不进入上下文，少了一条把注入内容送进模型的通道；但工具结果以字符串形式交给运行中的代码，可能夹带会被执行环境当作代码解释的内容，官方文档专门提醒要校验来自外部来源的工具结果。这类校验必须由工具执行层（第 5 章）完成。

如果不用托管版本，而在自己的基础设施上实现同一模式，官方文档给出两条路：在应用进程里直接执行模型写的代码，实现最简单，但等于在沙箱外运行不可信代码；或者自建沙箱，并设计沙箱内外的工具调用协议，安全但构建和维护成本高。其中沙箱隔离的部分见 12.2 节。

与 8.2.7 对照，Interpreter Skills 把已知有效的流程沉淀为库，适合反复出现的例程；程序化工具调用让模型为眼前的任务写一次性流程，适合数据量大、步骤多但不值得沉淀为库的任务。两者都是“外层裁量、内层确定”，区别在于内层代码的作者是工程师还是模型。

8.2.9 本小节小结

状态机提供了优雅的方式来描述工作流的行为。OpenClaw 的 Lobster 引擎通过 YAML 声明式定义和确定性执行，使复杂工作流的编写和维护变得简洁可靠；结合 Python 实现的执行引擎，我们可以处理条件分支、循环、重试和错误恢复等复杂场景。而 LangChain Deep Agents 的 Interpreter Skills 则代表另一条路径——把确定性封装为可复用的代码模块，由模型决定调用时机，实现“外层裁量、内层确定”。程序化工具调用更进一步，内层的确定性代码由模型当场编写，但每一次工具调用仍要经过 Harness 的检查。下一节将探讨多个智能体如何协调执行更复杂的任务。