162 lines
6.7 KiB
Python
162 lines
6.7 KiB
Python
|
"""Evolution Engine — 从历史轨迹中学习, 反哺各智能体 prompt, 实现自进化。
|
||
|
|
|
||
|
|
闭环(自进化):
|
||
|
|
轨迹采集 → 奖励打分 → 归纳成功/失败模式 → 生成进化建议 →
|
||
|
|
反哺/追加到各 agent 的系统提示词(可选启用)
|
||
|
|
|
||
|
|
Level 1 实现:
|
||
|
|
- analyze: 从已保存轨迹中总结高分/低分模式(常见工具序列、步骤分布)
|
||
|
|
- generate_insights: 生成可读的进化洞察(markdown)
|
||
|
|
- evolve_prompt: 将洞察追加到对应 agent 的 system prompt(可选)
|
||
|
|
|
||
|
|
轻量自进化的核心: 不编造"强化学习训练", 而是做"基于经验的 prompt 增强"
|
||
|
|
——把历史中高奖励轨迹用过的工具调用序列、成功策略, 有选择地注入
|
||
|
|
未来 agent 的提示词, 让系统越用越聪明。
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
from collections import Counter
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from .reward import compute_reward_v1
|
||
|
|
from .trajectory import TrajectoryRecorder
|
||
|
|
|
||
|
|
|
||
|
|
class EvolutionEngine:
|
||
|
|
"""自进化引擎(Level 1 轻量版)。"""
|
||
|
|
|
||
|
|
def __init__(
|
||
|
|
self,
|
||
|
|
base_dir: Path,
|
||
|
|
recorder: TrajectoryRecorder,
|
||
|
|
enabled: bool = True,
|
||
|
|
):
|
||
|
|
self._base_dir = Path(base_dir)
|
||
|
|
self._recorder = recorder
|
||
|
|
self.enabled = enabled
|
||
|
|
self._insights_dir = self._base_dir / "xingbang_evolution" / "insights"
|
||
|
|
self._insights_dir.mkdir(parents=True, exist_ok=True)
|
||
|
|
|
||
|
|
# ── 分析历史轨迹 ────────────────────────────────────
|
||
|
|
def _load_all_trajectories(self) -> list[dict[str, Any]]:
|
||
|
|
trajs = []
|
||
|
|
for p in self._recorder.list_trajectories():
|
||
|
|
try:
|
||
|
|
trajs.append(json.loads(p.read_text(encoding="utf-8")))
|
||
|
|
except Exception:
|
||
|
|
continue
|
||
|
|
return trajs
|
||
|
|
|
||
|
|
def analyze(self) -> dict[str, Any]:
|
||
|
|
"""分析全部历史轨迹, 返回统计与模式。"""
|
||
|
|
trajectories = self._load_all_trajectories()
|
||
|
|
if not trajectories:
|
||
|
|
return {"count": 0, "message": "尚无轨迹样本,先积累数据"}
|
||
|
|
|
||
|
|
# 计算每条奖励
|
||
|
|
scored = []
|
||
|
|
for t in trajectories:
|
||
|
|
r = compute_reward_v1(t)
|
||
|
|
t["_reward"] = r
|
||
|
|
scored.append(t)
|
||
|
|
|
||
|
|
scored.sort(key=lambda x: x["_reward"], reverse=True)
|
||
|
|
high = scored[: max(1, len(scored) // 4)] # 前 25%
|
||
|
|
low = scored[-max(1, len(scored) // 4):] # 后 25%
|
||
|
|
|
||
|
|
# 归纳高频工具序列
|
||
|
|
def agent_tool_sequences(trajs):
|
||
|
|
counter = Counter()
|
||
|
|
for t in trajs:
|
||
|
|
for s in t.get("steps", []):
|
||
|
|
seq = tuple(s.get("tools", []))
|
||
|
|
if seq:
|
||
|
|
counter[seq] += 1
|
||
|
|
return counter.most_common(5)
|
||
|
|
|
||
|
|
high_patterns = {
|
||
|
|
"seqs": agent_tool_sequences(high),
|
||
|
|
"avg_steps": sum(len(t["steps"]) for t in high) / max(1, len(high)),
|
||
|
|
"success_rate": sum(1 for t in high if t["final_result"].get("success"))
|
||
|
|
/ max(1, len(high)),
|
||
|
|
}
|
||
|
|
return {
|
||
|
|
"count": len(trajectories),
|
||
|
|
"avg_reward": round(sum(t["_reward"] for t in scored) / len(scored), 2),
|
||
|
|
"high": {"samples": len(high), "patterns": high_patterns},
|
||
|
|
"low": {
|
||
|
|
"samples": len(low),
|
||
|
|
"avg_steps": sum(len(t["steps"]) for t in low) / max(1, len(low)),
|
||
|
|
},
|
||
|
|
"top_agents_used": self._agent_usage(scored),
|
||
|
|
}
|
||
|
|
|
||
|
|
def _agent_usage(self, trajs) -> dict:
|
||
|
|
counter = Counter()
|
||
|
|
for t in trajs:
|
||
|
|
for s in t.get("steps", []):
|
||
|
|
counter[s.get("agent", "?")] += 1
|
||
|
|
return dict(counter.most_common())
|
||
|
|
|
||
|
|
# ── 生成进化洞察 ────────────────────────────────────
|
||
|
|
def generate_insights(self) -> str:
|
||
|
|
"""生成可读的 markdown 进化洞察报告。"""
|
||
|
|
analysis = self.analyze()
|
||
|
|
if analysis.get("count", 0) == 0:
|
||
|
|
return "# 自进化洞察\n\n尚无轨迹样本。运行审批流水线积累数据后,系统将自动进化。"
|
||
|
|
lines = [
|
||
|
|
f"# 兴邦 AI 评审·自进化洞察",
|
||
|
|
f"\n**样本数**: {analysis['count']} 条轨迹 | **平均奖励**: {analysis['avg_reward']}",
|
||
|
|
"",
|
||
|
|
"## ✅ 高分轨迹模式 (前 25%)",
|
||
|
|
]
|
||
|
|
hp = analysis["high"]
|
||
|
|
lines.append(f"- 平均步骤数: {hp['patterns']['avg_steps']:.1f}")
|
||
|
|
lines.append(f"- 成功率: {hp['patterns']['success_rate']*100:.0f}%")
|
||
|
|
lines.append("- 高频工具序列:")
|
||
|
|
for seq, n in hp["patterns"]["seqs"]:
|
||
|
|
lines.append(f" - {' → '.join(seq)} (出现 {n} 次)")
|
||
|
|
lines += ["", "## ⚠️ 低分轨迹 (后 25%)", f"- 平均步骤数: {analysis['low']['avg_steps']:.1f}"]
|
||
|
|
lines += ["", "## 🧠 智能体使用分布", ""]
|
||
|
|
for agent, count in analysis.get("top_agents_used", {}).items():
|
||
|
|
lines.append(f"- {agent}: {count} 次")
|
||
|
|
lines += [
|
||
|
|
"",
|
||
|
|
"## 进化建议",
|
||
|
|
"1. 高分轨迹的工具序列可在各 agent 的 system prompt 中作为'推荐工作流'注入",
|
||
|
|
"2. 步骤数偏多(>5)的 agent 提示词应强调'一次到位,减少反复读取'",
|
||
|
|
"3. 持续积累轨迹 → 达到一定量后启动更高阶优化(LoRA 微调)",
|
||
|
|
]
|
||
|
|
return "\n".join(lines)
|
||
|
|
|
||
|
|
def save_insights(self) -> Path:
|
||
|
|
"""保存洞察报告到 insights 目录。"""
|
||
|
|
report = self.generate_insights()
|
||
|
|
path = self._insights_dir / "evolution_insights.md"
|
||
|
|
path.write_text(report, encoding="utf-8")
|
||
|
|
return path
|
||
|
|
|
||
|
|
# ── 反哺 prompt (自进化闭环) ────────────────────────
|
||
|
|
def evolve_prompt_suggestion(self, agent_key: str) -> str | None:
|
||
|
|
"""为指定 agent 生成可注入的 prompt 增强建议(基于历史高分轨迹)。
|
||
|
|
|
||
|
|
返回一段可追加到该 agent system_prompt 的文本, 若无历史则 None。
|
||
|
|
"""
|
||
|
|
analysis = self.analyze()
|
||
|
|
if analysis.get("count", 0) == 0:
|
||
|
|
return None
|
||
|
|
hp = analysis["high"]
|
||
|
|
seqs = hp["patterns"]["seqs"]
|
||
|
|
if not seqs:
|
||
|
|
return None
|
||
|
|
top_seq_text = " → ".join(seqs[0][0]) if seqs else ""
|
||
|
|
suggestion = (
|
||
|
|
f"\n\n[自进化推荐 · 基于 {analysis['count']} 条历史轨迹] "
|
||
|
|
f"高分任务常用工具序列: {top_seq_text}。"
|
||
|
|
f"请优先复用该模式,减少无效工具往返,一次到位完成任务。"
|
||
|
|
)
|
||
|
|
return suggestion
|