Agent Workflow Runtime 架构拆解：把 Agent Loop 从提示词搬进代码，长任务才真正稳了

在构建 AI Agent 时，很多开发者习惯把整个任务流程塞进一条提示词里——比如“你是一个客服助手，先理解问题，再检索知识库，然后生成回复，最后记录日志”。这种提示词驱动的 Agent Loop 在简单场景下可行，但随着任务变长、环境变复杂，问题就暴露了：提示词膨胀导致 LLM 理解偏差、状态丢失难以恢复、错误处理全靠模型“猜”。真正的解法，是把 Agent 的执行逻辑从提示词中抽离，搬进代码，用 Workflow Runtime 来接管。本文将从原理层面拆解 Agent Workflow Runtime 架构，并通过可运行代码展示如何构建一个可靠的长任务执行引擎。## 为什么提示词驱动的 Agent Loop 不可靠？先看一个典型的提示词驱动 Agent 示例：python# 提示词驱动的 Agent Loop（不稳定版本）import openaidef agent_loop(user_input): prompt = f""" 你是一个任务助手。请按步骤执行： 1. 解析用户输入 2. 决定动作（search, compute, respond） 3. 执行动作 4. 输出结果 用户输入: {user_input} 请直接输出步骤结果。 """ response = openai.ChatCompletion.create( model="gpt-4", messages=[{"role": "user", "content": prompt}] ) return response.choices[0].message.content# 长任务场景：查询天气并计算温差print(agent_loop("查询北京和上海今天的温度，然后计算温差"))这段代码的问题在于：- 提示词膨胀：每个步骤的细节都要写入提示词，LLM 可能忽略或误解。- 无状态恢复：如果中间调用失败（如 API 超时），整个流程必须从头重试。- 错误处理随机：LLM 可能跳过步骤或输出格式错误，导致下游解析失败。## 核心原理：将 Agent Loop 建模为状态机Agent Workflow Runtime 的核心思想是将 Agent 执行过程建模为有限状态机（FSM）。每个步骤是一个状态，步骤间的转移由代码逻辑控制，而不是由 LLM 猜测。这样，LLM 只负责“思考”，而“执行”和“决策”由 runtime 接管。一个完整的 Workflow Runtime 架构包含以下组件：- Workflow Engine：管理状态机，驱动步骤转移。- Node（节点）：每个状态的具体执行单元，如 InputNode、LLMNode、ToolNode。- Context（上下文）：跨步骤共享的数据存储，支持持久化和恢复。- Error Handler：异常处理机制，支持重试、回退或终止。## 代码实践：构建一个可运行的 Workflow Runtime### 示例 1：基础 Workflow Runtime 实现下面是一个轻量级的 Workflow Runtime，支持定义节点和状态转移：pythonfrom typing import Dict, Any, Optionalfrom abc import ABC, abstractmethod# 1. 定义节点基类class Node(ABC): @abstractmethod def execute(self, context: Dict[str, Any]) -> Optional[str]: """执行节点逻辑，返回下一个节点名称，None表示终止""" pass# 2. 实现具体节点class InputNode(Node): def __init__(self, user_input: str): self.user_input = user_input def execute(self, context): context["user_input"] = self.user_input print(f"[InputNode] 接收输入: {self.user_input}") return "parse_node" # 转移到下一个节点class ParseNode(Node): def execute(self, context): user_input = context["user_input"] # 简单解析：提取关键词 if "天气" in user_input: context["action"] = "weather_query" elif "计算" in user_input: context["action"] = "compute" else: context["action"] = "respond" print(f"[ParseNode] 解析动作: {context['action']}") return "llm_node"class LLMNode(Node): def execute(self, context): action = context["action"] # 模拟 LLM 调用 print(f"[LLMNode] 思考动作: {action}") context["llm_result"] = f"执行{action}的结果" return "output_node"class OutputNode(Node): def execute(self, context): result = context.get("llm_result", "无结果") print(f"[OutputNode] 输出: {result}") return None # 终止流程# 3. 定义 Workflow Engineclass WorkflowEngine: def __init__(self, nodes: Dict[str, Node]): self.nodes = nodes self.context = {} def run(self, start_node: str): current = start_node while current is not None: node = self.nodes.get(current) if not node: raise ValueError(f"未知节点: {current}") try: next_node = node.execute(self.context) current = next_node except Exception as e: print(f"[错误] 节点 {current} 执行失败: {e}") break print("Workflow 完成")# 4. 运行示例if __name__ == "__main__": nodes = { "input_node": InputNode("查询天气"), "parse_node": ParseNode(), "llm_node": LLMNode(), "output_node": OutputNode() } engine = WorkflowEngine(nodes) engine.run("input_node")输出示例：[InputNode] 接收输入: 查询天气[ParseNode] 解析动作: weather_query[LLMNode] 思考动作: weather_query[OutputNode] 输出: 执行weather_query的结果Workflow 完成这个示例展示了如何将 Agent Loop 拆解为明确的节点和转移。但实际长任务中，还需要处理状态持久化和错误恢复。### 示例 2：带状态持久化的 Workflow Runtime长任务可能运行数小时，系统崩溃后需要恢复。下面引入 Checkpoint 机制：pythonimport jsonimport picklefrom typing import Dict, Anyfrom pathlib import Path# 1. 增强节点：支持 checkpointclass CheckpointNode(Node): def __init__(self, node_name: str, next_node: str): self.node_name = node_name self.next_node = next_node def execute(self, context): print(f"[CheckpointNode] 保存上下文到 {self.node_name}.pkl") with open(f"{self.node_name}.pkl", "wb") as f: pickle.dump(context, f) return self.next_node# 2. 支持恢复的 Workflow Engineclass RecoverableWorkflowEngine(WorkflowEngine): def __init__(self, nodes: Dict[str, Node], checkpoint_dir: str = "./checkpoints"): super().__init__(nodes) self.checkpoint_dir = Path(checkpoint_dir) self.checkpoint_dir.mkdir(exist_ok=True) def save_checkpoint(self, node_name: str): filepath = self.checkpoint_dir / f"{node_name}.ckpt" with open(filepath, "wb") as f: pickle.dump({"current_node": node_name, "context": self.context}, f) print(f"[Checkpoint] 已保存: {filepath}") def load_checkpoint(self) -> Optional[str]: # 查找最新的 checkpoint ckpt_files = sorted(self.checkpoint_dir.glob("*.ckpt"), reverse=True) if not ckpt_files: return None with open(ckpt_files[0], "rb") as f: data = pickle.load(f) self.context = data["context"] print(f"[恢复] 从 {ckpt_files[0]} 恢复，节点: {data['current_node']}") return data["current_node"] def run(self, start_node: str, resume: bool = False): if resume: restored_node = self.load_checkpoint() if restored_node: start_node = restored_node current = start_node while current is not None: node = self.nodes.get(current) if not node: raise ValueError(f"未知节点: {current}") try: next_node = node.execute(self.context) # 在每个节点执行后保存 checkpoint self.save_checkpoint(current) current = next_node except Exception as e: print(f"[错误] 节点 {current} 执行失败: {e}") # 错误时保存 checkpoint 以便重试 self.save_checkpoint(current) break print("Workflow 完成")# 3. 模拟长任务：需要外部 API 调用的节点class LongTaskNode(Node): def __init__(self, task_name: str, duration: int = 5): self.task_name = task_name self.duration = duration def execute(self, context): import time print(f"[LongTaskNode] 开始 {self.task_name}，耗时 {self.duration}s...") time.sleep(self.duration) # 模拟长时间执行 context[self.task_name] = "完成" print(f"[LongTaskNode] {self.task_name} 完成") return "checkpoint_node"# 4. 运行带恢复的示例if __name__ == "__main__": nodes = { "input_node": InputNode("执行长任务"), "long_task_1": LongTaskNode("数据收集", duration=3), "checkpoint_node": CheckpointNode("checkpoint_1", "long_task_2"), "long_task_2": LongTaskNode("数据分析", duration=4), "output_node": OutputNode() } engine = RecoverableWorkflowEngine(nodes) # 第一次运行（假设中途崩溃） print("=== 第一次运行 ===") engine.run("input_node") # 模拟崩溃：手动中断后，再次运行 print("\n=== 恢复运行 ===") engine.run("input_node", resume=True)关键点：- CheckpointNode 在每个步骤后保存上下文，允许从任意点恢复。- RecoverableWorkflowEngine 支持 resume 参数，自动加载最近的 checkpoint。- 即使 LongTaskNode 执行中崩溃，恢复后可以从 checkpoint 节点继续。## 架构优势与场景适用性将 Agent Loop 搬进代码带来的核心收益：1. 确定性执行：节点转移由代码控制，避免 LLM 的随机性导致流程偏离。2. 细粒度错误恢复：每个节点可以独立重试或回退，而不影响整个流程。3. 可观测性：通过日志、checkpoint 和 metrics，可以精确追踪每个步骤的状态。4. 可扩展性：新增节点只需继承 Node 并实现 execute，无需修改提示词。典型适用场景包括：- 多步骤 RAG 流程：检索→排序→生成→验证。- 复杂工具编排：调用多个 API 并依赖结果。- 长时间异步任务：如数据爬取、模型训练后处理。## 总结Agent Workflow Runtime 架构的本质，是将 Agent 的“思考”与“执行”解耦。LLM 回归其核心能力——逻辑推理和自然语言理解，而流程控制、状态管理、错误恢复等“基础设施”交给代码。这种设计让长任务不再是 LLM 的负担，而是可预测、可恢复、可调试的工程系统。从提示词到代码的这一步迁移，是 Agent 系统从“玩具”走向“生产”的必经之路。