self-evolution-eval¶
第8章 · Agent 的自我进化 · 配套项目
chapter8/self-evolution-eval
项目说明¶
实验 8-6：为自我进化 Agent 设计评估数据集（★★★）¶
评估一个 Agent 的"自我进化"能力——即在没有现成工具时，自己去发现、创造并复用工具—— 需要一套专门的评估数据集与验证方法。难点在于：任务不能暗示工具名（否则退化成"记忆固定工具模式"）， 而且"结果对不对"只是最表层的信号，还要看它怎么发现、造得好不好、下次会不会复用。
本目录给出一套可运行的配套实现：20 个跨领域任务的数据集 + 四层分层验证 harness + 一个可控参考 Agent + 一键演示。
目录结构¶
| 文件 | 说明 |
|---|---|
dataset.json |
20 个不同领域的工具需求任务。每条含 目标描述（不暗示工具名）、参考方案（推荐库 + API 示例）、已知陷阱（废弃库 / 需付费注册的 API）、正确性判据。 |
harness.py |
四层验证 harness：FourLayerEvaluator.evaluate(task, trajectory, variant_trajectory)。 |
agent.py |
参考被测 Agent（可控 mock 版自我进化 Agent）+ ToolRegistry 工具注册表。 |
demo.py |
一键演示：python demo.py。 |
config.py |
读取 API Key、构造 OpenAI 兼容客户端。 |
requirements.txt / env.example |
依赖与环境变量示例。 |
数据集设计原则¶
- 只说目标、不暗示工具名。例如"获取某 YouTube 视频字幕"而非"用
youtube-transcript-api"； "查询加密货币实时价格趋势"而非"用 CoinGecko API"。这样才能真正考察 Agent 的发现/创造能力， 而不是它对某个库名的记忆。 - 20 个不同领域：多媒体、金融/加密货币、科学计算、地理编码、社交媒体、IoT、天气、NLP、图像、 PDF、天文、化学、生物信息、音频、汇率、股票、地理空间、RSS、二维码、时区。领域越分散，越能避免 模型套用固定模式。
- 每条任务都附 参考方案（
reference_solution：推荐开源库列表 + 典型 API 示例）与 已知陷阱（known_pitfalls：deprecated_libraries废弃库、paid_or_registration_apis需付费/注册的 API）， 供第 2、3 层做判定依据。 correctness_criteria给出第 1 层的可核对判据（正则或关键词）；discovery_keywords给出第 2 层 判断"搜索是否切题"的关键词；tool_name+variant_goal支撑第 4 层"第二次相似任务是否复用"。mock_answer仅用于驱动本仓库的可控参考 Agent 跑通 harness，真实被测 Agent 不依赖它。
四层分层验证¶
harness 输入一条被测 Agent 的运行轨迹（trajectory：工具调用序列 + 创造的工具代码 + 最终答案；
schema 见 agent.py 顶部注释），输出四层各自分数与总评：
| 层 | 名称 | 方法 | 判据来源 |
|---|---|---|---|
| L1 | 任务正确性 | 规则/判据核对最终答案 | correctness_criteria |
| L2 | 工具发现有效性 | 启发式分析搜索关键词 / 是否访问网页 / 选了哪个库 | discovery_keywords + reference_solution + known_pitfalls |
| L3 | 工具创造质量 | LLM-as-a-Judge 按 Rubric 打分（错误处理 / 参数校验 / 文档 / 健壮性，各 0-3） | 被测 Agent 创造的工具代码 |
| L4 | 工具复用能力 | 分析"第二次相似任务"轨迹：是否直接检索已注册工具而非重复搜索创建 | variant_trajectory 的动作序列 |
- L2 是纯启发式（无需 LLM）：选中推荐库(0.40) + 搜索切题(0.25) + 避开陷阱(0.25) + 访问网页(0.10)。 选了废弃库/付费 API 会被判"未避开陷阱"。
- L3 是唯一必须调用 LLM 的层：把工具函数代码交给 judge，按 4 维 Rubric 返回 JSON 分数与中文点评。
- L4 通过第二次相似任务（
variant_goal）的动作序列区分：retrieve_tool（复用）vssearch+create_tool（重复劳动）。 - 某层不适用（如复用轨迹不产生新工具，则 L2/L3 记 N/A）时，总评在可用层间按权重重新归一。
如何用 harness 评估你自己的 Agent¶
让你的被测 Agent 产出符合 agent.py 中 schema 的轨迹（steps / created_tools / final_answer），
第二次相似任务复用同一个 ToolRegistry，然后：
from harness import FourLayerEvaluator
evaluator = FourLayerEvaluator(judge_model="gpt-5.6-luna") # 默认四层全跑
# 只跑确定性层（不需要联网）：evaluator = FourLayerEvaluator(layers=("L1","L2","L4"))
report = evaluator.evaluate(task, first_trajectory, variant_trajectory)
print(report["layers"], report["summary"]["overall"])
layers= 用于选择实际运行哪些层——只有 L3 需要联网调用 LLM，去掉 L3 即可完全离线评估。
未选中的层记 score=None（N/A），总评在可用层间按权重重新归一。
运行¶
pip install -r requirements.txt
cp env.example .env # 填入 OPENAI_API_KEY（默认 provider=openai, 模型 gpt-5.6-luna）
python demo.py # 默认：strong 跑 3 个任务 + weak 对照 1 个（联网，含 L3）
python demo.py --quick # 快速演示：strong / weak 各只跑 1 个任务，省时省钱
python demo.py --tasks task-01,task-07 # 指定要评估的任务 id
python demo.py --help # 查看全部参数（中文说明）
完整参数（python demo.py --help）：
| 参数 | 作用 |
|---|---|
--all |
评估全部 20 个任务（默认自动切到结果表输出，不逐条刷屏） |
--tasks IDS |
逗号分隔的任务 id，指定评估哪几条 |
--quick |
strong / weak 各只跑 1 个任务 |
--layers L1,L2,L4 |
选择运行哪些验证层（仅 L3 需联网，去掉即可离线） |
--profile {strong,weak,both} |
选择被测参考画像；缺省保留默认（strong 全部 + weak 第一个） |
--offline |
离线模式：不调用任何 LLM（strong 用离线工具模板），自动跳过 L3 |
--provider {openai,moonshot,ark} |
覆盖供应商 |
--agent-model / --judge-model |
覆盖造工具模型 / L3 裁判模型 |
--table |
只打印"每任务 × 每层"结果表，不打印逐任务详报 |
--output PATH |
把完整评分结果（含各层明细）写出为 JSON |
demo.py 会：打印数据集概览与几条不暗示工具名的任务示例 → 用 strong 参考 Agent 跑完
四层验证 → 用 weak 参考 Agent 做对照 → 最后打印一张 每任务 × 每层结果表，横向对比各任务在四层上的得分。
离线运行输出（无需 API Key，python demo.py --all --offline --profile both）¶
只跑三个确定性层（L1/L2/L4，L3 记 N/A），可完整复现，用来展示"每任务 × 每层"结果表与 strong/weak 区分度：
每任务 × 每层 结果表（N/A = 该层不适用或未选择）
------------------------------------------------------------------------------
任务 领域 画像 L1 L2 L3 L4 总评
task-01 多媒体 strong 1.000 1.000 N/A 1.000 1.000
task-02 金融数据 / 加密货币 strong 1.000 1.000 N/A 1.000 1.000
...
task-10 文档处理 strong 1.000 0.750 N/A 1.000 0.917
task-19 编码 / 二维码 strong 1.000 0.750 N/A 1.000 0.917
...
task-01 多媒体 weak 1.000 0.000 N/A 0.000 0.467
task-03 科学计算 weak 1.000 0.250 N/A 0.000 0.550
task-10 文档处理 weak 1.000 0.400 N/A 0.000 0.600
...
strong 在 L2（发现）/L4（复用）上普遍满分，weak 因选了废弃/付费库且从不复用而显著更低；两者 L1 都可能为 1
（碰巧答对）——正说明"结果正确"不足以评判自我进化能力。L3（工具创造质量）需联网调用 LLM 裁判，
去掉 --offline 并配置 API Key 后即可补上（见下方联网输出）。
联网运行输出（含 L3，节选）¶
strong（好发现 + LLM 生成的高质量工具 + 复用）：
■ 任务 task-01 (多媒体) | 画像=strong
L1 任务正确性 : 1.000
L2 工具发现有效性 : 1.000 | 选中推荐库=True 避开陷阱=True（选库：['youtube-transcript-api']）
L3 工具创造质量 : 1.000 | Rubric 4 维合计 12/12
Rubric: 错误处理=3 参数校验=3 文档=3 健壮性=3
LLM-Judge 点评: 代码在错误处理、参数校验、文档完整性和健壮性方面表现优秀……
L4 工具复用能力 : 1.000 | 直接检索并复用已注册工具（未重复搜索/创建）
>> 总评 overall : 1.000
[复用探针] 第二次相似任务的动作序列: ['retrieve_tool', 'call_tool', 'final_answer']
weak（坏发现：选了废弃库 pytube + 粗糙 stub + 从不复用）：
■ 任务 task-01 (多媒体) | 画像=weak
L1 任务正确性 : 1.000
L2 工具发现有效性 : 0.000 | 选中推荐库=False 避开陷阱=False（选库：['pytube（字幕/caption 功能长期失效）']）
L3 工具创造质量 : 0.000 | Rubric 4 维合计 0/12
LLM-Judge 点评: 代码缺乏错误处理、参数校验和文档说明，且实现不符合任务目标。
L4 工具复用能力 : 0.000 | 未复用，重复了搜索与工具创建
>> 总评 overall : 0.350
[复用探针] 第二次相似任务的动作序列: ['search', 'select_library', 'create_tool', 'register_tool', 'call_tool', 'final_answer']
两个画像在 L2/L3/L4 上被清晰区分；注意 weak 的 L1 仍可能为 1（碰巧答对），正说明"结果正确"不足以 评判自我进化能力，必须分层看发现 / 创造 / 复用。
配置说明 / 如何适配¶
- 换模型：
AGENT_MODEL（被测 Agent 造工具）、JUDGE_MODEL（第 3 层裁判，默认gpt-5.6-luna）。 - 换供应商 / 网关：默认
PROVIDER=openai，读OPENAI_API_KEY；也支持PROVIDER=moonshot（MOONSHOT_API_KEY） 或PROVIDER=ark（ARK_API_KEY），会自动切换 base_url 与默认模型（见config.py）。 - 换任务 / 输入：编辑
dataset.json新增任务（保持"只说目标、不暗示工具名"原则），或用--tasks task-xx,...指定评估哪几条；把你自己 Agent 的轨迹按agent.py顶部 schema 喂给FourLayerEvaluator.evaluate即可评估真实 Agent。 - 统一兜底：若所选 provider 的 Key 缺失，但设置了
OPENROUTER_API_KEY，会自动改走 OpenRouter， 并把模型名映射到openai/gpt-5.6-luna/anthropic/claude-opus-4.8等（见config.py）。
局限¶
- 内置的
SelfEvolutionAgent是可控参考 Agent（mock 版），用于把四层 harness 跑通并展示 strong/weak 的区分度， 并非真实联网的强 Agent；L1 的"碰巧答对"正是用来说明"结果正确不足以评判自我进化能力"。 - L3 依赖 LLM-as-a-Judge，分数会随裁判模型与采样有小幅波动；L2/L4 为可解释的启发式，判据写死在 harness 中。
--offline模式下 strong 画像用离线工具模板（而非真调 LLM 生成）造工具，因此 L3 无法离线运行（记 N/A）； 它用来在无 API Key 时确定性地复现 L1/L2/L4 三层与结果表。要评估真实的工具创造质量仍需联网跑 L3。- 数据集为 20 条教学规模样本，覆盖面广但每领域仅 1 条，重在方法论演示而非统计显著性。
源代码¶
agent.py¶
"""
参考被测 Agent（可控的最小版"自我进化"Agent）。
它不是要做出真实联网的强 Agent，而是一个"可控 mock"：能按不同"画像(profile)"
产出真实 / 半真实的运行轨迹（trajectory），用来把四层验证 harness 跑通并展示区分度。
关键点：
- 工具"创造"步骤是真实的：strong 画像会真的调用 LLM 生成工具代码，供第 3 层
LLM-as-a-Judge 打分；weak 画像给一段粗糙 stub 以展示低分。
- 工具"发现"步骤按画像产出好 / 坏的搜索关键词与选库，供第 2 层启发式判定。
- 工具"复用"由共享的 ToolRegistry 支撑：strong 画像在第二次相似任务时会先查注册表，
命中即直接检索复用（不再搜索）；weak 画像永远重新搜索与重建，供第 4 层区分。
轨迹(trajectory) schema：
{
"task_id": str,
"goal": str,
"profile": str,
"steps": [ {"action": "...", ...}, ... ], # 见下方各 action
"created_tools": [ {"name": str, "code": str} ],
"final_answer": str
}
step 的 action 取值：
search {"action":"search","query":str}
read_web {"action":"read_web","url":str}
select_library{"action":"select_library","library":str}
create_tool {"action":"create_tool","name":str,"code":str}
register_tool {"action":"register_tool","name":str}
retrieve_tool {"action":"retrieve_tool","name":str,"source":"registry"}
call_tool {"action":"call_tool","name":str,"args":dict,"result":str}
final_answer {"action":"final_answer","text":str}
"""
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from config import Config
# ---------------------------------------------------------------------------
# 工具注册表：自我进化 Agent 把创造出来的工具持久化于此，供后续任务复用
# ---------------------------------------------------------------------------
class ToolRegistry:
def __init__(self):
self._tools: Dict[str, dict] = {}
def has(self, name: str) -> bool:
return name in self._tools
def register(self, name: str, code: str, task_id: str):
self._tools[name] = {"code": code, "task_id": task_id}
def get(self, name: str) -> Optional[dict]:
return self._tools.get(name)
def names(self) -> List[str]:
return list(self._tools)
# ---------------------------------------------------------------------------
# Agent 画像：把"好 / 坏"的行为参数化
# ---------------------------------------------------------------------------
@dataclass
class Profile:
name: str
discovery_quality: str # "good" | "bad"
tool_quality: str # "good"(调 LLM 生成) | "sloppy"(粗糙 stub)
reuse_registry: bool # 相似任务是否先查注册表复用
STRONG = Profile("strong", discovery_quality="good", tool_quality="good", reuse_registry=True)
WEAK = Profile("weak", discovery_quality="bad", tool_quality="sloppy", reuse_registry=False)
_TOOL_GEN_SYSTEM = (
"You are a senior Python engineer building a reusable utility tool. "
"Return ONLY a single self-contained Python function (plus its imports). "
"The function MUST have: a clear docstring (purpose, args, returns, raises), "
"input validation, and try/except error handling with helpful messages. "
"No example usage, no markdown fences, code only."
)
def _sloppy_tool_code(tool_name: str, library: str) -> str:
"""weak 画像使用的粗糙 stub：无 docstring、无校验、无错误处理。"""
top = library.split("(")[0].split(">=")[0].strip().replace("-", "_")
return (
f"def {tool_name}(x):\n"
f" import {top or 'requests'}\n"
f" return {top or 'requests'}.run(x)\n"
)
def _offline_good_tool_code(task: dict, library: str) -> str:
"""离线模式下 strong 画像使用的高质量工具模板（有 docstring / 参数校验 / try-except），
无需调用 LLM，便于在没有 API Key 时演示 L1/L2/L4 三个确定性层的评分。
模板刻意做到 L3 Rubric 的四个维度都齐备，因此即使联网跑 L3 也应得高分。"""
top = library.split("(")[0].split(">=")[0].strip().replace("-", "_") or "requests"
name = task["tool_name"]
goal = task["goal"].replace('"', "'")
return (
f"import {top}\n\n\n"
f"def {name}(query: str, timeout: int = 30):\n"
f' """{goal}\n\n'
f" Args:\n"
f" query: 目标标识（如 URL / ID / 查询串），非空字符串。\n"
f" timeout: 网络请求超时时间（秒），必须为正整数。\n"
f" Returns:\n"
f" 工具执行结果。\n"
f" Raises:\n"
f" ValueError: 入参非法时抛出。\n"
f" RuntimeError: 底层调用失败时抛出。\n"
f' """\n'
f" if not isinstance(query, str) or not query.strip():\n"
f" raise ValueError('query 必须为非空字符串')\n"
f" if not isinstance(timeout, int) or timeout <= 0:\n"
f" raise ValueError('timeout 必须为正整数')\n"
f" try:\n"
f" return {top}.run(query, timeout=timeout)\n"
f" except Exception as exc: # noqa: BLE001\n"
f" raise RuntimeError(f'{name} 执行失败: {{exc}}') from exc\n"
)
class SelfEvolutionAgent:
"""可控的自我进化 Agent。同一个 registry 在多次 run 之间共享以支持复用。"""
def __init__(self, registry: ToolRegistry, model: Optional[str] = None, offline: bool = False):
self.registry = registry
self.model = Config.resolve_default_model(model)
self.offline = offline # True 时用离线工具模板，不调用 LLM（用于无 Key 演示确定性层）
self._client = None # 懒加载，只有真正需要生成工具时才建连接
@property
def client(self):
if self._client is None:
self._client = Config.get_client()
return self._client
# -- 真实调用 LLM 生成工具代码（第 3 层 judge 的输入来源） --------------
def _generate_tool_code(self, task: dict, library: str) -> str:
prompt = (
f"Write a Python function named `{task['tool_name']}` that accomplishes "
f"this goal:\n{task['goal']}\n\n"
f"Prefer using the library `{library}`."
)
resp = self.client.chat.completions.create(
model=self.model,
temperature=Config.TEMPERATURE,
messages=[
{"role": "system", "content": _TOOL_GEN_SYSTEM},
{"role": "user", "content": prompt},
],
)
code = resp.choices[0].message.content or ""
# 去掉可能出现的 markdown 代码围栏
code = code.strip()
if code.startswith("```"):
code = code.split("```", 2)[1]
if code.startswith("python"):
code = code[len("python"):]
code = code.strip("`").strip()
return code
# -- 发现阶段：产出搜索关键词、访问网页、选库 -------------------------
def _discovery_steps(self, task: dict, profile: Profile):
steps = []
if profile.discovery_quality == "good":
kws = task.get("discovery_keywords", [])[:3]
query = " ".join(kws) if kws else task["goal"]
steps.append({"action": "search", "query": f"{query} python library"})
lib = task["reference_solution"]["libraries"][0]
top = lib.split("(")[0].strip()
steps.append({"action": "read_web", "url": f"https://pypi.org/project/{top}/"})
steps.append({"action": "select_library", "library": lib})
return steps, lib
else:
# 坏发现：关键词泛泛、且选了一个已废弃/需付费的库
steps.append({"action": "search", "query": "how to do this quickly easy python"})
pit = task.get("known_pitfalls", {})
bad = (pit.get("deprecated_libraries") or pit.get("paid_or_registration_apis") or ["requests"])[0]
steps.append({"action": "select_library", "library": bad})
return steps, bad
# -- 主流程 ----------------------------------------------------------
def run(self, task: dict, profile: Profile, use_variant: bool = False) -> dict:
"""跑一个任务，返回轨迹。use_variant=True 表示这是"第二次相似任务"（复用探针）。"""
goal = task["variant_goal"] if use_variant else task["goal"]
tool_name = task["tool_name"]
traj = {
"task_id": task["id"],
"goal": goal,
"profile": profile.name,
"is_variant": use_variant,
"steps": [],
"created_tools": [],
"final_answer": "",
}
# 1) 复用检查：strong 画像先查注册表
if profile.reuse_registry and self.registry.has(tool_name):
traj["steps"].append({"action": "retrieve_tool", "name": tool_name, "source": "registry"})
traj["steps"].append({
"action": "call_tool", "name": tool_name, "args": {"goal": goal},
"result": "(复用已注册工具，直接得到结果)",
})
traj["final_answer"] = task["mock_answer"]
traj["steps"].append({"action": "final_answer", "text": traj["final_answer"]})
return traj
# 2) 发现阶段
disc_steps, library = self._discovery_steps(task, profile)
traj["steps"].extend(disc_steps)
# 3) 创造阶段
if profile.tool_quality == "good":
code = (
_offline_good_tool_code(task, library)
if self.offline
else self._generate_tool_code(task, library)
)
else:
code = _sloppy_tool_code(tool_name, library)
traj["steps"].append({"action": "create_tool", "name": tool_name, "code": code})
traj["created_tools"].append({"name": tool_name, "code": code})
# 4) 注册（供复用）
self.registry.register(tool_name, code, task["id"])
traj["steps"].append({"action": "register_tool", "name": tool_name})
# 5) 调用并给出答案
traj["steps"].append({
"action": "call_tool", "name": tool_name, "args": {"goal": goal},
"result": "(工具执行完成)",
})
traj["final_answer"] = task["mock_answer"]
traj["steps"].append({"action": "final_answer", "text": traj["final_answer"]})
return traj
config.py¶
"""
实验 8-6 配置模块：统一读取 API Key、构造 OpenAI 客户端。
支持以下 OpenAI 兼容服务（按 PROVIDER 选择）：
- openai （默认，读 OPENAI_API_KEY，默认模型 gpt-5.6-luna）
- moonshot （读 MOONSHOT_API_KEY，Kimi，默认 kimi-k3）
- ark （读 ARK_API_KEY，火山方舟）
统一的 OpenRouter 兜底（fallback）：
若所选 provider 自己的 Key 缺失，但设置了 OPENROUTER_API_KEY，则自动改走
OpenRouter（https://openrouter.ai/api/v1），并把模型名映射到 OpenRouter 命名：
gpt-* -> openai/gpt-*
claude-* -> anthropic/claude-opus-4.8
含 "/" -> 原样透传
其它 -> openai/gpt-5.6-luna
这样在没有 OpenAI 直连 Key 时也能一键跑通。
"""
import os
from typing import Optional
from dotenv import load_dotenv
from openai import OpenAI
load_dotenv()
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
# 各 provider 的 base_url 与默认模型
_PROVIDERS = {
"openai": {
"key_env": "OPENAI_API_KEY",
"base_url": None, # OpenAI 官方默认地址
"default_model": "gpt-5.6-luna",
},
"moonshot": {
"key_env": "MOONSHOT_API_KEY",
"base_url": "https://api.moonshot.cn/v1",
"default_model": "kimi-k3",
},
"ark": {
"key_env": "ARK_API_KEY",
"base_url": "https://ark.cn-beijing.volces.com/api/v3",
"default_model": "doubao-seed-1-6-250615",
},
}
def _to_openrouter_model(model: str) -> str:
"""把常见模型名映射到 OpenRouter 命名空间。"""
if not model:
return "openai/gpt-5.6-luna"
if "/" in model:
return model
if model.startswith("gpt-"):
return "openai/" + model
if model.startswith("claude-"):
return "anthropic/claude-opus-4.8"
return "openai/gpt-5.6-luna"
class Config:
# 被测 Agent（工具创造）默认模型
PROVIDER: str = os.getenv("PROVIDER", "openai").lower()
AGENT_MODEL: str = os.getenv("AGENT_MODEL", "gpt-5.6-luna")
# LLM-as-a-Judge 使用的模型（第 3 层工具创造质量打分）
JUDGE_MODEL: str = os.getenv("JUDGE_MODEL", "gpt-5.6-luna")
TEMPERATURE: float = float(os.getenv("TEMPERATURE", "0.2"))
@classmethod
def provider_meta(cls) -> dict:
if cls.PROVIDER not in _PROVIDERS:
raise ValueError(
f"未知 PROVIDER={cls.PROVIDER}，可选：{list(_PROVIDERS)}"
)
return _PROVIDERS[cls.PROVIDER]
@classmethod
def _use_openrouter(cls) -> bool:
"""所选 provider 的 Key 缺失、但有 OPENROUTER_API_KEY 时，走 OpenRouter 兜底。"""
meta = cls.provider_meta()
return (not os.getenv(meta["key_env"])) and bool(os.getenv("OPENROUTER_API_KEY"))
@classmethod
def map_model(cls, model: str) -> str:
"""在 OpenRouter 兜底路径下，把模型名映射到 OpenRouter 命名；否则原样返回。"""
return _to_openrouter_model(model) if cls._use_openrouter() else model
@classmethod
def get_client(cls) -> OpenAI:
"""构造并返回 OpenAI 兼容客户端（优先直连，缺 Key 时走 OpenRouter 兜底）。"""
meta = cls.provider_meta()
if cls._use_openrouter():
return OpenAI(
api_key=os.getenv("OPENROUTER_API_KEY"),
base_url=OPENROUTER_BASE_URL,
)
api_key = os.getenv(meta["key_env"], "")
if not api_key:
raise RuntimeError(
f"未找到 {meta['key_env']}，也未设置 OPENROUTER_API_KEY。"
f"请在 .env 中配置其一（OpenRouter 可作为统一兜底）。"
)
kwargs = {"api_key": api_key}
if meta["base_url"]:
kwargs["base_url"] = meta["base_url"]
return OpenAI(**kwargs)
@classmethod
def resolve_default_model(cls, override: Optional[str] = None) -> str:
"""解析被测 Agent 的模型：处理 provider 默认回退与 OpenRouter 命名映射。"""
meta = cls.provider_meta()
model = override or cls.AGENT_MODEL
# 非 openai provider 下若仍是 gpt-* 默认值，则回退到该 provider 的默认模型
if not override and cls.PROVIDER != "openai" and model.startswith("gpt-"):
model = meta["default_model"]
return cls.map_model(model)
demo.py¶
"""
实验 8-6 一键演示：python demo.py
流程：
1) 打印数据集概览与几条"不暗示工具名"的任务示例。
2) 用 strong 参考 Agent 在 2-3 个任务上跑完四层验证（含真实 LLM-as-a-Judge 打分）。
3) 用 weak 参考 Agent 做对照，展示四层的区分度。
4) 复用层对照：strong 第二次相似任务直接检索已注册工具；weak 则重复搜索创建。
5) 最后打印一张"每任务 × 每层"的结果表，横向对比各任务在四层上的得分。
用法：
python demo.py # 默认：strong 跑 3 个任务 + weak 对照 1 个任务
python demo.py --quick # 快速演示：strong / weak 各只跑 1 个任务，省时省钱
python demo.py --tasks task-01,task-07 # 指定要评估的任务 id（逗号分隔）
python demo.py --all --offline # 无 Key 离线跑完全部 20 个任务的确定性层（L1/L2/L4）
python demo.py --layers L1,L2,L4 # 只跑指定层（不含 L3 即无需联网）
python demo.py --output results.json # 把完整评分结果写出为 JSON
完整参数见 python demo.py --help。
"""
import argparse
import json
import os
import sys
import unicodedata
from agent import STRONG, WEAK, SelfEvolutionAgent, ToolRegistry
from config import Config
from harness import ALL_LAYERS, FourLayerEvaluator
HERE = os.path.dirname(os.path.abspath(__file__))
DEMO_TASK_IDS = ["task-01", "task-17", "task-07"] # 多媒体 / 地理空间 / 天气
_PROFILES = {"strong": STRONG, "weak": WEAK}
def load_dataset():
with open(os.path.join(HERE, "dataset.json"), encoding="utf-8") as f:
return json.load(f)
def fmt(x):
return "N/A" if x is None else f"{x:.3f}"
def _disp_w(s: str) -> int:
"""考虑中日韩全角字符的显示宽度，用于表格对齐。"""
return sum(2 if unicodedata.east_asian_width(c) in "WF" else 1 for c in str(s))
def _pad(s: str, width: int) -> str:
s = str(s)
return s + " " * max(0, width - _disp_w(s))
def print_dataset_overview(ds):
tasks = ds["tasks"]
print("=" * 78)
print(f"数据集：{ds['meta']['name']} 共 {len(tasks)} 个任务，覆盖领域：")
domains = [t["domain"] for t in tasks]
print(" " + " | ".join(domains))
print("-" * 78)
print("任务示例（注意：只说目标，不暗示工具名 / 库名）：")
for t in tasks[:4]:
print(f" [{t['id']}] ({t['domain']}) {t['goal']}")
print("=" * 78 + "\n")
def print_report(rep):
L = rep["layers"]
print(f"■ 任务 {rep['task_id']} ({rep['domain']}) | 画像={rep['profile']}")
print(f" L1 任务正确性 : {fmt(L['L1']['score'])} | {L['L1']['detail']}")
print(f" L2 工具发现有效性 : {fmt(L['L2']['score'])} | {L['L2']['detail']}")
l3 = L["L3"]
print(f" L3 工具创造质量 : {fmt(l3['score'])} | {l3['detail']}")
if l3.get("rubric"):
r = l3["rubric"]
print(f" Rubric: 错误处理={r.get('error_handling')} 参数校验={r.get('input_validation')} "
f"文档={r.get('documentation')} 健壮性={r.get('robustness')}")
print(f" LLM-Judge 点评: {r.get('comment', '')}")
print(f" L4 工具复用能力 : {fmt(L['L4']['score'])} | {L['L4']['detail']}")
print(f" >> 总评 overall : {fmt(rep['summary']['overall'])} (计入层: {rep['summary']['used_layers']})")
print()
def print_results_table(reports):
"""打印'每任务 × 每层'结果表：横向对比各任务在 L1-L4 与总评上的得分。"""
if not reports:
return
headers = ["任务", "领域", "画像", "L1", "L2", "L3", "L4", "总评"]
rows = []
for rep in reports:
L = rep["layers"]
rows.append([
rep["task_id"],
rep["domain"],
rep.get("profile") or "-",
fmt(L["L1"]["score"]), fmt(L["L2"]["score"]),
fmt(L["L3"]["score"]), fmt(L["L4"]["score"]),
fmt(rep["summary"]["overall"]),
])
widths = [max(_disp_w(headers[i]), *(_disp_w(r[i]) for r in rows)) for i in range(len(headers))]
print("=" * 78)
print("每任务 × 每层 结果表（N/A = 该层不适用或未选择）")
print("-" * 78)
print(" ".join(_pad(headers[i], widths[i]) for i in range(len(headers))))
for r in rows:
print(" ".join(_pad(r[i], widths[i]) for i in range(len(r))))
print("=" * 78 + "\n")
def run_profile(name, profile, tasks, evaluator, offline=False, verbose=True):
"""跑一个画像下的全部任务，返回每个任务的四层评分报告列表。"""
if verbose:
print("#" * 78)
print(f"# 用 {name} 参考 Agent 评估（每个任务：先做首次任务，再做相似任务测复用）")
print("#" * 78 + "\n")
registry = ToolRegistry() # 每个画像独立的注册表
agent = SelfEvolutionAgent(registry=registry, model=Config.AGENT_MODEL, offline=offline)
reports = []
for task in tasks:
first = agent.run(task, profile, use_variant=False) # 首次：发现+创造+注册（strong 真调 LLM 生成工具）
variant = agent.run(task, profile, use_variant=True) # 第二次相似任务：测复用
rep = evaluator.evaluate(task, first, variant)
reports.append(rep)
if verbose:
print_report(rep)
# 展示复用层的轨迹差异证据
v_actions = [s["action"] for s in variant["steps"]]
print(f" [复用探针] 第二次相似任务的动作序列: {v_actions}")
print()
return reports
def parse_args():
ap = argparse.ArgumentParser(
description="实验 8-6：为自我进化 Agent 设计评估数据集 · 四层验证演示",
formatter_class=argparse.RawTextHelpFormatter,
epilog=(
"示例：\n"
" python demo.py 默认：strong 跑 3 个任务 + weak 对照 1 个\n"
" python demo.py --quick strong / weak 各只跑 1 个任务，省时省钱\n"
" python demo.py --tasks task-01,task-07 指定要评估的任务 id\n"
" python demo.py --all --offline 无 Key 离线跑完全部 20 个任务的确定性层\n"
" python demo.py --layers L1,L2,L4 只跑指定层（不含 L3 即无需联网）\n"
" python demo.py --profile both --table 两个画像都跑，只看结果表\n"
" python demo.py --output results.json 把完整评分结果写出为 JSON"
),
)
# ---- 任务选择 ----
ap.add_argument("--quick", action="store_true",
help="快速演示：strong / weak 各只跑 1 个任务，减少 API 调用与耗时。")
ap.add_argument("--all", action="store_true",
help="评估数据集中全部 20 个任务（默认自动切到结果表输出）。")
ap.add_argument("--tasks", metavar="IDS",
help="逗号分隔的任务 id（如 task-01,task-07），缺省用内置的示例任务。")
# ---- 层与画像选择 ----
ap.add_argument("--layers", metavar="L1,L2,...",
help="选择运行哪些验证层，逗号分隔（默认四层全跑）。\n"
"仅 L3 需联网调用 LLM，去掉 L3 即可纯离线运行。")
ap.add_argument("--profile", choices=["strong", "weak", "both"],
help="被测参考 Agent 画像：strong / weak / both。\n"
"缺省保留默认行为（strong 跑全部选中任务 + weak 只跑第一个）。")
ap.add_argument("--offline", action="store_true",
help="离线模式：不调用任何 LLM（strong 用离线工具模板），\n"
"并自动跳过 L3。用于无 API Key 时演示 L1/L2/L4 确定性层。")
# ---- 模型 / 供应商 ----
ap.add_argument("--provider", choices=["openai", "moonshot", "ark"],
help="覆盖 PROVIDER（默认读环境变量，缺省 openai）。")
ap.add_argument("--agent-model", metavar="MODEL",
help="覆盖被测 Agent 造工具用的模型（默认读 AGENT_MODEL）。")
ap.add_argument("--judge-model", metavar="MODEL",
help="覆盖 L3 LLM-as-a-Judge 用的模型（默认读 JUDGE_MODEL）。")
# ---- 输出 ----
ap.add_argument("--table", action="store_true",
help="只打印'每任务 × 每层'结果表，不打印逐任务的详细分层报告。")
ap.add_argument("--output", metavar="PATH",
help="把完整评分结果（含各层明细）写出为 JSON 文件。")
return ap.parse_args()
def resolve_layers(args):
"""根据 --layers / --offline 决定实际运行哪些层。"""
if args.layers:
want = [x.strip().upper() for x in args.layers.split(",") if x.strip()]
bad = [x for x in want if x not in ALL_LAYERS]
if bad:
print(f"[错误] 未知的层：{bad}，可选：{list(ALL_LAYERS)}")
sys.exit(1)
layers = tuple(x for x in ALL_LAYERS if x in want) # 归一到 L1..L4 顺序
elif args.offline:
layers = ("L1", "L2", "L4") # 离线默认跳过需联网的 L3
else:
layers = ALL_LAYERS
if args.offline and "L3" in layers:
print("[提示] 离线模式无法运行 L3（需联网 LLM 裁判），已自动从本次层中移除 L3。\n")
layers = tuple(x for x in layers if x != "L3")
return layers
def main():
args = parse_args()
# 应用供应商 / 模型的命令行覆盖（优先级高于环境变量）
if args.provider:
Config.PROVIDER = args.provider
if args.agent_model:
Config.AGENT_MODEL = args.agent_model
if args.judge_model:
Config.JUDGE_MODEL = args.judge_model
layers = resolve_layers(args)
# 非离线模式：strong 画像造工具会真的调 LLM、L3 也需联网，因此需要可用的客户端。
if not args.offline:
try:
Config.get_client()
except Exception as e:
print(f"[配置错误] {e}")
print("提示：若只想演示确定性层，可用 `python demo.py --offline`（无需 API Key）。")
sys.exit(1)
mode = "离线(offline)" if args.offline else "联网(online)"
print(f"运行模式={mode} PROVIDER={Config.PROVIDER} "
f"AGENT_MODEL={Config.resolve_default_model()} JUDGE_MODEL={Config.JUDGE_MODEL}")
print(f"本次运行的验证层：{list(layers)}\n")
ds = load_dataset()
print_dataset_overview(ds)
by_id = {t["id"]: t for t in ds["tasks"]}
# 任务选择：--tasks 指定 > --all 全部 > --quick 取 1 个 > 默认 3 个示例任务
if args.tasks:
want = [i.strip() for i in args.tasks.split(",") if i.strip()]
missing = [i for i in want if i not in by_id]
if missing:
print(f"[错误] 数据集中不存在这些任务 id：{missing}")
sys.exit(1)
task_ids = want
elif args.all:
task_ids = [t["id"] for t in ds["tasks"]]
elif args.quick:
task_ids = DEMO_TASK_IDS[:1]
else:
task_ids = DEMO_TASK_IDS
tasks = [by_id[i] for i in task_ids]
evaluator = FourLayerEvaluator(judge_model=Config.JUDGE_MODEL, layers=layers)
# 任务较多（--all）时默认只出结果表，避免逐任务详报刷屏
verbose = not (args.table or (args.all and not args.tasks))
all_reports = []
if args.profile in (None, "strong", "both"):
# strong：好发现 + 高质量工具（离线用模板 / 联网调 LLM）+ 复用
strong_tasks = tasks
all_reports += run_profile("STRONG(强)", STRONG, strong_tasks, evaluator,
offline=args.offline, verbose=verbose)
if args.profile in ("weak", "both"):
weak_tasks = tasks
all_reports += run_profile("WEAK(弱)", WEAK, weak_tasks, evaluator,
offline=args.offline, verbose=verbose)
elif args.profile is None:
# 默认行为：weak 只跑第一个任务，凸显四层区分度
all_reports += run_profile("WEAK(弱)", WEAK, tasks[:1], evaluator,
offline=args.offline, verbose=verbose)
print_results_table(all_reports)
if args.output:
with open(args.output, "w", encoding="utf-8") as f:
json.dump({"layers": list(layers), "reports": all_reports}, f,
ensure_ascii=False, indent=2)
print(f"[已写出] 完整评分结果 -> {args.output}\n")
print("=" * 78)
print("结论：四层验证对'强/弱'两种被测 Agent 给出了不同分数；")
print("其中 L2 依据搜索关键词/选库判定发现有效性，L3 由 LLM-as-a-Judge 按 Rubric 对")
print("工具代码打分，L4 通过第二次相似任务的动作序列区分'复用'与'重复搜索'。")
print("=" * 78)
if __name__ == "__main__":
main()
harness.py¶
"""
四层分层验证 harness。
输入：一条被测 Agent 的运行轨迹（trajectory，schema 见 agent.py），以及对应的任务定义。
输出：四层各自的分数与总评。
四层：
L1 任务正确性 —— 用 dataset 的 correctness_criteria（规则/判据）核对最终答案。
L2 工具发现有效性 —— 启发式分析搜索关键词 / 访问网页 / 选库，判断发现是否切题且避开陷阱。
L3 工具创造质量 —— LLM-as-a-Judge，按 Rubric 给创造出的工具代码打分（错误处理/参数校验/文档）。
L4 工具复用能力 —— 分析"第二次相似任务"轨迹，是否直接检索已注册工具而非重复搜索创建。
"""
import json
import re
from typing import Optional
from config import Config
# 各层在总评中的权重（若某层 N/A 则在可用层间按比例重新归一）
LAYER_WEIGHTS = {"L1": 0.35, "L2": 0.25, "L3": 0.25, "L4": 0.15}
# 全部四层，供 CLI / 上层选择使用
ALL_LAYERS = ("L1", "L2", "L3", "L4")
# ---------------------------------------------------------------------------
# L1 任务正确性
# ---------------------------------------------------------------------------
def layer1_correctness(task: dict, trajectory: dict) -> dict:
answer = trajectory.get("final_answer", "") or ""
crit = task["correctness_criteria"]
check = crit["check"]
passed = False
if check == "regex":
passed = re.search(crit["pattern"], answer) is not None
elif check == "contains_any":
low = answer.lower()
passed = any(v.lower() in low for v in crit["values"])
return {
"score": 1.0 if passed else 0.0,
"passed": passed,
"detail": f"判据[{check}] -> {'通过' if passed else '未通过'}；{crit['description']}",
}
# ---------------------------------------------------------------------------
# L2 工具发现有效性
# ---------------------------------------------------------------------------
def _selected_libraries(trajectory: dict):
return [s["library"] for s in trajectory["steps"] if s["action"] == "select_library"]
def _search_queries(trajectory: dict):
return [s["query"] for s in trajectory["steps"] if s["action"] == "search"]
def layer2_discovery(task: dict, trajectory: dict) -> dict:
steps = trajectory["steps"]
reused = any(s["action"] == "retrieve_tool" for s in steps)
did_discovery = any(s["action"] in ("search", "select_library", "create_tool") for s in steps)
if reused and not did_discovery:
# 本次是复用，没有新的发现活动 —— 该层不适用
return {"score": None, "detail": "本次直接复用已注册工具，无新发现活动，L2 不适用。"}
queries = _search_queries(trajectory)
selected = _selected_libraries(trajectory)
kws = [k.lower() for k in task.get("discovery_keywords", [])]
recommended = [l.lower() for l in task["reference_solution"]["libraries"]]
pit = task.get("known_pitfalls", {})
bad_libs = [b.lower() for b in (pit.get("deprecated_libraries", []) + pit.get("paid_or_registration_apis", []))]
# 各项启发式指标
on_topic = any(any(k in q.lower() for k in kws) for q in queries) if queries else False
visited_web = any(s["action"] == "read_web" for s in steps)
def _match(lib, pool):
lo = lib.lower()
return any(p.split("(")[0].strip() in lo or lo in p for p in pool)
selected_recommended = any(_match(l, recommended) for l in selected)
hit_pitfall = any(_match(l, bad_libs) for l in selected)
avoided_pitfalls = not hit_pitfall
score = (
0.40 * selected_recommended
+ 0.25 * on_topic
+ 0.25 * avoided_pitfalls
+ 0.10 * visited_web
)
return {
"score": round(score, 3),
"components": {
"on_topic_search": on_topic,
"visited_web": visited_web,
"selected_recommended_lib": selected_recommended,
"avoided_pitfalls": avoided_pitfalls,
},
"selected_libraries": selected,
"detail": (
f"搜索切题={on_topic} 访问网页={visited_web} 选中推荐库={selected_recommended} "
f"避开陷阱={avoided_pitfalls}（选库：{selected}）"
),
}
# ---------------------------------------------------------------------------
# L3 工具创造质量（LLM-as-a-Judge，按 Rubric）
# ---------------------------------------------------------------------------
_JUDGE_SYSTEM = (
"你是一名严格的代码评审专家，负责评估一个自我进化 Agent 自动创造的 Python 工具函数的质量。"
"请只依据给定的代码本身打分，按下面 4 个维度各打 0-3 分（0=完全没有，1=很弱，2=一般，3=优秀）：\n"
" error_handling 错误处理：是否用 try/except 处理网络/IO/解析等异常，给出有用信息。\n"
" input_validation 参数校验：是否检查入参类型/取值/边界，非法输入是否报错。\n"
" documentation 文档完整性：是否有清晰 docstring 说明用途、参数、返回、异常。\n"
" robustness 健壮性与契合度：实现是否契合任务目标、是否考虑边界与失败情形。\n"
"只返回 JSON，形如："
'{"error_handling":int,"input_validation":int,"documentation":int,"robustness":int,"comment":"简短中文点评"}'
)
def _parse_judge_json(text: str) -> Optional[dict]:
try:
return json.loads(text)
except Exception:
m = re.search(r"\{.*\}", text, re.DOTALL)
if m:
try:
return json.loads(m.group(0))
except Exception:
return None
return None
def layer3_tool_quality(task: dict, trajectory: dict, judge_model: Optional[str] = None) -> dict:
created = trajectory.get("created_tools", [])
if not created:
return {"score": None, "detail": "本次轨迹未创造新工具（可能为复用），L3 不适用。"}
tool = created[0]
model = Config.map_model(judge_model or Config.JUDGE_MODEL)
client = Config.get_client()
user = (
f"任务目标：{task['goal']}\n\n"
f"Agent 创造的工具函数 `{tool['name']}` 代码如下：\n```python\n{tool['code']}\n```"
)
kwargs = dict(
model=model,
temperature=0.0,
messages=[
{"role": "system", "content": _JUDGE_SYSTEM},
{"role": "user", "content": user},
],
)
try:
resp = client.chat.completions.create(response_format={"type": "json_object"}, **kwargs)
except Exception:
resp = client.chat.completions.create(**kwargs) # 部分模型不支持 json_object
raw = resp.choices[0].message.content or ""
rubric = _parse_judge_json(raw)
if not rubric:
return {"score": 0.0, "rubric": None, "judge_text": raw, "detail": "judge 输出无法解析为 JSON。"}
dims = ["error_handling", "input_validation", "documentation", "robustness"]
total = sum(int(rubric.get(d, 0)) for d in dims)
score = round(total / (3 * len(dims)), 3)
return {
"score": score,
"rubric": rubric,
"judge_text": raw,
"tool_name": tool["name"],
"detail": (
f"Rubric 4 维合计 {total}/12 -> 归一 {score}；"
f"点评：{rubric.get('comment', '')}"
),
}
# ---------------------------------------------------------------------------
# L4 工具复用能力（分析第二次相似任务的轨迹）
# ---------------------------------------------------------------------------
def layer4_reuse(task: dict, variant_trajectory: dict) -> dict:
if variant_trajectory is None:
return {"score": None, "detail": "未提供第二次相似任务轨迹，L4 未测。"}
steps = variant_trajectory["steps"]
retrieved = any(
s["action"] == "retrieve_tool" and s.get("name") == task["tool_name"] for s in steps
)
re_searched = any(s["action"] == "search" for s in steps)
re_created = any(s["action"] == "create_tool" for s in steps)
if retrieved and not re_searched and not re_created:
score, verdict = 1.0, "直接检索并复用已注册工具（未重复搜索/创建）"
elif retrieved and (re_searched or re_created):
score, verdict = 0.5, "检索到工具但仍有重复搜索/创建"
else:
score, verdict = 0.0, "未复用，重复了搜索与工具创建"
return {
"score": score,
"retrieved_from_registry": retrieved,
"re_searched": re_searched,
"re_created": re_created,
"detail": verdict,
}
# ---------------------------------------------------------------------------
# 汇总
# ---------------------------------------------------------------------------
def aggregate(layers: dict) -> dict:
avail = {k: v["score"] for k, v in layers.items() if v.get("score") is not None}
if not avail:
return {"overall": None, "used_layers": []}
wsum = sum(LAYER_WEIGHTS[k] for k in avail)
overall = sum(LAYER_WEIGHTS[k] * s for k, s in avail.items()) / wsum
return {"overall": round(overall, 3), "used_layers": list(avail)}
class FourLayerEvaluator:
"""把四层封装到一起。variant_trajectory 用于 L4。
layers 指定实际运行哪些层（默认四层全跑）。只有 L3 需要联网调用 LLM，
因此离线场景可传 layers=("L1","L2","L4") 跳过 L3——未选中的层记 N/A，不参与总评。"""
def __init__(self, judge_model: Optional[str] = None, layers=ALL_LAYERS):
self.judge_model = judge_model or Config.JUDGE_MODEL
self.layers = tuple(layers)
def evaluate(self, task: dict, trajectory: dict, variant_trajectory: Optional[dict] = None) -> dict:
skipped = {"score": None, "detail": "（本次未选择该层，记 N/A）"}
layers = {
"L1": layer1_correctness(task, trajectory) if "L1" in self.layers else dict(skipped),
"L2": layer2_discovery(task, trajectory) if "L2" in self.layers else dict(skipped),
"L3": (
layer3_tool_quality(task, trajectory, self.judge_model)
if "L3" in self.layers else dict(skipped)
),
"L4": layer4_reuse(task, variant_trajectory) if "L4" in self.layers else dict(skipped),
}
return {
"task_id": task["id"],
"domain": task["domain"],
"profile": trajectory.get("profile"),
"layers": layers,
"summary": aggregate(layers),
}
dataset.json¶
{
"meta": {
"name": "self-evolution-eval",
"version": "1.0",
"experiment": "8-6 为自我进化 Agent 设计评估数据集",
"design_principles": [
"任务描述只说目标、不暗示工具名（如'获取某 YouTube 视频字幕'而非'用 youtube-transcript-api'）",
"覆盖 20 个不同领域，避免 Agent 记忆固定的工具模式",
"每条任务附参考方案（推荐开源库 + 典型 API 示例）与已知陷阱（废弃库 / 需付费或注册的 API）",
"correctness_criteria 用于第 1 层任务正确性判定；discovery_keywords 用于第 2 层工具发现有效性判定",
"tool_name / variant_goal 用于第 4 层工具复用能力测试（第二次相似任务应直接检索已有工具）",
"mock_answer 仅用于驱动'可控参考 Agent'跑通 harness，真实被测 Agent 不依赖它"
]
},
"tasks": [
{
"id": "task-01",
"domain": "多媒体",
"goal": "给定一个 YouTube 视频链接，获取该视频的英文字幕全文文本。",
"variant_goal": "再给你另一个 YouTube 视频链接，同样取回它的英文字幕文本。",
"tool_name": "fetch_youtube_transcript",
"reference_solution": {
"libraries": ["youtube-transcript-api"],
"api_example": "from youtube_transcript_api import YouTubeTranscriptApi; text = ' '.join(x['text'] for x in YouTubeTranscriptApi.get_transcript(video_id, languages=['en']))"
},
"known_pitfalls": {
"deprecated_libraries": ["pytube（字幕/caption 功能长期失效）"],
"paid_or_registration_apis": ["YouTube Data API v3（需 Google Cloud 项目、配额与 OAuth）"],
"notes": "video_id 需从 URL 中解析；部分视频禁用字幕会抛异常，需捕获处理。"
},
"discovery_keywords": ["youtube", "transcript", "字幕", "subtitle", "caption"],
"correctness_criteria": {
"check": "contains_any",
"values": ["transcript", "字幕", "youtube_transcript_api"],
"description": "最终答案应包含取回的字幕文本或明确说明来源方法。"
},
"mock_answer": "已通过 youtube_transcript_api 取回该视频英文字幕，共 1240 词，开头为：'Welcome back to the channel today we...'（transcript 文本已获取）。"
},
{
"id": "task-02",
"domain": "金融数据 / 加密货币",
"goal": "查询比特币当前实时价格（美元），并给出最近 7 天的价格走势数据。",
"variant_goal": "查询以太坊当前实时价格与最近 7 天走势。",
"tool_name": "crypto_price_trend",
"reference_solution": {
"libraries": ["pycoingecko", "requests"],
"api_example": "GET https://api.coingecko.com/api/v3/coins/bitcoin/market_chart?vs_currency=usd&days=7 （无需 API Key）"
},
"known_pitfalls": {
"deprecated_libraries": ["cryptocompare（免费额度收紧，需注册 Key）"],
"paid_or_registration_apis": ["CoinMarketCap API（需注册 API Key）", "Binance 部分接口有地域限制"],
"notes": "CoinGecko 公共接口有速率限制，需处理 429。"
},
"discovery_keywords": ["coingecko", "crypto", "price", "bitcoin", "加密货币", "行情"],
"correctness_criteria": {
"check": "regex",
"pattern": "(?i)(bitcoin|btc).*(\\$|usd|美元|price)",
"description": "最终答案应给出比特币的美元价格并提及 7 天趋势。"
},
"mock_answer": "通过 CoinGecko 公共 API 查询：Bitcoin 当前价格约 $63,120 USD；最近 7 天从 $59,800 上涨到 $63,120，涨幅约 5.6%，呈上升趋势。"
},
{
"id": "task-03",
"domain": "科学计算",
"goal": "求方程 x^3 - 2x - 5 = 0 在实数域上的一个数值根，精确到小数点后 6 位。",
"variant_goal": "再求方程 cos(x) - x = 0 的一个数值根。",
"tool_name": "solve_nonlinear_root",
"reference_solution": {
"libraries": ["scipy", "sympy"],
"api_example": "from scipy.optimize import brentq; root = brentq(lambda x: x**3 - 2*x - 5, 1, 3)"
},
"known_pitfalls": {
"deprecated_libraries": [],
"paid_or_registration_apis": [],
"notes": "用 eval 直接执行用户表达式有安全风险；应约束求根区间或用符号求导做牛顿法。"
},
"discovery_keywords": ["scipy", "optimize", "root", "brentq", "fsolve", "求根", "数值"],
"correctness_criteria": {
"check": "regex",
"pattern": "2\\.09455[0-9]?",
"description": "该方程的实根约为 2.094551。"
},
"mock_answer": "使用 scipy.optimize.brentq 在区间 [1,3] 求得根为 x ≈ 2.094551。"
},
{
"id": "task-04",
"domain": "地理信息 / 地理编码",
"goal": "把地址'北京市天安门广场'解析为经纬度坐标。",
"variant_goal": "把地址'上海市外滩'解析为经纬度坐标。",
"tool_name": "geocode_address",
"reference_solution": {
"libraries": ["geopy"],
"api_example": "from geopy.geocoders import Nominatim; loc = Nominatim(user_agent='demo').geocode('Tiananmen Square, Beijing'); (loc.latitude, loc.longitude)"
},
"known_pitfalls": {
"deprecated_libraries": [],
"paid_or_registration_apis": ["Google Maps Geocoding API（需计费 API Key）", "高德/百度地图 API（需注册 Key）"],
"notes": "Nominatim 使用政策要求设置 user_agent 且每秒最多 1 次请求。"
},
"discovery_keywords": ["geopy", "nominatim", "geocode", "经纬度", "geocoding", "openstreetmap"],
"correctness_criteria": {
"check": "regex",
"pattern": "39\\.9[0-9]",
"description": "天安门纬度约 39.90，经度约 116.39。"
},
"mock_answer": "使用 geopy 的 Nominatim(OpenStreetMap) 地理编码：天安门广场坐标约为纬度 39.9055, 经度 116.3976。"
},
{
"id": "task-05",
"domain": "社交媒体",
"goal": "获取某个公开 Reddit 子版块（如 r/python）当前的热门帖子标题列表（前 10 条）。",
"variant_goal": "获取 r/MachineLearning 子版块当前热门帖子标题前 10 条。",
"tool_name": "reddit_hot_titles",
"reference_solution": {
"libraries": ["praw", "requests"],
"api_example": "GET https://www.reddit.com/r/python/hot.json?limit=10 并带自定义 User-Agent；或用 praw（需注册 script app）"
},
"known_pitfalls": {
"deprecated_libraries": [],
"paid_or_registration_apis": ["Reddit 官方 API 现需 OAuth 应用注册", "Twitter/X API 已改为付费"],
"notes": "匿名 .json 端点仍可用但必须带 User-Agent，否则被限流/封禁。"
},
"discovery_keywords": ["reddit", "praw", "hot", "subreddit", "帖子", "json"],
"correctness_criteria": {
"check": "contains_any",
"values": ["reddit", "r/python", "title", "标题"],
"description": "最终答案应列出该子版块的热门帖子标题。"
},
"mock_answer": "通过 Reddit 的 r/python/hot.json 端点（带 User-Agent）取回前 10 条热门标题，例如：'What are you working on this week?'、'Async in Python explained' 等。"
},
{
"id": "task-06",
"domain": "IoT / 物联网",
"goal": "订阅某 MQTT 主题（如 sensor/temperature），接收一条消息并读出其中的温度数值。",
"variant_goal": "订阅 MQTT 主题 sensor/humidity 并读出一条消息中的湿度值。",
"tool_name": "mqtt_read_sensor",
"reference_solution": {
"libraries": ["paho-mqtt"],
"api_example": "import paho.mqtt.client as mqtt; c=mqtt.Client(); c.on_message=cb; c.connect('test.mosquitto.org',1883); c.subscribe('sensor/temperature'); c.loop_start()"
},
"known_pitfalls": {
"deprecated_libraries": [],
"paid_or_registration_apis": ["部分托管 MQTT Broker（HiveMQ Cloud 等）需注册凭据"],
"notes": "应使用公共测试 broker（如 test.mosquitto.org）；需设超时避免永久阻塞。"
},
"discovery_keywords": ["mqtt", "paho", "broker", "subscribe", "sensor", "订阅"],
"correctness_criteria": {
"check": "contains_any",
"values": ["mqtt", "paho", "温度", "temperature"],
"description": "最终答案应说明用 MQTT 订阅并读到温度值。"
},
"mock_answer": "使用 paho-mqtt 连接 test.mosquitto.org:1883，订阅 sensor/temperature，收到一条消息 payload='23.5'，解析得温度 23.5°C。"
},
{
"id": "task-07",
"domain": "天气",
"goal": "获取某城市（如东京）未来 3 天的天气预报（最高/最低气温）。",
"variant_goal": "获取伦敦未来 3 天的天气预报。",
"tool_name": "weather_forecast",
"reference_solution": {
"libraries": ["requests", "openmeteo-requests"],
"api_example": "GET https://api.open-meteo.com/v1/forecast?latitude=35.68&longitude=139.69&daily=temperature_2m_max,temperature_2m_min&forecast_days=3 （免费无需 Key）"
},
"known_pitfalls": {
"deprecated_libraries": ["Dark Sky API（被 Apple 收购并关停）"],
"paid_or_registration_apis": ["OpenWeatherMap 免费层需注册 API Key 且有调用限制"],
"notes": "Open-Meteo 免费、无需 Key，是最省事的选择。"
},
"discovery_keywords": ["open-meteo", "weather", "forecast", "天气", "预报", "temperature"],
"correctness_criteria": {
"check": "regex",
"pattern": "(?i)(°c|气温|temperature|max|min|最高|最低)",
"description": "最终答案应给出未来 3 天的高低温预报。"
},
"mock_answer": "通过 Open-Meteo 免费 API 获取东京未来 3 天预报：Day1 最高 28°C 最低 21°C；Day2 27°C/20°C；Day3 29°C/22°C。"
},
{
"id": "task-08",
"domain": "自然语言处理",
"goal": "判断一段给定文本属于哪种自然语言（语种检测）。",
"variant_goal": "对另一段文本做语种检测。",
"tool_name": "detect_language",
"reference_solution": {
"libraries": ["langdetect", "langid", "fasttext"],
"api_example": "from langdetect import detect, DetectorFactory; DetectorFactory.seed=0; detect('Bonjour tout le monde') # -> 'fr'"
},
"known_pitfalls": {
"deprecated_libraries": [],
"paid_or_registration_apis": ["Google Cloud Translation detectLanguage（需计费）"],
"notes": "langdetect 结果不确定，需设置 DetectorFactory.seed 以复现。"
},
"discovery_keywords": ["langdetect", "language detection", "langid", "语种", "fasttext"],
"correctness_criteria": {
"check": "contains_any",
"values": ["langdetect", "langid", "语种", "language"],
"description": "最终答案应给出检测到的语种。"
},
"mock_answer": "使用 langdetect（已设 seed=0 保证可复现）检测，文本 'Bonjour tout le monde' 被判定为法语 (fr)。"
},
{
"id": "task-09",
"domain": "图像处理",
"goal": "读取一张 JPEG 照片的 EXIF 元数据，取出拍摄时间与相机型号。",
"variant_goal": "读取另一张照片的 EXIF，取出拍摄时间与 GPS 坐标。",
"tool_name": "read_image_exif",
"reference_solution": {
"libraries": ["Pillow", "exifread"],
"api_example": "from PIL import Image; from PIL.ExifTags import TAGS; exif={TAGS.get(k,k):v for k,v in Image.open('p.jpg')._getexif().items()}"
},
"known_pitfalls": {
"deprecated_libraries": [],
"paid_or_registration_apis": [],
"notes": "无 EXIF 的图片 _getexif() 返回 None，需判空；GPS 需额外用 GPSTAGS 解析。"
},
"discovery_keywords": ["exif", "pillow", "pil", "exifread", "元数据", "metadata"],
"correctness_criteria": {
"check": "contains_any",
"values": ["exif", "pillow", "拍摄时间", "DateTime"],
"description": "最终答案应给出 EXIF 中的拍摄时间/相机信息。"
},
"mock_answer": "用 Pillow 的 ExifTags 解析 EXIF：DateTimeOriginal=2023:08:12 14:05:33，相机型号 Model=Canon EOS R6。"
},
{
"id": "task-10",
"domain": "文档处理",
"goal": "从一个 PDF 文件中抽取所有页面的纯文本内容。",
"variant_goal": "从另一个 PDF 文件抽取纯文本。",
"tool_name": "extract_pdf_text",
"reference_solution": {
"libraries": ["pypdf", "pdfplumber"],
"api_example": "from pypdf import PdfReader; text='\\n'.join(p.extract_text() or '' for p in PdfReader('a.pdf').pages)"
},
"known_pitfalls": {
"deprecated_libraries": ["PyPDF2（已并入并更名为 pypdf，旧包停止维护）", "textract（长期未维护、依赖冲突多）"],
"paid_or_registration_apis": ["Adobe PDF Extract API（需注册凭据）"],
"notes": "扫描版 PDF 需 OCR（pytesseract），纯文本抽取会得到空串。"
},
"discovery_keywords": ["pypdf", "pdfplumber", "pdf", "extract text", "抽取", "文本"],
"correctness_criteria": {
"check": "contains_any",
"values": ["pypdf", "pdfplumber", "PDF", "文本"],
"description": "最终答案应说明用 pypdf/pdfplumber 抽取到文本。"
},
"mock_answer": "使用 pypdf（PyPDF2 的现代替代）逐页调用 extract_text()，共抽取 12 页文本，合计约 8,300 字符。"
},
{
"id": "task-11",
"domain": "天文",
"goal": "计算某地点（如北京，纬度39.9 经度116.4）在指定日期的日出与日落时间。",
"variant_goal": "计算上海在指定日期的日出日落时间。",
"tool_name": "sun_rise_set",
"reference_solution": {
"libraries": ["astral", "skyfield", "ephem"],
"api_example": "from astral import LocationInfo; from astral.sun import sun; s=sun(LocationInfo('Beijing','China','Asia/Shanghai',39.9,116.4).observer, date=d)"
},
"known_pitfalls": {
"deprecated_libraries": ["pyephem（已更名为 ephem，旧名不再更新）"],
"paid_or_registration_apis": [],
"notes": "注意时区处理，astral 返回 UTC，需转本地时区。"
},
"discovery_keywords": ["astral", "sunrise", "sunset", "日出", "日落", "ephem", "skyfield"],
"correctness_criteria": {
"check": "regex",
"pattern": "(?i)(sunrise|sunset|日出|日落)",
"description": "最终答案应给出日出与日落时间。"
},
"mock_answer": "使用 astral 计算北京某日：日出 (sunrise) 约 05:12，日落 (sunset) 约 19:38（本地时区 Asia/Shanghai）。"
},
{
"id": "task-12",
"domain": "化学",
"goal": "根据分子式（如 C6H12O6）计算其分子量。",
"variant_goal": "计算分子式 C8H10N4O2（咖啡因）的分子量。",
"tool_name": "molecular_weight",
"reference_solution": {
"libraries": ["molmass", "periodictable", "rdkit"],
"api_example": "from molmass import Formula; Formula('C6H12O6').mass # -> 180.156"
},
"known_pitfalls": {
"deprecated_libraries": [],
"paid_or_registration_apis": [],
"notes": "rdkit 安装较重（建议 conda）；molmass/periodictable 纯 Python 更轻量。"
},
"discovery_keywords": ["molmass", "molecular weight", "分子量", "periodictable", "rdkit"],
"correctness_criteria": {
"check": "regex",
"pattern": "180\\.1[0-9]",
"description": "葡萄糖 C6H12O6 分子量约 180.16 g/mol。"
},
"mock_answer": "使用 molmass 计算：Formula('C6H12O6').mass ≈ 180.16 g/mol。"
},
{
"id": "task-13",
"domain": "生物信息",
"goal": "给定一段 DNA 序列，计算其 GC 含量百分比，并翻译为对应的蛋白质序列。",
"variant_goal": "给定另一段 DNA 序列，计算 GC 含量并翻译成蛋白质。",
"tool_name": "dna_gc_and_translate",
"reference_solution": {
"libraries": ["biopython"],
"api_example": "from Bio.Seq import Seq; from Bio.SeqUtils import gc_fraction; s=Seq('ATGGCC...'); gc_fraction(s); s.translate()"
},
"known_pitfalls": {
"deprecated_libraries": ["Bio.SeqUtils.GC（新版本改为 gc_fraction，旧函数已弃用）"],
"paid_or_registration_apis": [],
"notes": "序列长度需为 3 的倍数才能整齐翻译，否则末尾残基被丢弃并告警。"
},
"discovery_keywords": ["biopython", "bio.seq", "gc content", "translate", "GC含量", "蛋白质"],
"correctness_criteria": {
"check": "regex",
"pattern": "(?i)(gc|translate|蛋白|protein)",
"description": "最终答案应给出 GC 含量与翻译后的蛋白序列。"
},
"mock_answer": "使用 Biopython：序列 GC 含量约 58.3%，translate() 得到蛋白序列 'MAIVMGR*'。"
},
{
"id": "task-14",
"domain": "音频",
"goal": "读取一个音频文件（mp3/wav）的时长（秒）与采样率。",
"variant_goal": "读取另一个音频文件的时长与采样率。",
"tool_name": "audio_metadata",
"reference_solution": {
"libraries": ["mutagen", "soundfile", "librosa"],
"api_example": "from mutagen import File; f=File('a.mp3'); f.info.length; f.info.sample_rate"
},
"known_pitfalls": {
"deprecated_libraries": ["eyed3（仅支持 mp3，功能受限）"],
"paid_or_registration_apis": [],
"notes": "librosa 依赖较重（numba/llvmlite）；仅取元数据用 mutagen/soundfile 更轻。"
},
"discovery_keywords": ["mutagen", "soundfile", "librosa", "音频", "时长", "sample rate"],
"correctness_criteria": {
"check": "regex",
"pattern": "(?i)(采样率|sample|hz|时长|duration|秒)",
"description": "最终答案应给出时长与采样率。"
},
"mock_answer": "使用 mutagen 读取 a.mp3：时长 duration ≈ 213.4 秒，采样率 sample_rate = 44100 Hz。"
},
{
"id": "task-15",
"domain": "汇率 / 金融",
"goal": "获取美元兑人民币的当前实时汇率。",
"variant_goal": "获取欧元兑日元的当前实时汇率。",
"tool_name": "fx_rate",
"reference_solution": {
"libraries": ["requests"],
"api_example": "GET https://api.frankfurter.app/latest?from=USD&to=CNY （免费无需 Key）；或 https://open.er-api.com/v6/latest/USD"
},
"known_pitfalls": {
"deprecated_libraries": ["forex-python（依赖的免费源经常失效）"],
"paid_or_registration_apis": ["fixer.io 免费层需 Key 且只能以 EUR 为基准", "Open Exchange Rates 需注册"],
"notes": "frankfurter.app 数据源为欧洲央行，只含法币、每工作日更新。"
},
"discovery_keywords": ["frankfurter", "exchange rate", "汇率", "forex", "er-api", "currency"],
"correctness_criteria": {
"check": "regex",
"pattern": "(?i)(usd|美元).*(cny|人民币|7\\.|rate)",
"description": "最终答案应给出美元兑人民币汇率数值。"
},
"mock_answer": "通过 frankfurter.app 免费 API 查询：1 USD ≈ 7.18 CNY（数据源欧洲央行，每工作日更新）。"
},
{
"id": "task-16",
"domain": "股票市场",
"goal": "获取某只股票（如 AAPL）最近一个月的每日收盘价。",
"variant_goal": "获取 MSFT 最近一个月的每日收盘价。",
"tool_name": "stock_history",
"reference_solution": {
"libraries": ["yfinance"],
"api_example": "import yfinance as yf; df = yf.Ticker('AAPL').history(period='1mo'); df['Close']"
},
"known_pitfalls": {
"deprecated_libraries": ["pandas-datareader 的 Yahoo 源（早已失效）"],
"paid_or_registration_apis": ["Alpha Vantage 免费 Key 每分钟仅 5 次", "IEX Cloud 已转为付费"],
"notes": "yfinance 为非官方抓取，偶发限流；需处理空 DataFrame。"
},
"discovery_keywords": ["yfinance", "stock", "收盘价", "close", "ticker", "历史"],
"correctness_criteria": {
"check": "contains_any",
"values": ["yfinance", "AAPL", "收盘", "close"],
"description": "最终答案应给出该股票近一月收盘价序列。"
},
"mock_answer": "使用 yfinance 取 AAPL period='1mo' 历史数据，得到 22 个交易日收盘价，例如最近一日 Close=214.29。"
},
{
"id": "task-17",
"domain": "地理空间计算",
"goal": "计算两组经纬度坐标之间的球面（大圆）距离，单位公里。",
"variant_goal": "计算另外两组经纬度之间的大圆距离。",
"tool_name": "haversine_distance",
"reference_solution": {
"libraries": ["haversine", "geopy"],
"api_example": "from geopy.distance import geodesic; geodesic((39.9,116.4),(31.2,121.5)).km"
},
"known_pitfalls": {
"deprecated_libraries": [],
"paid_or_registration_apis": [],
"notes": "手写 haversine 时注意角度转弧度、地球半径取 6371km；geodesic 更精确（椭球模型）。"
},
"discovery_keywords": ["haversine", "geodesic", "distance", "距离", "经纬度", "great circle"],
"correctness_criteria": {
"check": "regex",
"pattern": "(?i)(km|公里|distance|距离)",
"description": "最终答案应给出两点间公里数。"
},
"mock_answer": "用 geopy.distance.geodesic 计算北京(39.9,116.4)与上海(31.2,121.5)距离约 1067.6 km。"
},
{
"id": "task-18",
"domain": "新闻 / RSS",
"goal": "解析一个 RSS/Atom 订阅源，取出最新 5 条条目的标题与链接。",
"variant_goal": "解析另一个 RSS 源，取出最新 5 条标题与链接。",
"tool_name": "parse_rss_feed",
"reference_solution": {
"libraries": ["feedparser"],
"api_example": "import feedparser; d=feedparser.parse(url); [(e.title,e.link) for e in d.entries[:5]]"
},
"known_pitfalls": {
"deprecated_libraries": [],
"paid_or_registration_apis": [],
"notes": "feedparser 对畸形 XML 有容错；需检查 d.bozo 判断解析是否有告警。"
},
"discovery_keywords": ["feedparser", "rss", "atom", "订阅", "feed", "parse"],
"correctness_criteria": {
"check": "contains_any",
"values": ["feedparser", "rss", "标题", "title"],
"description": "最终答案应给出订阅源最新条目标题。"
},
"mock_answer": "使用 feedparser 解析该 RSS 源，取回最新 5 条，例如 '(title) OpenAI releases ...' -> (link) https://... 。"
},
{
"id": "task-19",
"domain": "编码 / 二维码",
"goal": "生成一张包含指定文本（如某网址）的二维码图片并保存为 PNG。",
"variant_goal": "为另一段文本生成二维码 PNG。",
"tool_name": "generate_qrcode",
"reference_solution": {
"libraries": ["qrcode", "segno"],
"api_example": "import qrcode; qrcode.make('https://example.com').save('out.png') # 需 pillow"
},
"known_pitfalls": {
"deprecated_libraries": ["pyqrcode（长期未维护，依赖 pypng）"],
"paid_or_registration_apis": ["各类在线二维码生成 API（部分限量/需 Key）"],
"notes": "qrcode 库保存 PNG 需要安装 pillow；segno 无额外依赖。"
},
"discovery_keywords": ["qrcode", "segno", "二维码", "png", "qr"],
"correctness_criteria": {
"check": "contains_any",
"values": ["qrcode", "segno", "二维码", "png"],
"description": "最终答案应说明已生成二维码 PNG。"
},
"mock_answer": "使用 qrcode 库：qrcode.make('https://example.com').save('out.png')，已生成 290x290 的二维码 PNG。"
},
{
"id": "task-20",
"domain": "时间 / 时区",
"goal": "把一个给定的本地时间从纽约时区转换到东京时区，正确处理夏令时。",
"variant_goal": "把一个伦敦时间转换到上海时区。",
"tool_name": "convert_timezone",
"reference_solution": {
"libraries": ["zoneinfo(标准库)", "pytz", "python-dateutil"],
"api_example": "from datetime import datetime; from zoneinfo import ZoneInfo; dt.replace(tzinfo=ZoneInfo('America/New_York')).astimezone(ZoneInfo('Asia/Tokyo'))"
},
"known_pitfalls": {
"deprecated_libraries": ["直接用 datetime.timezone 固定偏移（无法处理夏令时）"],
"paid_or_registration_apis": [],
"notes": "pytz 需用 localize() 而非直接 replace()，否则会带上历史 LMT 偏移；Py3.9+ 优先 zoneinfo。"
},
"discovery_keywords": ["zoneinfo", "pytz", "timezone", "时区", "夏令时", "astimezone", "dateutil"],
"correctness_criteria": {
"check": "regex",
"pattern": "(?i)(tokyo|东京|asia/tokyo|时区|timezone)",
"description": "最终答案应给出转换后的东京时间。"
},
"mock_answer": "使用标准库 zoneinfo：纽约 2024-06-01 09:00 (America/New_York, EDT) 转为东京时间为 2024-06-01 22:00 (Asia/Tokyo)，已正确处理夏令时。"
}
]
}