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

Prompt 工程在 Agent 工作流中的设计原则

news 2026/6/21 20:13:57

Prompt 工程在 Agent 工作流中的设计原则

一、Prompt 稳定性的核心挑战

上周还能正常输出结构化分析的 Prompt，这周可能就开始自由发挥。问题往往不在模型本身——虽然模型更新确实会影响表现——而在于 Prompt 设计缺乏工程化思维。自然语言指令天生存在模糊性，生产环境需要的却是确定性行为。

Agent 工作流放大了这个问题。单轮对话中，Prompt 偏差可以通过人工调整修正；但在多步流程里，每个环节的误差都会累积放大。这就要求我们把 Prompt 设计从"写一句话"提升到"构建状态机"的层面。

二、Agent 工作流的 Prompt 架构设计

graph TD subgraph Prompt 模板层 A[系统角色 Prompt] --> B[任务指令 Prompt] B --> C[输出格式 Prompt] C --> D[约束与安全 Prompt] end subgraph 运行时编排层 D --> E[上下文窗口管理器] E --> F[多步状态机] F --> G[工具调用路由] G --> H[输出解析与验证] end subgraph 反馈优化层 H --> I[输出质量评估] I --> J[自动重试与降级] J --> K[Prompt 版本追踪] K --> L[A/B 测试与迭代] end L --> A

系统角色 Prompt 需要明确定义 Agent 的行为边界：允许的操作范围、禁止的行为类型、不确定信息的处理方式。每个约束都应具备可测试性。

多步状态机是工作流的核心。Agent 通过"规划-执行-观察-反思"的循环推进任务，每步 Prompt 必须携带完整状态信息：已完成步骤、当前观察结果、下一步计划。

上下文管理容易被忽视。对话历史快速膨胀时，需要采用摘要压缩早期内容、保留最近 N 轮完整对话、截断工具结果等策略。

三、生产级 Prompt 编排实现

class AgentState(Enum): PLANNING = "planning" # 分析任务，制定步骤 EXECUTING = "executing" # 调用工具，获取信息 OBSERVING = "observing" # 分析执行结果 REFLECTING = "reflecting" # 判断是否需要调整 RESPONDING = "responding" # 生成最终答案 FAILED = "failed" # 无法完成任务 class PromptTemplate: def __init__(self, name, version, template, required_vars=None): self.name = name self.version = version self.template = template self.required_vars = required_vars or [] def render(self, **kwargs): missing = set(self.required_vars) - set(kwargs.keys()) if missing: raise ValueError(f"缺少变量: {missing}") return self.template.format(**kwargs) class ContextWindowManager: def manage(self, messages, max_tokens=8000): # 按优先级保留内容：系统指令 > 最近对话 > 工具结果 > 早期对话 system_msgs = [m for m in messages if m.get("role") == "system"] chat_msgs = [m for m in messages if m.get("role") in ["user", "assistant"]] tool_msgs = [m for m in messages if m.get("tool_call_id")] # 截断长工具结果 truncated_tools = [] for m in tool_msgs: content = m["content"] if len(content) > 500: truncated_tools.append({ **m, "content": content[:300] + "\n[...已截断...]" }) else: truncated_tools.append(m) # 保留最近 3 轮完整对话 recent_chat = chat_msgs[-6:] if len(chat_msgs) > 6 else chat_msgs early_chat = chat_msgs[:-6] if len(chat_msgs) > 6 else [] # 压缩早期对话为摘要 summary_msg = None if early_chat: summary_text = " | ".join(m["content"][:50] for m in early_chat) summary_msg = { "role": "system", "content": f"[早期对话摘要] {summary_text}" } result = system_msgs[:] if summary_msg: result.append(summary_msg) result.extend(truncated_tools) result.extend(recent_chat) return result