当前位置：首页 > news >正文

Prompt Engineering 与 Agent 工作流：从单次调用到自主决策的编排架构

news 2026/6/25 12:08:03

Prompt Engineering 与 Agent 工作流：从单次调用到自主决策的编排架构

一、单次 Prompt 的局限

大语言模型单次调用有个硬伤：只能根据当前输入生成一次结果，没法主动说“我需要更多数据”或者“这问题得拆成小任务”。简单问答时问题不大，但处理复杂任务时容易出三种问题。

第一，信息不够瞎编。比如用户让规划京都三天行程，模型只能靠训练数据里的通用信息编行程，查不到实时机票价格、酒店空房或天气。这种“规划”本质是编故事。第二，复杂任务容易漏步骤。像“分析财报并给投资建议”这种任务，要提取数据、算指标、做行业对比、推导结论。模型单次调用时往往直接甩结论，跳过中间步骤，推理过程没法验证。第三，错了一次全错。如果第一步算错数，后面所有推理都建立在错误基础上，模型自己还发现不了。

Agent 工作流就是把单次调用拆成多步循环，每一步的输出当下一步的输入，关键节点还能调工具、自我反思。Prompt Engineering 就是控制这个循环的“编程语言”——靠设计好的提示词模板，让模型在合适时机做对决定。

二、Agent 工作流的控制流架构

生产级的 Agent 工作流不是简单“循环调用 LLM”，而是规划、执行、观察、反思四个阶段组成的闭环系统。

graph TB INPUT[用户任务输入] --> PLAN[规划阶段：任务分解与工具选择] subgraph 执行循环 PLAN --> EXEC[执行阶段：调用工具/生成中间结果] EXEC --> OBS[观察阶段：解析工具返回/评估结果质量] OBS --> REFLECT{反思阶段：结果是否满足目标？} REFLECT -->|不满足：需要更多信息| RETRY[调整策略：换工具/改参数/补充上下文] RETRY --> EXEC REFLECT -->|不满足：任务需要拆分| SUB[子任务分解：递归进入子循环] SUB --> PLAN REFLECT -->|满足| DONE[输出最终结果] end DONE --> VERIFY[结果验证：格式检查/事实核验] VERIFY --> OUTPUT[结构化输出] subgraph 上下文管理 CM[上下文窗口管理器] MEM[长期记忆存储] TOOL[工具注册表] end CM -.-> PLAN CM -.-> EXEC CM -.-> REFLECT MEM -.-> PLAN TOOL -.-> EXEC

规划阶段是 Agent 和单次调用的核心区别。模型得把任务拆成可执行的子步骤，给每个步骤选工具，还要理清步骤间的依赖关系和执行顺序。Prompt 模板不仅要说明任务目标，还得提供工具规格和选择策略。

执行阶段调外部工具拿实时数据或执行操作。工具类型包括搜索、代码执行、数据库查询、API 调用。每个工具的输入输出格式得在工具注册表里定义清楚，模型按定义生成调用指令。

观察阶段解析工具返回的结果，评估质量。搜索结果为空？说明查询词得改。代码执行报错？说明生成的代码有问题。观察阶段输出结构化的“观察摘要”，包含结果内容、质量评估和异常标记。

反思阶段是 Agent 的“元认知”能力。模型得判断当前结果是否满足原始目标，不满足的话决定是重试（换工具/改参数）、拆分（进一步分解步骤）还是放弃（标记无法完成并说明原因）。反思能力直接决定 Agent 可靠性——没反思的 Agent 要么死循环，要么输出烂结果。

三、Agent 工作流的核心编排代码

下面这段代码实现了基于 ReAct 模式的 Agent 工作流编排器，包含规划、执行、观察、反思四个阶段的完整逻辑。

from dataclasses import dataclass, field from enum import Enum from typing import Any, Callable, Optional from datetime import datetime import json class StepStatus(Enum): """步骤执行状态""" PENDING = "pending" RUNNING = "running" SUCCESS = "success" FAILED = "failed" SKIPPED = "skipped" class ToolType(Enum): """工具类型枚举""" SEARCH = "search" CODE = "code_interpreter" SQL = "sql_executor" API = "rest_client" LLM = "llm_call" @dataclass class ToolDefinition: """工具定义——注册表中的工具规格说明""" name: str tool_type: ToolType description: str # 工具的功能描述，供模型选择时参考 parameters_schema: dict # JSON Schema 格式的参数定义 handler: Optional[Callable] = None # 实际执行函数 @dataclass class PlanStep: """规划步骤——任务分解后的单个子步骤""" step_id: str description: str tool_name: Optional[str] = None # 使用的工具名称，None 表示纯 LLM 推理 tool_params: dict = field(default_factory=dict) depends_on: list[str] = field(default_factory=list) # 依赖的前置步骤 ID status: StepStatus = StepStatus.PENDING result: Any = None error: Optional[str] = None retry_count: int = 0 max_retries: int = 2 @dataclass class AgentState: """Agent 运行状态——上下文管理器的核心数据结构""" task: str # 原始任务描述 steps: list[PlanStep] = field(default_factory=list) observations: list[dict] = field(default_factory=list) reflections: list[dict] = field(default_factory=list) total_tokens: int = 0 max_iterations: int = 10 # 防止无限循环的安全阈值 current_iteration: int = 0 class ToolRegistry: """工具注册表——管理所有可用工具的定义与执行""" def __init__(self): self._tools: dict[str, ToolDefinition] = {} def register(self, tool: ToolDefinition) -> None: """注册一个工具""" self._tools[tool.name] = tool def get(self, name: str) -> Optional[ToolDefinition]: """按名称获取工具定义""" return self._tools.get(name) def get_tool_descriptions(self) -> str: """生成供 Prompt 使用的工具描述文本""" descriptions = [] for tool in self._tools.values(): param_desc = json.dumps(tool.parameters_schema, ensure_ascii=False) descriptions.append( f"- {tool.name}（{tool.tool_type.value}）：{tool.description}\n" f" 参数格式：{param_desc}" ) return "\n".join(descriptions) async def execute(self, tool_name: str, params: dict) -> dict: """执行指定工具""" tool = self._tools.get(tool_name) if not tool: return {"success": False, "error": f"工具 {tool_name} 未注册"} if not tool.handler: return {"success": False, "error": f"工具 {tool_name} 无执行函数"} try: result = await tool.handler(**params) return {"success": True, "data": result} except Exception as e: return {"success": False, "error": str(e)} class Planner: """规划器——将任务分解为可执行的步骤序列""" PLAN_PROMPT_TEMPLATE = """你是一个任务规划专家。请将以下任务分解为可执行的步骤。 可用工具： {tool_descriptions} 任务：{task} 请输出 JSON 格式的步骤列表，每个步骤包含： - step_id：步骤编号（如 s1, s2） - description：步骤描述 - tool_name：使用的工具名称（纯推理步骤填 null） - tool_params：工具参数（纯推理步骤填 {{}}） - depends_on：依赖的前置步骤 ID 列表 注意： 1. 每个步骤应该是原子性的，只完成一个明确的子目标 2. 需要实时信息的步骤必须使用搜索工具 3. 需要计算或代码执行的步骤必须使用代码执行工具 4. 步骤之间应有明确的依赖关系，避免并行执行导致信息缺失 输出格式： ```json {{"steps": [...]}} ```""" async def plan(self, task: str, registry: ToolRegistry) -> list[PlanStep]: """根据任务描述生成执行计划""" prompt = self.PLAN_PROMPT_TEMPLATE.format( tool_descriptions=registry.get_tool_descriptions(), task=task, ) # 生产环境中调用 LLM 生成计划 # 此处展示解析逻辑框架 steps = self._parse_plan_response("placeholder") return steps def _parse_plan_response(self, response: str) -> list[PlanStep]: """解析 LLM 返回的计划 JSON""" # 生产环境中需处理 JSON 解析异常和格式校验 return [] class Reflector: """反思器——评估执行结果并决定后续策略""" REFLECT_PROMPT_TEMPLATE = """请评估当前步骤的执行结果是否满足目标。 原始任务：{task} 当前步骤：{step_description} 执行结果：{result} 结果状态：{status} 请判断： 1. 结果是否满足当前步骤的目标？（satisfied / unsatisfied） 2. 如果不满足，原因是什么？ 3. 建议的下一步行动： - retry：换参数重试当前步骤 - retool：换一个工具执行 - decompose：将当前步骤进一步拆分 - abort：标记任务无法完成 输出 JSON 格式： ```json {{"satisfied": true/false, "reason": "...", "action": "...", "suggestion": "..."}} ```""" def reflect( self, task: str, step: PlanStep, ) -> dict: """对执行结果进行反思评估""" # 简化实现：基于结果状态判断 if step.status == StepStatus.SUCCESS and step.result is not None: # 检查结果是否为空或无效 if isinstance(step.result, str) and len(step.result.strip()) == 0: return { "satisfied": False, "reason": "执行结果为空", "action": "retry", "suggestion": "尝试修改查询参数", } return {"satisfied": True, "reason": "步骤目标已达成"} if step.status == StepStatus.FAILED: if step.retry_count < step.max_retries: return { "satisfied": False, "reason": f"执行失败：{step.error}", "action": "retry", "suggestion": "检查参数格式并重试", } return { "satisfied": False, "reason": f"重试 {step.retry_count} 次后仍失败", "action": "retool", "suggestion": "尝试使用其他工具或调整策略", } return {"satisfied": False, "reason": "未知状态", "action": "retry"} class ContextWindowManager: """上下文窗口管理器——控制注入 LLM 的上下文内容与大小""" def __init__(self, max_tokens: int = 8000): self.max_tokens = max_tokens def build_execution_context( self, state: AgentState, current_step: PlanStep, ) -> str: """为当前步骤构建执行上下文""" context_parts = [f"原始任务：{state.task}\n"] # 添加已完成步骤的结果摘要 completed_steps = [ s for s in state.steps if s.status == StepStatus.SUCCESS and s.step_id in current_step.depends_on ] if completed_steps: context_parts.append("前置步骤结果：") for step in completed_steps: # 截断过长的结果，保留关键信息 result_str = str(step.result) if len(result_str) > 500: result_str = result_str[:500] + "...[已截断]" context_parts.append(f" {step.description}：{result_str}") # 添加反思历史 if state.reflections: recent_reflections = state.reflections[-3:] # 只保留最近3条反思 context_parts.append("历史反思：") for r in recent_reflections: context_parts.append(f" - {r.get('reason', '')}") context = "\n".join(context_parts) # 粗略估算 token 数，超限时截断 estimated_tokens = len(context) * 1.5 if estimated_tokens > self.max_tokens: # 优先保留任务描述和前置步骤结果，截断反思历史 context_parts = context_parts[:2 + len(completed_steps)] context = "\n".join(context_parts) return context class AgentOrchestrator: """Agent 编排器——串联规划、执行、观察、反思四个阶段""" def __init__(self, registry: ToolRegistry): self.registry = registry self.planner = Planner() self.reflector = Reflector() self.context_manager = ContextWindowManager() async def run(self, task: str) -> dict: """执行完整的 Agent 工作流""" state = AgentState(task=task) # Phase 1: 规划 state.steps = await self.planner.plan(task, self.registry) if not state.steps: return {"success": False, "error": "无法生成执行计划"} # Phase 2: 执行循环 for step in state.steps: state.current_iteration += 1 if state.current_iteration > state.max_iterations: return { "success": False, "error": f"超过最大迭代次数 {state.max_iterations}", "partial_results": self._collect_results(state), } # 等待依赖步骤完成 if not self._check_dependencies(step, state): step.status = StepStatus.SKIPPED step.error = "前置依赖未满足" continue # 执行当前步骤 await self._execute_step(step, state) # 反思与重试 if step.status != StepStatus.SUCCESS: reflection = self.reflector.reflect(task, step) state.reflections.append(reflection) if reflection["action"] == "retry" and step.retry_count < step.max_retries: step.retry_count += 1 step.status = StepStatus.PENDING await self._execute_step(step, state) # Phase 3: 结果汇总 results = self._collect_results(state) return { "success": all(s.status == StepStatus.SUCCESS for s in state.steps), "steps": len(state.steps), "completed": sum(1 for s in state.steps if s.status == StepStatus.SUCCESS), "results": results, "total_iterations": state.current_iteration, } async def _execute_step(self, step: PlanStep, state: AgentState) -> None: """执行单个步骤""" step.status = StepStatus.RUNNING if step.tool_name: # 调用外部工具 result = await self.registry.execute(step.tool_name, step.tool_params) if result["success"]: step.result = result["data"] step.status = StepStatus.SUCCESS else: step.error = result["error"] step.status = StepStatus.FAILED else: # 纯 LLM 推理步骤 context = self.context_manager.build_execution_context(state, step) # 生产环境中调用 LLM 生成推理结果 step.result = f"[LLM 推理] {step.description}" step.status = StepStatus.SUCCESS # 记录观察结果 state.observations.append({ "step_id": step.step_id, "status": step.status.value, "result_preview": str(step.result)[:200] if step.result else None, "timestamp": datetime.now().isoformat(), }) def _check_dependencies(self, step: PlanStep, state: AgentState) -> bool: """检查步骤的前置依赖是否已满足""" step_map = {s.step_id: s for s in state.steps} for dep_id in step.depends_on: dep_step = step_map.get(dep_id) if not dep_step or dep_step.status != StepStatus.SUCCESS: return False return True def _collect_results(self, state: AgentState) -> dict: """收集所有步骤的执行结果""" return { step.step_id: { "description": step.description, "status": step.status.value, "result": step.result, } for step in state.steps }

这段代码有三个关键设计点值得注意。

规划与执行分离。Planner在任务开始时一次性生成完整步骤序列，而不是每步执行时动态决定下一步。这种“先规划后执行”比“边做边想”更可控——执行前能审查计划合理性，避免 Agent 陷入无意义循环。代价是灵活性降低——执行中发现计划不合理时，需要额外反思机制调整策略。

依赖关系图。每个PlanStep通过depends_on字段声明前置依赖，编排器执行前检查依赖是否满足。这保证了信息流正确性——算毛利率的步骤必须在提取财务数据步骤之后执行。依赖没满足时，步骤会被跳过而不是强行执行，避免基于缺失信息产生错误结果。

上下文窗口管理。ContextWindowManager控制注入 LLM 的上下文大小，避免超出 Token 限制。策略是优先保留任务描述和前置步骤结果，截断反思历史。因为前置步骤结果是当前步骤的必要输入，反思历史只是辅助参考。

四、Agent 工作流的工程代价

Agent 工作流落地时面临三个核心 Trade-offs，它们相互关联且难以同时优化。

延迟的链式累积。一个 5 步的 Agent 任务，每步含 1 次 LLM 调用和 1 次工具调用，总延迟大约：规划（2s）+ 5 * (LLM 推理 1.5s + 工具调用 0.5s + 反思 1s) = 13.5s。如果某步需要重试，延迟可能超 20s。对实时交互场景来说，这个延迟不可接受。优化方向包括：并行执行无依赖步骤、用更快模型（如 GPT-4o-mini）处理简单步骤、缓存工具调用结果。但并行执行会增加上下文管理复杂度，更快模型可能降低推理质量，缓存可能导致数据不一致。

成本的指数增长。Agent 工作流每步都是一次独立 LLM 调用，加上规划和反思阶段的额外调用，5 步任务可能消耗 10-15 次 LLM 调用。按 GPT-4 定价算，单次任务成本约 $0.3-0.5，是单次调用的 10 倍以上。更隐蔽的成本是“无效重试”——反思器判断需要重试，但重试后结果没改善，白白消耗 Token。建议设置总 Token 预算上限，超预算后强制终止并返回已有结果。

可靠性的不确定性。Agent 工作流的可靠性是各步骤可靠性的乘积。如果每步成功率 95%，5 步串联后整体成功率仅 77%。这意味着近四分之一任务会部分失败。提升可靠性手段包括：增加重试次数、引入人工审核节点、用更保守规划策略（减少步骤数量但增加每步确定性）。但这些手段都会增加延迟或成本。

适用边界：Agent 工作流适合需要多步推理和外部工具调用的复杂任务，比如数据分析报告生成、多源信息整合、自动化运维操作。不适合简单问答或单步生成任务——这些场景下单次 LLM 调用更高效、更可靠、更便宜。

五、总结

Agent 工作流的核心价值是把大模型从“单次响应器”变成“自主决策者”，通过规划-执行-观察-反思的闭环循环，实现复杂任务的多步推理和工具调用。Prompt Engineering 在这个架构里扮演“编程语言”角色——靠设计好的提示词模板，控制模型在规划阶段分解任务、执行阶段选择工具、反思阶段评估结果。

落地路线建议：

先从单工具 Agent 起步。初期只注册一个搜索工具，验证规划-执行-反思闭环是否通畅。多工具选择逻辑比单工具复杂得多，过早引入会增加调试难度。

规划阶段用更强模型。规划质量直接决定后续执行效率，建议规划阶段用 GPT-4 级别模型，执行和反思阶段用更快模型，平衡质量与成本。

设置硬性安全阈值。包括最大迭代次数、总 Token 预算和单步超时时间。Agent 工作流最大风险是无限循环，安全阈值是不可省略的防护措施。

记录完整执行轨迹。每一步的输入、输出、工具调用和反思结果都应持久化存储。这不仅用于调试，更是评估 Agent 可靠性和优化 Prompt 模板的数据基础。

渐进式引入反思能力。初期可以跳过反思阶段，直接按计划顺序执行。反思能力需要精心设计的 Prompt 模板和足够测试数据，过早引入可能降低而非提升可靠性。

所做更改总结：