用 agno 生成 Tool-Call 轨迹训练数据:真实 Schema 校验、多轮模拟与法官过滤实战
【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno
本文基于 agno 仓库
cookbook/data_labeling/_25_tool_call_trajectories/目录讲解如何让框架"自产自销"函数调用(Function-Calling)SFT 数据:以真实 agno 工具 Schema 为约束生成单步 (query, tool call) 配对,用模拟用户与真实执行工具的助手对话产出多轮轨迹,再由温度 0 的法官过滤出可入训的 rollout。读完你将掌握一条从"运行中的 Agent"到"可训练的 Tool-Call 轨迹数据"的完整数据管线,以及每一环节在 agno 源码中的实现依据。
一、为什么让 Agent 自己生成训练数据
函数调用(Function Calling)是 Agent 与外部工具交互的核心能力。要训练模型学会"面对用户请求 → 选出正确工具 → 给出符合 Schema 的参数 → 依据真实执行结果继续",需要高质量的 (query, tool call) 配对与多轮工具使用轨迹。人工构造这类数据成本高、覆盖差,且容易与实际运行的工具 Schema 脱节。
agno 在cookbook/data_labeling/_25_tool_call_trajectories/中给出了一条完全不同的路径:让框架自己生成自己的训练数据。核心思路是:
- 从 agno 真正运行的 Toolkit(
CalculatorTools、DuckDuckGoTools)中提取每个函数的真实 JSON Schema; - 用生成器 Agent 依据这些 Schema 写出候选配对,再用纯代码校验器逐一核对(工具是否已知、参数能否解析、必需参数是否齐全、有无未知参数、原始类型是否匹配);
- 多轮数据则来自模拟用户与一个真实执行工具调用的助手之间的对话,执行过的调用(工具名、参数、结果)直接从
RunOutput中提取; - 最后用一个temperature-0 的法官判定每条轨迹是否真正达成了用户的阶段性目标,只有验证通过的 rollout 才进入训练集。
这种"keep-what-passes"的形态与目录_21_rejection_sampling/一脉相承,只是这里被采样的对象从单条响应变成了整条轨迹(_21_rejection_sampling/README.md)。
二、目录总览与三阶段管线
cookbook/data_labeling/_25_tool_call_trajectories/下包含四个文件:
| 文件 | 角色 | 产出 |
|---|---|---|
basic.py | Schema 校验的单步 (query, tool call) 配对生成 | data/generated/tool_call_sft.jsonl |
multi_turn_simulation.py | 模拟用户与真实执行工具的助手对话,生成多轮轨迹 | data/generated/multi_turn_trajectories.jsonl |
judge_filter.py | 温度 0 法官校验多轮轨迹,只保留验证成功的 rollout | data/generated/verified_trajectories.jsonl |
README.md/TEST_LOG.md | 使用说明与实测记录 | — |
整个管线是"单步配对 → 多轮模拟 → 法官过滤"的递进关系:basic.py解决"模型是否会按 Schema 发出格式正确的调用",multi_turn_simulation.py解决"模型能否在多步计算中连续、正确地使用工具",judge_filter.py解决"只有被验证成功过的轨迹才能进入训练"。
三、阶段一:basic.py—— 基于真实 Schema 的单步配对
3.1 从 Toolkit 提取真实 JSON Schema
basic.py的关键设计是:训练数据的 Schema 不是手写的,而是直接从 agno 真实运行的 Toolkit 中提取。toolkit_schemas()遍历工具包注册的每个函数,调用fn.process_entrypoint()让函数从其签名与 docstring 生成参数 Schema,然后组装成{name, description, parameters, source}结构:
def toolkit_schemas(toolkit: Any, source: str) -> Dict[str, dict]: """Extract each tool's real JSON schema from an agno toolkit.""" schemas = {} for name, fn in toolkit.functions.items(): fn.process_entrypoint() # populates fn.parameters from the signature and docstring schemas[name] = { "name": name, "description": fn.description, "parameters": fn.parameters, "source": source, } return schemas(见 basic.py)
这里的process_entrypoint()是 agno 框架内部处理函数入口、生成模型可用参数 Schema 的统一入口,位于 libs/agno/agno/tools/function.py("Process the entrypoint and make it ready for use by an agent.")。也就是说,校验器面对的 Schema 与 Agent 运行时实际交给大模型的 Schema 是同一份,不存在"训练数据与推理时工具定义不一致"的问题。
本 demo 汇入了两套真实工具包:
CalculatorTools():来自 libs/agno/agno/tools/calculator.py,注册了 8 个纯本地计算函数:add、subtract、multiply、divide、exponentiate、factorial、is_prime、square_root(见 calculator.py),每个函数返回 JSON 字符串,例如add返回{"operation": "addition", "result": ...};DuckDuckGoTools():来自 libs/agno/agno/tools/duckduckgo.py,是WebSearchTools的便捷包装,backend默认 "duckduckgo",提供搜索/新闻搜索函数。
3.2 用 Pydantic 输出模型约束生成器
生成器 Agent 使用结构化输出模型ToolCallPairs,内含N_PAIRS = 8个候选配对,每个配对由query(自然语言用户请求)、tool_name(Schema 列表中的精确工具名)、arguments_json(JSON 对象字符串形式的调用参数)组成:
class ToolCallPair(BaseModel): query: str = Field(..., description="A realistic user request answerable by a single call to one of the listed tools") tool_name: str = Field(..., description="The tool to call, exactly as named in the schema list") arguments_json: str = Field(..., description='The call arguments as a JSON object string, for example {"a": 2, "b": 3}') class ToolCallPairs(BaseModel): pairs: list[ToolCallPair] = Field(..., description="Candidate (query, tool call) pairs, each using a listed tool")(见 basic.py)
生成器 Agent 的指令明确要求:每个 query 必须能被恰好一次调用回答;参数必须严格满足 Schema(必需参数齐全、无多余参数、原始类型正确);要变化工具与措辞,并在 query 中放入具体数值:
generator = Agent( model="google:gemini-3.5-flash", instructions=( "You write function-calling SFT data. Given a list of tool JSON " "schemas, produce natural user queries that are each answerable by " "exactly one call to one of the tools, together with that call. " "Arguments must satisfy the tool's schema exactly: required params " "present, no extra params, correct primitive types. Vary the tools " "and the phrasing; put concrete values in every query." ), output_schema=ToolCallPairs, )(见 basic.py)
提示构造(build_prompt())会把全部工具的真实 Schema 以 JSON 序列化后拼入,并要求覆盖至少 5 个不同工具(见 basic.py)。
3.3 纯 stdlib 校验器:不依赖 LLM 的质量门槛
这是basic.py最值得借鉴的部分:候选数据由 LLM 生成,但验收完全由确定性代码完成。json_type_matches()实现 JSON Schema 原始类型与 Python 值的对应关系(注意integer与number都排除了bool,因为 Python 中True/False是int子类):
def json_type_matches(expected: Optional[str], value: Any) -> bool: if expected == "string": return isinstance(value, str) if expected == "integer": return isinstance(value, int) and not isinstance(value, bool) if expected == "number": return isinstance(value, (int, float)) and not isinstance(value, bool) if expected == "boolean": return isinstance(value, bool) if expected == "array": return isinstance(value, list) if expected == "object": return isinstance(value, dict) return True # schema declares no primitive type for this param: accept(见 basic.py)
validate_pair()按顺序执行五重检查:工具已知 → 参数可被json.loads解析 → 参数是 JSON 对象 → 所有required参数存在 → 无未知参数且原始类型匹配,任一失败即返回(None, reason):
def validate_pair(pair: ToolCallPair) -> Tuple[Optional[dict], str]: """Check one candidate against the real schema. Returns (arguments, reason).""" schema = SCHEMAS.get(pair.tool_name) if schema is None: return None, f"unknown tool '{pair.tool_name}'" try: arguments = json.loads(pair.arguments_json) except json.JSONDecodeError as exc: return None, f"arguments are not valid JSON: {exc.msg}" if not isinstance(arguments, dict): return None, "arguments are not a JSON object" properties = schema["parameters"].get("properties", {}) for name in schema["parameters"].get("required", []): if name not in arguments: return None, f"missing required param '{name}'" for name, value in arguments.items(): if name not in properties: return None, f"unknown param '{name}'" expected = properties[name].get("type") if not json_type_matches(expected, value): return None, f"param '{name}' should be {expected}, got {type(value).__name__}" return arguments, "ok"(见 basic.py)
主流程对N_PAIRS个候选逐一校验,仅将通过的配对写入data/generated/tool_call_sft.jsonl,并为每行附上schema_source溯源字段(如"agno.tools.calculator"),被丢弃的候选打印丢弃原因(见 basic.py)。
3.4 产出行示例
{"query": "Calculate the sum of 124.5 and 89.2", "tool_name": "add", "arguments": {"a": 124.5, "b": 89.2}, "schema_source": "agno.tools.calculator"}每一行的参数都已经过真实 Schema 校验,schema_source保证其来源可追溯。
四、阶段二:multi_turn_simulation.py—— 模拟用户驱动真实执行的多轮轨迹
4.1 双 Persona 用户模拟器
multi_turn_simulation.py用两个 Persona 模拟用户(USER_SIMS),每个都带一个多步计算目标,每轮只请求一步、绝不自己心算,达成全部目标后以单词DONE结束:
PERSONAS: Dict[str, str] = { "dinner_host": ( "You are simulating a user planning a dinner for friends. Your goal, " "step by step: (1) find the cost of 4 pizzas at 18.50 each, (2) add " "the 12.75 delivery fee, (3) split the total evenly among 5 people." ), "math_student": ( "You are simulating a student double-checking homework. Your goal, " "step by step: (1) find out whether 97 is a prime number, (2) compute " "12 factorial divided by 10 factorial." ), }(见 multi_turn_simulation.py)
助手端是一个真正配置了tools=[CalculatorTools()]的 agno Agent,指令要求每一步算术都走工具而不是心算,回答保持一两句并给出数值结果(见 multi_turn_simulation.py)。
4.2 模拟循环与轨迹提取
run_conversation()在MAX_TURNS = 3的硬上限内循环:模拟用户先产出下一条 user 消息(若是DONE则剥离该标记并结束);随后把完整对话记录交给助手,助手真实地执行工具调用并回答;最后遍历assistant_run.tools,把实际执行过的调用(工具名、参数、结果)逐条追加到tool_calls:
for tool in assistant_run.tools or []: tool_calls.append( { "tool_name": tool.tool_name, "arguments": tool.tool_args, "result": tool.result, } ) turns += 1(见 multi_turn_simulation.py)
这里的关键点是:tool_calls里记录的是RunOutput.tools中真实执行并返回结果的调用,而不是模型"打算"调用的内容——这正是"真实 tool-executing rollouts"的含义。每条对话产出一行,包含persona、完整messages、执行过的tool_calls与turns轮数,写入data/generated/multi_turn_trajectories.jsonl:
{"persona": "dinner_host", "messages": [{"role": "user", "content": "Hey! I'm planning a dinner with some friends ..."}, ...], "tool_calls": [{"tool_name": "multiply", "arguments": {"b": 18.5, "a": 4}, "result": "{\"operation\": \"multiplication\", \"result\": 74.0}"}, ...], "turns": 3}五、阶段三:judge_filter.py—— 温度 0 法官把关
多轮轨迹虽然"真实执行"了,但执行成功不等于"达成了用户目标"。judge_filter.py解决最后一个问题:哪些 rollout 才值得进训练集。
5.1 独立可跑的过滤脚本
它从multi_turn_simulation.py导入PERSONAS、render_transcript、run_conversation并重新运行模拟,因此不依赖任何预先存在的 JSONL,可独立执行(见 judge_filter.py)。
5.2 法官的判定结构
法官用Gemini(id="gemini-3.5-flash", temperature=0)构造(温度 0 保证判定可复现),输出模型TrajectoryVerdict由success与reason组成:
class TrajectoryVerdict(BaseModel): success: bool = Field( ..., description="True only if every step of the user's goal was answered correctly using the executed tool calls", ) reason: str = Field( ..., description="One or two sentences citing the specific tool calls or answers behind the verdict", )(见 judge_filter.py)
法官指令要求逐条核对:助手是否用正确的参数与正确的算术完成了目标的每一步;数字错误、跳步、最终答案没有对应执行过的工具调用,均判失败(见 judge_filter.py)。评审提示(build_judge_prompt())把 Persona 目标、对话全文、执行过的工具调用三者一并交给法官(见 judge_filter.py)。
5.3 通过才保留,理由进 provenance
只有verdict.success为真的轨迹才会写入data/generated/verified_trajectories.jsonl,且法官的理由被记入provenance:
{"persona": "math_student", "messages": [...], "tool_calls": [...], "turns": 3, "provenance": {"judge": "gemini-3.5-flash", "reason": "The assistant correctly checked if 97 is prime using the 'is_prime' tool and computed 12 factorial divided by 10 factorial ... obtaining the correct result of 132."}}失败轨迹则打印原因并丢弃(见 judge_filter.py)。至此,"数据是否可入训"由可验证的判定背书,而不是由生成过程默认成立。
六、设计边界:离线、确定性与刻意收敛
目录 README 明确说明了一个刻意为之的边界:执行侧只使用离线 CalculatorTools。basic.py中的 DuckDuckGo 工具只以 Schema 形式出现、从不真正调用,因此整个 demo 在工具侧是确定性的,除模型 API 外不需要任何网络请求。这带来两个好处:
- 训练数据生成过程可复现、可审计(工具执行结果恒定);
- 演示可以聚焦"Schema 校验 + 轨迹提取 + 法官过滤"这三个数据工程要点,不被外部服务的不确定性干扰。
TEST_LOG.md的实测记录也印证了这一点:在一次运行中math_student用multiply(12, 11)一步化简 12!/10!,法官正确接受了这一等价简化并引用"which simplifies to 12 * 11 ... obtaining the correct result of 132",说明校验器/法官评估的是"目标是否达成 + 结果是否正确",而非僵化的步骤顺序。
七、运行方式与环境要求
在仓库根目录下依次执行三个脚本即可:
python cookbook/data_labeling/_25_tool_call_trajectories/basic.py python cookbook/data_labeling/_25_tool_call_trajectories/multi_turn_simulation.py python cookbook/data_labeling/_25_tool_call_trajectories/judge_filter.py- 依赖:需要
GOOGLE_API_KEY(三个脚本均使用google:gemini-3.5-flash模型); - 环境前提:已在环境中安装 agno(demo 实测版本为 agno 2.7.4,见 TEST_LOG.md);
- 产出位置:所有行写入
cookbook/data_labeling/_25_tool_call_trajectories/data/generated/,该目录已被 gitignore,运行脚本即可重新生成; - 规模参考:
basic.py一次生成 8 个配对;multi_turn_simulation.py每次为 2 个 Persona 各跑至多 3 轮;judge_filter.py重跑模拟后再逐条判定。据 TEST_LOG 实测,一次完整运行得到2 行多轮轨迹 / 8 次执行工具调用(dinner_host3 轮 3 次调用,math_student3 轮 5 次调用),工具调用次数会因模型行为在运行间略有浮动。
八、何时使用这套管线
README 的 "When to use" 给出了三个明确的适用场景:
- 单步配对:当你要教模型针对固定 Schema 发出格式良好的调用(对应
basic.py); - 多轮轨迹:当你要教模型在上下文中进行多步工具使用、并带有真实执行结果(对应
multi_turn_simulation.py); - 法官过滤:当只有验证成功过的 rollout 才应进入训练(对应
judge_filter.py)。
前提是你已经在用带类型化工具(typed tools)的 agno Agent 运行任务。此外,管线的上下游衔接也已打通:
- 上游形态参考:
keep-what-passes的思路与_21_rejection_sampling/一致,区别在于这里被采样的单元是整条轨迹而非单条响应; - 下游加工:需要大规模去重、过滤与混配时,使用
_22_dataset_curation/(LLM 质量门 + MinHash 近重复检测 + 13-gram 去污染,其中 dedup 与 decontamination 均为纯 stdlib、完全确定性的实现)。
九、实测结论与调优提示
TEST_LOG.md记录了 2026-07-18 针对gemini-3.5-flash、agno 2.7.4 的实测:
basic.py通过率:两次运行均"wrote 8 rows ... kept 8, dropped 0",模型对这类简单 Schema 能稳定产出精确参数(数字写成 JSON number、factorial/is_prime写整数、max_results仅在查询需要时才出现)。校验器该次运行未触发,但 kept/dropped 会随运行变化——这正是纯代码校验器存在的意义:把不确定的 LLM 输出收敛为确定的数据质量。multi_turn_simulation.py:dinner_host依次执行multiply(4, 18.5) → 74.0、add(74, 12.75) → 86.75、divide(86.75, 5) → 17.35,全部正确;math_student执行is_prime(97)、factorial(12)、factorial(10)、divide → 132.0及一次验证性multiply(12, 11)。导出器在写 JSONL 前剥离DONE标记,因此目标完成与否在数据中不可见,而dinner_host在 3 轮上限处结束(最后一问仍在回答)正是"硬轮数上限下的多步目标"的预期形态。judge_filter.py:实测 "wrote 2 rows ... kept 2, dropped 0",两条轨迹均通过验证;失败路径当时仅由代码检视覆盖,因为上下文中有正确的计算器算术时,助手很少在这些小目标上失败。
这几点对实际使用有直接提示:不要把生成器的输出直接当训练数据,而是永远让确定性校验器或法官决定取舍;多轮数据要在"轮数上限"与"目标完整性"之间取舍;工具调用次数存在运行间浮动,数据量统计应基于多次运行的平均水平。
结语
_25_tool_call_trajectories演示了一条对训练数据工程很有参考价值的模式:Schema 取自运行时、候选由 Agent 生成、质量由确定性代码与可验证判定把关、溯源字段全程留痕。无论是为 Function-Calling 微调构造单步配对,还是为多步工具使用构造带真实执行结果的多轮轨迹,这套"生成 → 校验 → 过滤"的三段式管线都可以直接迁移到你自己的 agno 工具集上。
【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考