1. Iris 搜索 Agent 本地评测:Token 消耗不在最终答案,而在 search_planner 与 answer_synthesizer
最近复现 AllSpark 开源的 Iris 搜索 Agent 本地评测时,日志里最先暴露的不是权重加载失败,而是search_planner阶段返回401 invalid api key。我把 OpenAI 兼容调用切到 TaoToken(https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=iris_eval_intro),先到官网拿 Key,再把评测脚本里的base_url设为https://taotoken.net/api。Iris 35B / 397B 这类 Search Agent 不是一次 chat completion 就结束:评测问题先进入搜索规划器,生成查询、改写查询、过滤检索结果,最后才由答案汇总器组织输出。真正按轮次消耗 Token 的,是搜索规划和答案汇总这两类 OpenAI 兼容调用。
很多人第一次跑 Search Agent 评测,会把注意力都放在模型权重、GPU 显存和检索指标上。实际跑起来后,最容易卡住的是调用链:搜索规划器要多次调用模型来产出查询词,答案汇总器要把检索片段拼进上下文再生成答案。如果评测集有 500 条问题,每条问题规划 3 次、汇总 1 次,那就是 2000 次左右的 API 调用。每一次调用都有 prompt token、completion token、延迟和状态码,任何一次 401、404、429 都会污染结果。
所以本文不是讨论 Iris 权重怎么合并,也不是重复开源公告。这里聚焦一个更工程化的问题:本地评测 Iris 搜索 Agent 时,如何把search_planner和answer_synthesizer的 OpenAI 兼容调用统一接到 TaoToken,并把base_url配成https://taotoken.net/api,回填YOUR_API_KEY后跑通评测命令,再通过日志核对每轮搜索调用的成本与错误。
如果你还没有 Key,可以先通过官网入口进入控制台:https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=iris_eval_get_key 。拿到 Key 后不要写进代码,后面所有配置都用环境变量或本地未提交配置文件。
2. 先拿 Key,再确认 OpenAI 兼容调用的 base_url 与请求路径
TaoToken 的接入方式和大多数 OpenAI 兼容服务类似:一个 API Key,一个 Base URL。Base URL 是:
https://taotoken.net/api注意这个 Base URL 在工具配置里不要额外加 UTM 参数,也不要手写/v1再重复。很多 404 都来自路径拼接错误。比如你的代码使用https://taotoken.net/api作为base_url,那么 chat completions 的实际请求路径通常是:
https://taotoken.net/api/chat/completions如果你在 SDK 里又配置了base_url=https://taotoken.net/api/v1,或者环境变量里重复拼了/chat/completions,就可能出现 404。排查时先把请求路径打印出来。
创建 Key 的步骤如下:
- 打开 TaoToken 官网:https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=iris_eval_get_key 。
- 登录后进入 API Keys 页面创建 Key。文末也会给出 API Keys deep link。
- 复制 Key,只显示一次,不要提交到 Git。
- 本地设置环境变量,用
YOUR_API_KEY占位替换。
export TAOTOKEN_API_KEY="YOUR_API_KEY" export TAOTOKEN_BASE_URL="https://taotoken.net/api" export TAOTOKEN_MODEL="你的可用模型 ID" export IRIS_EVAL_LOG="runs/taotoken_iris_calls.jsonl"如果你用 OpenAI Python SDK,可以这样初始化:
import os from openai import OpenAI client = OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url=os.environ["TAOTOKEN_BASE_URL"], ) resp = client.chat.completions.create( model=os.environ["TAOTOKEN_MODEL"], messages=[ {"role": "user", "content": "只回复 ok"} ], temperature=0, ) print(resp.choices[0].message.content)如果你更想先用 curl 验证 Key 和路径,可以本地执行:
curl -sS https://taotoken.net/api/chat/completions \ -H "Authorization: Bearer $TAOTOKEN_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "model": "'"$TAOTOKEN_MODEL"'", "messages": [ {"role": "user", "content": "只回复 ok"} ], "temperature": 0 }'这一步通过后,再改 Iris 评测脚本。否则你会把 Key 问题、路径问题、评测脚本问题混在一起排查,日志会非常乱。
3. 可复制的 Iris 评测脚本:把 search_planner 和 answer_synthesizer 接到 TaoToken
下面给一个最小可运行的评测骨架。它不是完整 Iris 复现,而是把 Search Agent 评测里最耗 Token 的两类调用抽出来:搜索规划和答案汇总。你可以把它接到自己的检索函数、数据集和评分脚本里。
先安装依赖:
pip install httpx然后创建iris_eval.py:
#!/usr/bin/env python3 import argparse import json import os import random import time from typing import Any import httpx BASE_URL = os.getenv("TAOTOKEN_BASE_URL", "https://taotoken.net/api") API_KEY = os.getenv("TAOTOKEN_API_KEY") MODEL = os.getenv("TAOTOKEN_MODEL") LOG_PATH = os.getenv("IRIS_EVAL_LOG", "runs/taotoken_iris_calls.jsonl") if not API_KEY: raise SystemExit("请先 export TAOTOKEN_API_KEY=YOUR_API_KEY") if not MODEL: raise SystemExit("请先 export TAOTOKEN_MODEL=你的可用模型 ID") def append_log(record: dict[str, Any]) -> None: parent = os.path.dirname(LOG_PATH) if parent: os.makedirs(parent, exist_ok=True) with open(LOG_PATH, "a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") def chat(stage: str, messages: list[dict[str, str]], temperature: float = 0.2, max_tokens: int = 1024) -> str: url = f"{BASE_URL}/chat/completions" headers = { "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json", } payload = { "model": MODEL, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, } last_error = None for attempt in range(5): start = time.time() try: resp = httpx.post(url, headers=headers, json=payload, timeout=180) elapsed_ms = int((time.time() - start) * 1000) usage = {} content = "" if resp.status_code == 200: data = resp.json() usage = data.get("usage", {}) content = data["choices"][0]["message"]["content"] else: content = resp.text[:800] append_log({ "stage": stage, "model": MODEL, "status": "ok" if resp.status_code == 200 else "http_error", "http_status": resp.status_code, "elapsed_ms": elapsed_ms, "attempt": attempt + 1, "usage": usage, "preview": content[:200], }) if resp.status_code == 429: sleep_s = (2 ** attempt) + random.random() time.sleep(sleep_s) last_error = f"429 rate limited, retry after {sleep_s:.2f}s" continue resp.raise_for_status() return content except Exception as exc: elapsed_ms = int((time.time() - start) * 1000) last_error = str(exc) append_log({ "stage": stage, "model": MODEL, "status": "exception", "http_status": None, "elapsed_ms": elapsed_ms, "attempt": attempt + 1, "error": last_error, }) time.sleep((2 ** attempt) + random.random()) raise RuntimeError(f"{stage} failed after retries: {last_error}") def search_planner(question: str) -> list[str]: system = "你是搜索规划器。只输出 JSON 数组,每项是一条检索查询,不要解释。" user = f"问题:{question}\n请给出 1 到 3 条检索查询。" text = chat( "search_planner", [ {"role": "system", "content": system}, {"role": "user", "content": user}, ], temperature=0.1, max_tokens=256, ) try: queries = json.loads(text) if isinstance(queries, list): return [str(q) for q in queries][:3] except Exception: pass return [question] def local_search(queries: list[str]) -> list[dict[str, str]]: """ 这里接你本地的检索函数。 可以是 BM25、向量库、已有评测包里的 retriever。 本文不直接连接任何生产库,检索命令只在本地执行。 """ docs = [] for idx, q in enumerate(queries): docs.append({ "title": f"本地检索片段 {idx + 1}", "snippet": f"与查询“{q}”相关的本地测试文本。请替换为真实检索结果。", "source": "local_fixture", }) return docs def answer_synthesizer(question: str, docs: list[dict[str, str]]) -> str: context = "\n\n".join( f"[{i + 1}] {d.get('title', '')}\n{d.get('snippet', '')}" for i, d in enumerate(docs) ) system = "你是答案汇总器。只基于给定检索片段回答,不要编造不存在的信息。" user = f"问题:{question}\n\n检索片段:\n{context}\n\n请给出简洁答案。" return chat( "answer_synthesizer", [ {"role": "system", "content": system}, {"role": "user", "content": user}, ], temperature=0.2, max_tokens=2048, ) def run_one(item: dict[str, Any]) -> dict[str, Any]: question = item["question"] queries = search_planner(question) docs = local_search(queries) answer = answer_synthesizer(question, docs) return { "id": item.get("id"), "question": question, "queries": queries, "answer": answer, "doc_count": len(docs), } def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--dataset", required=True) parser.add_argument("--out", required=True) parser.add_argument("--limit", type=int, default=50) parser.add_argument("--concurrency", type=int, default=1) args = parser.parse_args() os.makedirs(os.path.dirname(args.out), exist_ok=True) items = [] with open(args.dataset, "r", encoding="utf-8") as f: for line in f: if line.strip(): items.append(json.loads(line)) items = items[: args.limit] with open(args.out, "w", encoding="utf-8") as out: for item in items: try: result = run_one(item) out.write(json.dumps(result, ensure_ascii=False) + "\n") out.flush() print(f"ok {result.get('id')} queries={result.get('queries')}") except Exception as exc: error_row = {"id": item.get("id"), "error": str(exc)} out.write(json.dumps(error_row, ensure_ascii=False) + "\n") out.flush() print(f"failed {item.get('id')}: {exc}") if __name__ == "__main__": main()这个脚本的关键点有三个:
第一,BASE_URL默认是https://taotoken.net/api,不额外加 UTM。
第二,Key 从TAOTOKEN_API_KEY读取,代码里只出现YOUR_API_KEY占位。
第三,每次chat调用都写入 JSONL 日志,包含 stage、http_status、elapsed_ms、usage。这样你才能知道 Token 消耗到底发生在搜索规划还是答案汇总。
准备一个本地评测集data/iris_eval.jsonl,每行至少包含id和question:
{"id": "q001", "question": "Iris 搜索 Agent 在本地评测时,搜索规划和答案汇总分别消耗什么资源?"} {"id": "q002", "question": "如何把 OpenAI 兼容调用切换到 TaoToken 并验证 base_url?"}然后本地执行:
mkdir -p runs data export TAOTOKEN_API_KEY="YOUR_API_KEY" export TAOTOKEN_BASE_URL="https://taotoken.net/api" export TAOTOKEN_MODEL="你的可用模型 ID" export IRIS_EVAL_LOG="runs/taotoken_iris_calls.jsonl" python iris_eval.py \ --dataset data/iris_eval.jsonl \ --out runs/taotoken_iris.jsonl \ --limit 20 \ --concurrency 1先跑 20 条,不要一上来跑全量。Search Agent 的调用链比普通问答长,错误会在后期放大。
4. 对照日志:每轮搜索调用的 Token、延迟、错误码怎么看
跑完后不要只看最终答案。打开runs/taotoken_iris_calls.jsonl,逐行看 stage。典型日志会分成两类:
{"stage":"search_planner","status":"ok","http_status":200,"elapsed_ms":842,"usage":{"prompt_tokens":120,"completion_tokens":45,"total_tokens":165}} {"stage":"answer_synthesizer","status":"ok","http_status":200,"elapsed_ms":3210,"usage":{"prompt_tokens":2860,"completion_tokens":310,"total_tokens":3170}}从这里可以直观看到:搜索规划器单次 completion token 不高,但调用次数多;答案汇总器调用次数少,但 prompt token 很高,因为检索片段被拼进了上下文。Iris 35B 与 397B 同量级成绩对比时,如果你只记录最终准确率,不记录调用侧成本,就无法解释为什么某些配置慢、贵或失败。
可以用 jq 快速汇总:
jq -r ' select(.stage == "search_planner" or .stage == "answer_synthesizer") | [.stage, .status, .http_status, .elapsed_ms, (.usage.prompt_tokens // 0), (.usage.completion_tokens // 0)] | @tsv ' runs/taotoken_iris_calls.jsonl | column -t再统计每个 stage 的总 token:
jq -s ' group_by(.stage) | map({ stage: .[0].stage, calls: length, prompt_tokens: (map(.usage.prompt_tokens // 0) | add), completion_tokens: (map(.usage.completion_tokens // 0) | add) }) ' runs/taotoken_iris_calls.jsonl常见错误码与处理方式:
401 invalid api key:Key 没设置、复制不完整、环境变量未生效。先回到 TaoToken 官网检查 API Keys:https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=iris_eval_debug 。
404 not found:base_url拼接错误。确认工具里配置的是https://taotoken.net/api,不要重复/v1或/chat/completions。
429 rate limit:并发过高或短时间调用太多。把--concurrency降到 1,增加重试退避。Iris 评测里搜索规划器可能连续发请求,特别容易触发。
timeout:答案汇总器上下文太长,或者检索片段 top_k 太大。先减少每次传入的片段数量,再考虑增加超时。
JSON 解析失败:搜索规划器没有严格输出 JSON 数组。把 temperature 降到 0,提示词里明确“只输出 JSON 数组”,并在脚本里保留 fallback。
日志里还有一类隐性错误:HTTP 200 但内容是空字符串。这通常是模型输出被截断或参数不兼容。把max_tokens调大,或检查模型 ID 是否正确。
5. Claude Code settings.json 与 Codex config.toml 的正确写法
如果你在 Iris 评测之外,还要用 Claude Code 或 Codex 改评测脚本、整理日志,建议把 TaoToken 的配置分开写。不要把 Anthropic 的环境变量套到 Codex 上,也不要把 Codex 的 TOML 字段套到 Claude Code 上。
Claude Code 使用settings.json和ANTHROPIC_*环境变量。可以在项目级.claude/settings.json或用户级配置里写:
{ "env": { "ANTHROPIC_BASE_URL": "https://taotoken.net/api", "ANTHROPIC_AUTH_TOKEN": "YOUR_API_KEY", "ANTHROPIC_MODEL": "你的可用模型 ID", "ANTHROPIC_SMALL_FAST_MODEL": "你的可用小模型 ID" } }这里的ANTHROPIC_BASE_URL仍然使用https://taotoken.net/api,不带 UTM。ANTHROPIC_AUTH_TOKEN用YOUR_API_KEY占位。配置完成后,在终端启动 Claude Code,先做一个简单任务,确认不会出现 401 或 404。更多细节按官方文档走:https://taotoken.net/doc/ClaudeCodeAnthropic?utm_source=taotoken_aicg_blog_end&utm_content=iris_eval_claude_code_doc 。
Codex 使用config.toml,不要写ANTHROPIC_*。示例:
model = "你的可用模型 ID" model_provider = "taotoken" [model_providers.taotoken] name = "TaoToken" base_url = "https://taotoken.net/api" env_key = "TAOTOKEN_API_KEY" wire_api = "chat"然后在本地设置:
export TAOTOKEN_API_KEY="YOUR_API_KEY"Codex 启动后会读取TAOTOKEN_API_KEY。如果你的 Codex 版本要求不同的wire_api字段,以本地codex --help和实际报错为准。关键原则是:Claude Code 用ANTHROPIC_*,Codex 用 TOML 里的model_providers,两者不要混。
6. CC Switch 三件套:供应商、模型映射、项目环境变量
如果你同时在多个项目里切换 Claude Code、Codex 和 Iris 评测脚本,手工改配置很容易把 Key 和 Base URL 搞乱。CC Switch 的价值是把“供应商切换”和“项目配置”分开。建议按三件套管理:
第一件,供应商清单。只记录供应商名称、Base URL、Key 环境变量名。例如:
name: TaoToken base_url: https://taotoken.net/api env_key: TAOTOKEN_API_KEY第二件,模型映射。把不同工具需要的模型 ID 写进环境变量或映射表,不要硬编码到每个脚本里。
export TAOTOKEN_MODEL="你的可用模型 ID" export TAOTOKEN_SMALL_MODEL="你的可用小模型 ID"第三件,项目级环境变量文件。Iris 评测、Claude Code、Codex 都从同一组变量读,但各自只读自己需要的字段。可以创建.env.taotoken:
TAOTOKEN_API_KEY=YOUR_API_KEY TAOTOKEN_BASE_URL=https://taotoken.net/api TAOTOKEN_MODEL=你的可用模型 ID IRIS_EVAL_LOG=runs/taotoken_iris_calls.jsonl本地加载:
set -a source .env.taotoken set +a python iris_eval.py --dataset data/iris_eval.jsonl --out runs/taotoken_iris.jsonl --limit 20注意.env.taotoken要加入.gitignore。不要把YOUR_API_KEY替换成真实 Key 后提交。CC Switch 切换供应商时,只改变量文件或供应商条目,不要改评测代码。这样 Iris 评测的 base_url、Claude Code 的 settings.json、Codex 的 config.toml 互不污染。
7. 本地评测 Iris 35B / 397B 的排障清单
Iris 35B 和 397B 是不同规模的搜索 Agent 模型,本地评测时资源差异很大。无论你只跑评测 harness,还是把模型用 OpenAI 兼容服务托管在本地,下面这份清单都适用。
第一,先区分“模型推理”和“工具调用”。如果你把 Iris 本地托管为 OpenAI 兼容服务,那么search_planner的base_url可能指向本地服务;而答案汇总或裁判模型可以指向 TaoToken。不要把所有调用都混在一个 base_url 里。建议在日志里记录provider字段。
第二,控制输入长度。答案汇总器最容易爆 token,因为检索片段太多。先用 top_k=3 到 5 跑通,再逐步加。Iris 搜索 Agent 的评测重点是可复现,不是一次把所有上下文塞满。
第三,搜索规划器强制 JSON。提示词要短而硬:
你是搜索规划器。只输出 JSON 数组,例如 ["查询1","查询2"]。 不要输出 Markdown,不要解释。第四,并发不要太高。本地模型服务可能能承受并发,但外部 API 调用会有速率限制。Iris 评测脚本里搜索规划器会连续请求,建议先--concurrency 1,稳定后再加。
第五,日志必须落盘。只在控制台看输出,跑完就找不回 429 和空响应。JSONL 日志按行追加,方便后续统计。
第六,区分缓存与评测。为了提高速度,你可能会缓存搜索规划结果。但正式评测时,缓存会让不同模型共享同一批查询,影响对比。建议缓存只用于调试。
第七,不要直接连接生产数据库。Search Agent 的检索函数应该使用本地索引、测试快照或脱敏数据。所有 SQL 和检索命令由读者在本地执行。
第八,Key 不要出现在报错截图中。日志里只记录状态码和错误类型,不记录 Authorization 头。preview字段也不要包含 Key。
8. 把每轮搜索调用成本压到可观测,再谈同量级成绩对照
复现 AllSpark Iris 搜索 Agent 时,最值得先做的不是刷榜单,而是建立可观测的调用链。每轮搜索调用至少记录这些字段:
{ "stage": "search_planner", "model": "你的模型 ID", "provider": "taotoken", "status": "ok", "http_status": 200, "elapsed_ms": 842, "attempt": 1, "usage": { "prompt_tokens": 120, "completion_tokens": 45, "total_tokens": 165 } }有了这些字段,你才能回答几个关键问题:搜索规划器平均调用几次?答案汇总器平均 prompt token 是多少?429 出现在哪个 stage?哪个模型 ID 在本地评测集上更稳定?Iris 35B 与 397B 的同量级成绩对照,应该建立在同一份评测集、同一套检索函数、同一套日志口径上,而不是只用最终答案分数。
优化顺序建议:
- 先修 401、404、429、timeout。
- 再降
search_planner的 completion token,限制它只输出查询。 - 然后压缩
answer_synthesizer的输入片段,去掉重复文本。 - 最后才增加并发和批量评测。
如果你要长期跑 Claude Code、Codex 和搜索 Agent 评测,建议直接准备一个稳定的 Coding Plan,再用 API Keys 管理不同项目的 Key。先用模型对话验证请求格式,再把它接进评测脚本,最后对照日志修正参数。
文末按这个路径操作:先试模型对话,确认 Key 和 Base URL 可用;再考虑 Coding Plan;然后创建独立 API Key;最后按 Claude Code 文档配置本地工具。
模型对话:https://taotoken.net/models/detail/chat?utm_source=taotoken_aicg_blog_end&utm_content=iris_eval_chat
Coding Plan:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=iris_eval_coding_plan
创建 API Key:https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=iris_eval_api_keys
Claude Code 文档:https://taotoken.net/doc/ClaudeCodeAnthropic?utm_source=taotoken_aicg_blog_end&utm_content=iris_eval_claude_code_doc
TaoToken 官网入口:https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=iris_eval_footer
把base_url固定为https://taotoken.net/api,Key 用YOUR_API_KEY回填,先跑 20 条 Iris 评测,再打开runs/taotoken_iris_calls.jsonl对照每轮搜索调用的 Token、延迟和状态码。能稳定复现日志,再去比较 Iris 35B / 397B 在你本地评测集上的同量级表现,结论才站得住。