Prompt 工程与输出控制

Agent 的行为完全由 Prompt 驱动。一个精心设计的 System Prompt 是 Agent 可靠运行的基石。本章深入讲解 Agent Prompt 的设计模式和输出控制技巧。

Agent Prompt 架构

ReAct Prompt 模式

ReAct（Reasoning + Acting）是最经典的 Agent Prompt 框架，让模型交替进行思考和行动。

"""
ReAct Prompt 构建器
"""
from dataclasses import dataclass, field
@dataclass
class ToolSpec:
"""工具规格"""
name: str
description: str
parameters: dict
examples: list[str] = field(default_factory=list)
class ReActPromptBuilder:
"""ReAct Prompt 构建器"""
SYSTEM_TEMPLATE = """你是一个 {role}。
## 能力
{capabilities}
## 可用工具
{tools}
## 工作流程
按以下格式逐步思考和行动：
Thought: 分析当前情况，决定下一步
Action: 工具名称
Action Input: 工具输入参数（JSON 格式）
Observation: 工具返回结果
... （重复 Thought/Action/Observation 直到任务完成）
Thought: 我已经获得足够信息
Final Answer: 最终回答
## 约束
{constraints}"""
def __init__(self, role: str):
self.role = role
self.tools: list[ToolSpec] = []
self.capabilities: list[str] = []
self.constraints: list[str] = []
def add_tool(self, tool: ToolSpec) -> "ReActPromptBuilder":
self.tools.append(tool)
return self
def add_capability(self, capability: str) -> "ReActPromptBuilder":
self.capabilities.append(capability)
return self
def add_constraint(self, constraint: str) -> "ReActPromptBuilder":
self.constraints.append(constraint)
return self
def build(self) -> str:
"""构建完整的 System Prompt"""
tools_text = ""
for tool in self.tools:
tools_text += f"### {tool.name}\n"
tools_text += f"- 描述：{tool.description}\n"
tools_text += f"- 参数：{tool.parameters}\n"
if tool.examples:
tools_text += f"- 示例：{tool.examples[0]}\n"
tools_text += "\n"
return self.SYSTEM_TEMPLATE.format(
role=self.role,
capabilities="\n".join(f"- {c}" for c in self.capabilities),
tools=tools_text,
constraints="\n".join(f"- {c}" for c in self.constraints),
)

结构化输出控制

让 Agent 输出结构化数据（JSON/XML），而不是自由文本，是提高可靠性的关键。

"""
结构化输出解析
"""
import json
import re
from dataclasses import dataclass
@dataclass
class AgentAction:
"""Agent 动作"""
tool: str
input_params: dict
thought: str = ""
@dataclass
class AgentOutput:
"""Agent 输出"""
actions: list[AgentAction]
final_answer: str | None = None
raw_text: str = ""
class OutputParser:
"""Agent 输出解析器"""
# JSON 提取正则
JSON_PATTERN = re.compile(r"```json\s*(.*?)\s*```", re.DOTALL)
def parse_react(self, text: str) -> AgentOutput:
"""解析 ReAct 格式输出"""
actions = []
final_answer = None
lines = text.strip().split("\n")
current_thought = ""
i = 0
while i < len(lines):
line = lines[i].strip()
if line.startswith("Thought:"):
current_thought = line[8:].strip()
elif line.startswith("Action:"):
tool = line[7:].strip()
# 获取 Action Input
if i + 1 < len(lines) and lines[i + 1].strip().startswith("Action Input:"):
input_text = lines[i + 1].strip()[13:].strip()
try:
params = json.loads(input_text)
except json.JSONDecodeError:
params = {"raw": input_text}
actions.append(AgentAction(
tool=tool, input_params=params, thought=current_thought
))
i += 1
elif line.startswith("Final Answer:"):
final_answer = line[13:].strip()
i += 1
return AgentOutput(
actions=actions, final_answer=final_answer, raw_text=text
)
def parse_json(self, text: str) -> AgentOutput:
"""解析 JSON 格式输出"""
# 尝试从 markdown 代码块提取
match = self.JSON_PATTERN.search(text)
json_str = match.group(1) if match else text
try:
data = json.loads(json_str)
actions = [
AgentAction(
tool=a["tool"],
input_params=a.get("params", {}),
thought=a.get("thought", ""),
)
for a in data.get("actions", [])
]
return AgentOutput(
actions=actions,
final_answer=data.get("answer"),
raw_text=text,
)
except (json.JSONDecodeError, KeyError):
return AgentOutput(actions=[], raw_text=text)

Prompt 工程最佳实践

输出修复与重试

"""
输出修复机制
"""
from dataclasses import dataclass
@dataclass
class RepairConfig:
"""修复配置"""
max_retries: int = 3
repair_prompt: str = "你的上次输出格式不正确。请严格按照要求的格式重新输出。"
class OutputRepairer:
"""输出修复器"""
def __init__(self, llm_client, parser: OutputParser, config: RepairConfig | None = None):
self.llm = llm_client
self.parser = parser
self.config = config or RepairConfig()
def parse_with_repair(self, text: str, format_type: str = "react") -> AgentOutput:
"""解析输出，失败时自动修复"""
for attempt in range(self.config.max_retries):
if format_type == "react":
result = self.parser.parse_react(text)
else:
result = self.parser.parse_json(text)
# 验证解析结果
if result.actions or result.final_answer:
return result
# 修复
repair_prompt = (
f"{self.config.repair_prompt}\n\n原始输出：\n{text}\n\n修复后的输出："
)
text = self.llm.generate(repair_prompt)
print(f"  输出修复第 {attempt + 1} 次")
return AgentOutput(actions=[], raw_text=text)

本章小结

下一章：多 Agent 架构模式