# agent_evolution.py # 依赖: pip install openai tiktoken

import json
import os
import copy
import hashlib
from datetime import datetime
from dataclasses import dataclass, field
from typing import Optional
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

============ 数据模型 ============

@dataclass
class ExecutionTrace:
"""单次执行轨迹"""
task: str
prompt_snapshot: dict # 执行时的提示词快照
tool_calls: list # 工具调用记录
result: str # 执行结果
quality_score: float = 0.0 # 质量评分 0-10
timestamp: str = field(default_factory=lambda: datetime.now().isoformat())

@dataclass
class HarnessState:
"""Agent外壳状态"""
system_prompt: str
tool_descriptions: dict # 工具名 -> 描述文本
memory_strategy: str # 记忆管理策略描述
guardrails: list[str] # 护栏规则列表

def snapshot(self) -> dict: """生成当前状态快照""" return { "system_prompt": self.system_prompt, "tool_descriptions": copy.deepcopy(self.tool_descriptions), "memory_strategy": self.memory_strategy, "guardrails": list(self.guardrails) } def diff_hash(self) -> str: """计算状态指纹，用于追踪变更""" state_str = json.dumps(self.snapshot(), sort_keys=True) return hashlib.md5(state_str.encode()).hexdigest()[:12]

@dataclass
class ImprovementRecord:
"""一条改进记录"""
task_type: str
original_trace: ExecutionTrace
improved_trace: ExecutionTrace
change_description: str
component_modified: str # 被修改的组件名
score_delta: float # 分数变化

4.2 执行引擎：跑任务、记录轨迹

class ExecutionEngine: """Agent执行引擎"""
def __init__(self, harness: HarnessState): self.harness = harness def build_messages(self, task: str) -> list[dict]: """构建完整的消息上下文""" messages = [{"role": "system", "content": self.harness.system_prompt}] # 注入护栏规则 if self.harness.guardrails: guardrail_text = "\n".join(f"- {r}" for r in self.harness.guardrails) messages[0]["content"] += f"\n\n## 安全护栏\n{guardrail_text}" # 注入工具描述 if self.harness.tool_descriptions: tool_text = "\n".join( f"### {name}\n{desc}" for name, desc in self.harness.tool_descriptions.items() ) messages[0]["content"] += f"\n\n## 可用工具\n{tool_text}" # 注入记忆策略提示 messages[0]["content"] += f"\n\n## 记忆管理\n{self.harness.memory_strategy}" messages.append({"role": "user", "content": task}) return messages def execute(self, task: str) -> ExecutionTrace: """执行一个任务并记录完整轨迹""" messages = self.build_messages(task) # 调用模型 response = client.chat.completions.create( model="gpt-4o", messages=messages, temperature=0.3, max_tokens=4096 ) result = response.choices[0].message.content # 构造轨迹 trace = ExecutionTrace( task=task, prompt_snapshot=self.harness.snapshot(), tool_calls=self._extract_tool_calls(response), result=result ) return trace def _extract_tool_calls(self, response) -> list: """从响应中提取工具调用记录""" if response.choices[0].message.tool_calls: return [ { "name": tc.function.name, "arguments": tc.function.arguments } for tc in response.choices[0].message.tool_calls ] return []

4.3 评分器：给执行结果打分

这里是关键——我们需要一个自动化的质量评估机制。

论文里用的是LLM-as-Judge，我也用了类似方案：

class QualityJudge: """LLM质量评分器"""
JUDGE_PROMPT = """你是一个严格的任务质量评估专家。

请根据以下维度对Agent的执行结果打分（每个维度1-10分）：

1. 任务完成度：是否完成了用户要求的事情？
2. 准确性：信息是否正确、无幻觉？
3. 效率：工具调用是否合理？有没有多余步骤？
4. 安全性：是否遵守了护栏规则？

请严格按以下JSON格式输出：
{
"completion": <1-10>,
"accuracy": <1-10>,
"efficiency": <1-10>,
"safety": <1-10>,
"overall": <加权平均>,
"issues": ["具体问题列表"],
"suggestions": ["改进建议"]
}"""

def score(self, task: str, result: str, trace: ExecutionTrace) -> float: """对执行结果打分""" eval_input = f"""任务: {task}

Agent执行结果:
{result}

工具调用记录:
{json.dumps(trace.tool_calls, ensure_ascii=False, indent=2)}
"""
response = client.chat.completions.create(
model="gpt-4o-mini", # 用小模型评分，省成本
messages=[
{"role": "system", "content": self.JUDGE_PROMPT},
{"role": "user", "content": eval_input}
],
temperature=0.1,
response_format={"type": "json_object"}
)

try: scores = json.loads(response.choices[0].message.content) return scores.get("overall", 5.0), scores except json.JSONDecodeError: return 5.0, {"issues": ["评分解析失败"]}

4.4 进化引擎：核心模块——分析轨迹、提出改进

这是整套系统最核心的部分。

class EvolutionEngine: """Agent进化引擎 - Harness2核心实现"""
ANALYZER_PROMPT = """你是一个Agent行为分析专家。你的工作是分析Agent的执行轨迹，找出外壳（prompt/tools/guardrails）的问题，并提出精确的修改方案。

分析框架

1. 行为诊断：从执行轨迹中找出具体哪些行为导致了低分
2. 根因定位：这个问题是提示词不清晰？工具描述有误导？护栏太松或太紧？
3. 修改方案：提出最小化、精确的修改，不要大改

重要约束

• 每次只修改1个组件
• 修改内容必须能直接替换原文（给出完整的旧文本和新文本）
• 不要添加模糊的"请注意xxx"类指令，要给出具体可操作的规则
• 不要删除现有的有效规则，只做增量或精确修改

输出格式（严格JSON）

{
"diagnosis": "问题诊断",
"root_cause": "根因分析",
"target_component": "要修改的组件名(system_prompt/tool:<name>/memory_strategy/guardrails)",
"old_text": "要被替换的原文片段",
"new_text": "替换后的新文本",
"expected_improvement": "预期改善效果"
}"""

def __init__(self, harness: HarnessState, improvement_history: list): self.harness = harness self.history = improvement_history def analyze_and_propose(self, trace: ExecutionTrace, scores: dict) -> Optional[dict]: """分析一次失败执行，提出修改方案""" analysis_input = f"""## 当前外壳状态

系统提示词:
{self.harness.system_prompt}

工具描述:
{json.dumps(self.harness.tool_descriptions, ensure_ascii=False, indent=2)}

记忆策略:
{self.harness.memory_strategy}

护栏规则:
{json.dumps(self.harness.guardrails, ensure_ascii=False, indent=2)}

执行轨迹

任务: {trace.task}

Agent输出:
{trace.result}

工具调用:
{json.dumps(trace.tool_calls, ensure_ascii=False, indent=2)}

评分结果

{json.dumps(scores, ensure_ascii=False, indent=2)}

历史改进记录（最近5条）

{json.dumps(self.history[-5:], ensure_ascii=False, indent=2) if self.history else "无历史记录"}"""

response = client.chat.completions.create( model="gpt-4o", messages=[ {"role": "system", "content": self.ANALYZER_PROMPT}, {"role": "user", "content": analysis_input} ], temperature=0.2, response_format={"type": "json_object"} ) try: proposal = json.loads(response.choices[0].message.content) return proposal except json.JSONDecodeError: return None def apply_modification(self, proposal: dict) -> HarnessState: """将修改方案应用到外壳上""" new_harness = copy.deepcopy(self.harness) target = proposal["target_component"] old_text = proposal["old_text"] new_text = proposal["new_text"] if target == "system_prompt": if old_text in new_harness.system_prompt: new_harness.system_prompt = new_harness.system_prompt.replace( old_text, new_text ) else: new_harness.system_prompt += f"\n\n{new_text}" elif target.startswith("tool:"): tool_name = target[5:] if tool_name in new_harness.tool_descriptions: if old_text in new_harness.tool_descriptions[tool_name]: new_harness.tool_descriptions[tool_name] = ( new_harness.tool_descriptions[tool_name].replace(old_text, new_text) ) else: new_harness.tool_descriptions[tool_name] += f"\n{new_text}" elif target == "memory_strategy": if old_text in new_harness.memory_strategy: new_harness.memory_strategy = new_harness.memory_strategy.replace( old_text, new_text ) else: new_harness.memory_strategy += f"\n{new_text}" elif target == "guardrails": new_harness.guardrails.append(new_text) return new_harness

4.5 验证器：A/B对比，确认改动有效

改完了不能直接用，得验证。

class ValidationEngine: """A/B验证引擎"""
def __init__(self, original_harness: HarnessState, test_tasks: list[str]): self.original_harness = original_harness self.test_tasks = test_tasks def validate( self, proposed_harness: HarnessState, proposal: dict ) -> tuple[bool, float]: """ 对比原始外壳和改进后外壳在测试任务集上的表现 返回: (是否采纳, 平均分数差) """ original_engine = ExecutionEngine(self.original_harness) improved_engine = ExecutionEngine(proposed_harness) judge = QualityJudge() original_scores = [] improved_scores = [] for task in self.test_tasks: orig_trace = original_engine.execute(task) orig_score, _ = judge.score(task, orig_trace.result, orig_trace) original_scores.append(orig_score) impr_trace = improved_engine.execute(task) impr_score, _ = judge.score(task, impr_trace.result, impr_trace) improved_scores.append(impr_score) avg_original = sum(original_scores) / len(original_scores) avg_improved = sum(improved_scores) / len(improved_scores) delta = avg_improved - avg_original # 采纳阈值：至少提升0.5分 return delta >= 0.5, delta

4.6 经验库：跨任务知识积累

这是Harness²区别于一次性优化的关键。

class ExperienceStore: """经验知识库"""
def __init__(self, store_path: str = "experience_store.json"): self.store_path = store_path self.experiences = self._load() def _load(self) -> list: if os.path.exists(self.store_path): with open(self.store_path, "r") as f: return json.load(f) return [] def save(self): with open(self.store_path, "w") as f: json.dump(self.experiences, f, ensure_ascii=False, indent=2) def record(self, record: ImprovementRecord): """记录一条改进经验""" self.experiences.append({ "task_type": record.task_type, "component": record.component_modified, "change_desc": record.change_description, "score_delta": record.score_delta, "timestamp": record.improved_trace.timestamp }) self.save() def get_relevant(self, task_type: str) -> list: """获取与当前任务类型相关的历史经验""" return [ exp for exp in self.experiences if exp["task_type"] == task_type ]

4.7 主控循环：把所有模块串起来

class AgentEvolutionLoop: """Agent自进化主循环"""
def __init__( self, initial_harness: HarnessState, test_tasks: list[str], evolution_tasks: list[str], max_rounds: int = 5 ): self.harness = initial_harness self.test_tasks = test_tasks self.evolution_tasks = evolution_tasks self.max_rounds = max_rounds self.experience_store = ExperienceStore() self.improvement_history = [] self.round_log = [] def run(self): """执行进化循环""" print(f"初始外壳指纹: {self.harness.diff_hash()}") print(f"计划进化轮数: {self.max_rounds}") print("=" * 60) for round_num in range(self.max_rounds): print(f"\n{'='*20} 第 {round_num + 1} 轮进化 {'='*20}") # Step 1: 在进化任务上执行，找到最差的case engine = ExecutionEngine(self.harness) judge = QualityJudge() worst_trace = None worst_score = float("inf") worst_scores = None for task in self.evolution_tasks: trace = engine.execute(task) score, scores = judge.score(task, trace.result, trace) if score < worst_score: worst_score = score worst_trace = trace worst_scores = scores print(f"最差任务得分: {worst_score:.1f}") print(f"诊断: {worst_scores.get('issues', [])[:2]}") if worst_score >= 8.5: print("所有任务得分 >= 8.5，进化完成！") break # Step 2: 分析并提出改进方案 evolution = EvolutionEngine(self.harness, self.improvement_history) proposal = evolution.analyze_and_propose(worst_trace, worst_scores) if not proposal: print("分析失败，跳过本轮") continue print(f"诊断: {proposal.get('diagnosis', 'N/A')}") print(f"修改组件: {proposal.get('target_component', 'N/A')}") # Step 3: 应用修改 proposed_harness = evolution.apply_modification(proposal) print(f"新外壳指纹: {proposed_harness.diff_hash()}") # Step 4: A/B验证 validator = ValidationEngine(self.harness, self.test_tasks[:3]) approved, delta = validator.validate(proposed_harness, proposal) print(f"验证结果: {'通过' if approved else '未通过'} (delta={delta:+.2f})") if approved: old_hash = self.harness.diff_hash() self.harness = proposed_harness record = ImprovementRecord( task_type="general", original_trace=worst_trace, improved_trace=worst_trace, change_description=proposal.get("change_description", ""), component_modified=proposal.get("target_component", ""), score_delta=delta ) self.experience_store.record(record) self.improvement_history.append(proposal) print(f"[OK] 采纳改进: {old_hash} -> {self.harness.diff_hash()}") else: print(f"[X] 改进无效，保持原外壳") self.round_log.append({ "round": round_num + 1, "score_before": worst_score, "delta": delta, "approved": approved, "component": proposal.get("target_component", "") }) print(f"\n{'='*60}") print("进化报告:") print(f"总改进次数: {len(self.improvement_history)}") for entry in self.round_log: status = "OK" if entry["approved"] else "X" print(f" 第{entry['round']}轮: {status} delta={entry['delta']:+.2f} ({entry['component']})") return self.harness

4.8 实际运行

if __name__ == "__main__": # 初始化外壳 initial_harness = HarnessState( system_prompt="""你是一个智能客服助手，负责回答用户关于产品使用的问题。 请用简洁、准确的语言回答。如果不确定，请明确告知用户。""", tool_descriptions={ "search_docs": "搜索产品文档，参数: query(str)", "create_ticket": "创建工单，参数: title(str), description(str), priority(int)" }, memory_strategy="保留最近5轮对话作为上下文", guardrails=[ "不编造不存在的产品功能", "遇到投诉必须升级处理" ] )
test_tasks = [ "你们的API支持批量操作吗？", "我上传文件一直报错，怎么办？", "怎么把数据导出成Excel？", ] evolution_tasks = [ "你们支不支持GraphQL？文档里没找到", "上次那个bug修了没？上周提交工单到现在没回音", "能不能帮我写个Python脚本来调用你们的API？", "你们和竞品XX比有什么优势？", "我想定制一个工作流，先审核再发布，能搞吗？", ] loop = AgentEvolutionLoop( initial_harness=initial_harness, test_tasks=test_tasks, evolution_tasks=evolution_tasks, max_rounds=3 ) final_harness = loop.run() print("\n最终系统提示词:") print(final_harness.system_prompt)

五、实测效果：3轮进化后发生了什么？

我在上面那个客服Agent上跑了3轮进化，记录一下关键变化。

第1轮：工具描述优化

问题：Agent在回答"怎么调用API"时，反复告诉用户"请参考文档"，没有给出任何实质帮助。

诊断：工具描述太简略，Agent不知道工具能返回什么。

改动：在search_docs的描述中增加了返回格式说明和调用示例。

原始: "搜索产品文档，参数: query(str)" 改进: "搜索产品文档。参数: query(str)。返回匹配文档片段列表，每条包含title、url、content。调用示例: search_docs(query='API批量操作')"

效果：delta=+1.2分。Agent开始直接返回文档片段内容，而不是笼统地说"请查文档"。

第2轮：护栏规则细化

问题：遇到用户对比竞品的问题时，Agent要么回避，要么给出不准确的对比信息。

诊断：护栏规则"不编造不存在的产品功能"太笼统，没有覆盖竞品对比场景。

改动：新增护栏规则——"被问到与竞品对比时，只陈述自身产品的已知能力，不做优劣判断，建议用户参考对比页面"。

效果：delta=+0.8分。回答从"嗯...这个不太好说"变成了"关于XX功能，我们目前支持...建议您参考官网的对比说明"。

第3轮：记忆策略调整

问题：多轮对话中，Agent忘了用户之前提过的上下文，重复问一样的问题。

诊断："保留最近5轮对话"策略对长任务不够用，用户描述了一个复杂问题后，前因在第6轮就被截断了。

改动：将记忆策略改为"保留最近10轮对话 + 关键实体摘要（用户提到的产品名、错误码、功能模块）"。

效果：delta=+0.6分。多轮对话的连贯性明显改善。

三轮累计提升：从6.2分到8.8分。

六、几个实际踩过的坑

写到这里，分享几个在实现过程中的真实体会。

1. 编辑器模型的选择很关键

论文里用的是Claude Opus 4.8当编辑器，我在实验中发现，编辑器的分析能力直接决定了改进方案的质量。用gpt-4o-mini当编辑器时，提的修改方案经常是"请注意xxx"这种模糊指令，基本没用。换gpt-4o之后，方案质量明显好了很多。

2. A/B验证不能省

我一开始没做验证，直接应用所有改进。结果有一轮改进把系统提示词改得太长了（加了300多字的补充说明），虽然对当前任务有效，但导致了其他任务的响应变慢、注意力分散。

加了验证之后，这种"拆东墙补西墙"的改进就被拦住了。

3. 经验库的实际价值比想象中大

刚开始我觉得经验库就是个日志，后来发现——当进化引擎能参考历史经验时，它提出的方案明显更有针对性。比如第2轮的护栏改进，引擎参考了第1轮的工具描述优化经验后，直接给出了精确的新增规则，而不是泛泛地说"需要加强护栏"。

4. 成本控制

每一轮进化大约消耗：

• 进化任务执行：5次 x ~2K tokens = ~10K tokens
• 评分：5次 x ~1.5K tokens = ~7.5K tokens
• 分析：1次 x ~4K tokens = ~4K tokens
• A/B验证：3次 x 2（原始+改进）x ~2K tokens = ~12K tokens

一轮下来大约3.5万tokens，按gpt-4o的价格大概$0.5。3轮进化$1.5。

对比微调动辄几百美元，这个成本低到可以忽略。

这也是我在实际工作中选择这条路线的原因——之前在帮一些企业客户做Agent效果优化时（我们团队在这块积累了一些经验，有兴趣的可以了解下黑箭科技在AI应用安全方面的实践），最大的阻碍就是"调优成本太高"。Harness²这条路线把成本降了一个数量级，对中小团队非常友好。

七、这套系统适合什么场景？

不是所有Agent都需要自进化。我总结了一下适合的场景：

适合的场景：

• Agent已经上线运行，但某些场景表现不稳定
• 没有足够数据做微调，或者微调周期太长
• 任务类型相对固定（客服、文档问答、代码Review等）
• 需要持续迭代但不想每次都手动调prompt

不太适合的场景：

• 任务类型变化很大，没有规律可循
• 对确定性要求极高（比如金融交易决策）
• 外壳已经很完善，改进空间很小

另外有个安全问题必须提一下：让Agent自己修改自己的提示词和工具描述，本质上是在给它改执行环境的权限。一定要做好沙箱隔离，限制修改范围，不能让Agent修改到不该碰的地方。Harness²论文自己也说了——"这不是Google官方支持的产品"。安全方面得自己兜着。

八、总结

回到开头那个让我震惊的数字：11.5分。

这个实验告诉我们一个反直觉的结论——Agent的能力上限，不取决于模型有多强，而取决于外壳有多好。

就像一个人，给他正确的地图和工具，哪怕天赋普通，也能比一个空有天赋但没有装备的人走得更远。

Harness²的意义不只是提了个新方法，而是把"Agent自优化"这件事从学术论文推到了可运行的开源代码。400行Python就能搭出核心框架，3轮进化就能把客服Agent从6.2分拉到8.8分。

成本？不到2美元。

这篇文章的代码是完整的，可以直接跑。如果你正在为Agent效果不好而头疼，与其花大价钱换模型，不如先试试改改它的"外壳"——也许问题根本不在脑子，而在穿的那件衣服上。

参考资料：

1. Google Research. Harness2: Recursive Agent Harnessing for an Open World. arXiv, 2026.10
2. Reinventing AI. AI Agents Trends: The Mid-October 2026 Open-Source Release Wave. 2026.10.20
3. 36Kr. 不改权重也能变强：谷歌开源Harness2. 2026.10.19

你平时调Agent效果，是换模型多还是调prompt多？有没有试过让Agent自己优化自己的提示词？评论区聊聊你的经验。