在 Agent Loop 中嵌入确定性执行层

Agent 开发里有一个常见问题：这个场景该用 Workflow 还是 autonomous Agent？事实上这两者并不是互斥的，下面就针对这个问题简单讲解下一些架构上的设计与思考。

1. 相关的设计思想

$a.$ Anthropic：Workflow 与 Agent 不是二选一

Building Effective Agents 中是这样区分 Workflow 和 Agent 的：

Workflow：LLM 和工具通过预定义代码路径编排；
Agent：LLM 动态主导自己的流程和工具使用。

一个设计是：外层是 autonomous loop，内层某个决策被识别为”这一段不需要模型反复想”，于是交给一个确定性 Workflow 或 Plan DAG 执行。下面的

$b.$ 12-Factor Agents：Agent 主要由确定性代码组成

12-Factor Agents 的核心论点是：投产的 Agent 系统，大部分是确定性代码，只在需要语言理解或推理的地方插入 LLM：

Own your control flow：自己写循环，知道在哪里打断、在哪里交给子模块；
Tools are just structured outputs：LLM 输出结构，代码决定如何执行；
Launch, pause, and resume with simple APIs：能在工具选择后、执行前暂停，等待审批或恢复。

这三条共同指向一个设计：autonomous loop 不排斥确定性子系统，关键是由谁来触发、由谁来持久化、由谁来兜底失败。

2. 架构设计

下面是一个内嵌了确定性 Workflow 的 Agent 架构：

┌─────────────────────────────────────────┐
│  1. Autonomous Agent Loop               │  ReAct / 探索 / 工具选择 / 对话管理
├─────────────────────────────────────────┤
│  2. Workflow Runtime / Plan Executor    │  DAG / Plan、Schema、Artifact、Gate
├─────────────────────────────────────────┤
│  3. Bridge Layer                        │  触发方式：tool / meta-tool / prompt
└─────────────────────────────────────────┘

这个架构的核心假设是：根据任务具体类型决定组件的自由度。系统外层保持普通的 Agent Loop，内层把较为固定的业务流程等下沉为确定性代码。

$a.$ 第一层：Autonomous Agent Loop

外层 Agent Loop 负责所有不确定的事情：

理解用户意图；
决定什么时候进入 Workflow；
处理探索性问题、上下文追问、多轮对话；
在 Workflow 失败或暂停时接管，决定重试、回退还是转人工。

$b.$ 第二层：Workflow Runtime / Plan Executor

Workflow 之类的内层负责所有确定的事情：

DAG / Plan 定义：阶段、依赖、执行顺序等；
Schema 契约：每个阶段的输入输出的校验，必须要满足约定的 JSON schema 或其他；
Artifact 管理：中间产物持久化、传递交付、溯源等；
门控（Gate）：阶段完成前必须满足的条件；
Source-bound：某些确定性的步骤不需要模型写 JSON，直接自己传递 JSON Artifact 给确定性逻辑生成下一步的结果。

这一层不调用 LLM 做开放式推理，只按既定规则推进，从而做到可验证、可恢复。

这一层如果进一步抽象成可复用组件——阶段定义、状态文件、Guard/Verifier、阶段间 Handoff——就可以让不同业务流程挂载到同一套推进机制上。更完整的组件化设计见面向 Agent 的轻量 Workflow 设计。

$c.$ 第三层：Bridge Layer

按耦合程度从低到高，有三种方式将确定性执行层嵌入 Agent Loop 中：

方式	耦合程度	说明
Tool-based	较低	Workflow 被包装成 Agent 可调用的 tool，hook 可作为最终响应的补充机制
Meta-tool	较高	Agent 动态生成 Plan，交给内部 Plan Executor
Prompt-level	最高	完全依赖 prompt 约定约束 Agent 自律执行

接下来的章节就用具体的代码例子介绍一下这些大概是怎么实现的、以及确定性执行层可参考的实现方式。

3. 确定性执行层的具体接入与实现

$a.$ Tool-based

这是最常见也最解耦的方式：Workflow 被包装成 Agent 可调用的 tool，Agent Loop 通过标准 tool call 驱动 Workflow。

$i.$ 插件注册逻辑

插件入口 __init__.py 实现 register(ctx)，向 Agent runtime 注册 tool。以 Hermes Agent 为例：

def register(ctx) -> None:
  register_tools(ctx)     # 注册 xxx_workflow tool

register_tools(ctx) 把 <domain>_process_workflow 工具注册到 Agent 中；

$ii.$ Workflow 的定义与运行时分发

Workflow 本身可以用一个 YAML 来定义：

schema: process.workflow.v1
id: process.from-long-desc.v1
enabled: true
idempotency: session_workflow_input_hash
steps:
  - { id: parse, skill: parse-input, output_schema: process.parsed-input.v1 }
  - {
      id: route,
      skill: route-item,
      output_schema: process.process-plan.v1,
      requires: [parse],
    }
  ......

负责工作流确定性逻辑实现的 WorkflowRuntime 拆成几个职责明确的模块：

class WorkflowEngine:
  def __init__(self, *, resources=None, store=None, renderer=None, executor=None, gates=None):
    self.resources = resources or ResourceProvider()
    self.contracts = ContractRegistry(self.resources)
    self.store = store or RuntimeStore()
    self.renderer = renderer or ArtifactRenderer(...)
    self.executor = executor or StepExecutor(self.resources)
    self.gates = gates or ArtifactGates()

ResourceProvider：加载 YAML 工作流定义和 Schema；
ContractRegistry：校验模型提交的 JSON；
RuntimeStore：持久化任务、步骤、Artifact、交付意图；
ArtifactRenderer：生成 Excel / 结构图等文件；
StepExecutor：执行 source-bound 步骤；
ArtifactGates：检查阶段完成条件。

$iii.$ Agent 调用 Workflow 的方式

上面的 Workflow 只暴露一个工具：

def workflow_tool(args: dict[str, Any], **kwargs) -> str:
  runtime = WorkflowRuntime()
  action = args.get("action")
  if action == "start":
    return runtime.start(workflow_id=args["workflow_id"], session_id=args["session_id"], input_data=args["input"])
  elif action == "submit":
    return runtime.submit(task_id=args["task_id"], step_id=args["step_id"], skill=args["skill"], data=args["data"])
  elif action == "submit_from_source":
    return runtime.submit_from_source(task_id=args["task_id"], step_id=args["step_id"], skill=args["skill"])
  elif action == "prepare_delivery":
    return runtime.prepare_delivery(task_id=args["task_id"])

Agent Loop 与 Workflow 的交互流程如下：

用户发起请求，Agent 理解意图后，决定调用 <domain>_process_workflow(action="start", ...)；
WorkflowRuntime 在 SQLite 创建任务和第一个 ready 步骤，返回 next_actions；
Agent 看到 next_actions 里指示当前 ready 步骤是 parse，于是调用 <domain>_process_workflow(action="submit", step_id="parse", ...)；
Runtime 校验模型提交的 JSON Schema，登记 Artifact，激活下游步骤，返回新的 next_actions；
重复步骤 3-4，直到 next_actions 为空且任务到达终态。

每次调用都新建 WorkflowRuntime 实例，状态在 SQLite 中持久化。Agent 平台不需要维护 Workflow 内部状态，只需要记住 task_id 即可。

这里的关键是：Agent 对 Workflow 的”当前状态”是有可见性的。它通过每次 tool call 返回的 next_actions 知道当前哪些步骤是 ready 的、需要提交什么 Schema、是否需要用户确认。但 Agent 不知道阶段之间的推进逻辑——这就是 WorkflowEngine 内部的确定性规则推动的了。

$iv.$ source-bound 步骤

前面的架构设计中提到过：某些确定性的步骤不需要把结果给 Agent、而是直接自己传递 JSON Artifact 到下一个阶段：

# 模型调用
xxx_workflow(
  action="submit_from_source",
  task_id="task_xxx",
  step_id="assign",
  skill="material-assigner"
)

# Runtime 内部行为
upstream_artifact = store.load_artifact(task_id, step_id="material_query")
assignments = StepExecutor.run_adapter(
  adapter="xxx.xxx.v2",
  input=upstream_artifact
)
store.register_artifact(task_id, step_id="assign", payload=assignments)
store.submit_step(task_id, step_id="assign", status="succeeded")

Adapter 相关的设计参见 Agentic Resource 管理。

$b.$ Meta-tool：Agent 动态生成 Plan

这种方式把 Plan 的生成权也交给 Agent。Agent 在运行时发现”接下来这一段路径是确定的”，于是动态生成一个 Plan，交给 Plan Executor 批量执行。

以 nonoka 框架为例，它给 Agent 一个默认的 execute_plan 工具，ReAct 循环在每次请求 LLM 前，把它动态注入可用工具列表：

@staticmethod
def _available_tools(session: Session, blocked: set[str]) -> list[Any]:
  tools = [t for t in session.agent.tools
           if t.name not in blocked and t.name != DYNAMIC_PLAN_CAPABILITY.name]
  if DYNAMIC_PLAN_CAPABILITY.name not in blocked and not getattr(session, "_dynamic_plan_active", False):
    tools.append(DYNAMIC_PLAN_CAPABILITY)
  return tools

Agent 调用 execute_plan 时传入 objective 和 steps：

{
  "tool": "execute_plan",
  "arguments": {
    "objective": "Read config and restart service",
    "steps": [
      {
        "id": "read",
        "tool": "read_file",
        "args": { "path": "/etc/service/config.yaml" }
      },
      {
        "id": "restart",
        "tool": "run_command",
        "args": { "command": { "$ref": "read.content.command" } }
      }
    ]
  }
}

steps 每项必须有 id 和 tool；
args 里可以用 {"$ref": "step_id.path"} 引用前面步骤的结果，例如 {"$ref": "read.content.command"}；
底层会把 $ref 转成 Ref(step_id, path) 对象，与 PlanBuilder 的依赖解析逻辑共享。

DynamicPlanCapability.invoke 拿到调用后，会先做一系列安全校验，然后构建并执行 Plan：

# 安全校验
if getattr(session, "_dynamic_plan_active", False):
  return self._failure("recursion_error", "execute_plan cannot call itself recursively")

known_tools = {tool.name for tool in session.agent.tools}
for raw in steps:
  tool_name = raw["tool"]
  if tool_name not in known_tools:
    raise ValueError(f"unknown tool: {tool_name}")
  if tool_name == self.name:
    raise ValueError("execute_plan cannot be nested")
  capability = next(tool for tool in session.agent.tools if tool.name == tool_name)
  if getattr(capability, "external", False):
    raise ValueError(f"external tool is not supported in dynamic plans: {tool_name}")

# 构建 Plan
args = self._decode_refs(raw.get("args", {}))
builder.step(step_id, tool_name, depends_on=depends_on, **args)
plan = builder.build()

# 执行前后隔离外层 ReAct 会话状态
session._dynamic_plan_active = True
previous_plan = session.current_plan
previous_status = session.status
previous_end_time = session.end_time
try:
  result = await PlanExecutor().execute(plan, session, runner)
finally:
  session._dynamic_plan_active = False
  session.current_plan = previous_plan
  session.status = previous_status
  session.end_time = previous_end_time
  await runner.checkpoint_store.save_session(session.session_id, session.to_state())

执行结果以结构化字典返回给 Agent Loop：

payload = {
  "success": result.success,
  "objective": objective,
  "result": result.data,
  "completed_steps": {k: v.data for k, v in session.completed_steps.items() if k in seen},
}
if not result.success:
  payload.update({
    "error": result.error,
    "error_type": result.error_type,
    "failed_steps": {k: v.model_dump() for k, v in session.failed_steps.items() if k in seen},
  })
return payload

这种方式的耦合程度最高，因为 Agent 不仅要决定是否进入 Workflow，还要决定 Workflow 的内部结构。它的风险是：如果 Agent 生成的 Plan 有语义错误，执行器无法自愈。

在 Agent Loop 中嵌入确定性执行层

1. 相关的设计思想

a.a.a. Anthropic：Workflow 与 Agent 不是二选一

b.b.b. 12-Factor Agents：Agent 主要由确定性代码组成

2. 架构设计

a.a.a. 第一层：Autonomous Agent Loop

b.b.b. 第二层：Workflow Runtime / Plan Executor

c.c.c. 第三层：Bridge Layer

3. 确定性执行层的具体接入与实现

a.a.a. Tool-based

i.i.i. 插件注册逻辑

ii.ii.ii. Workflow 的定义与运行时分发

iii.iii.iii. Agent 调用 Workflow 的方式

iv.iv.iv. source-bound 步骤

b.b.b. Meta-tool：Agent 动态生成 Plan