预计阅读时间：11 分钟
Parallel 报告称，在使用 GPT‑6 Astra 处理劳动力市场数据的研究与综合任务后，其智能体的执行时间和成本都降到了此前模型的一半。这不是单纯比较一次模型调用的响应速度，而是在提醒团队：评估智能体模型时，应关注完成整个研究任务所付出的时间、Token、工具调用和重试成本。
“减半”应该放到完整任务链路中理解
劳动力市场研究通常不只是生成一段文字。一个可落地的智能体工作流往往包含：
- 检索职位、薪资、地区和时间范围等数据；
- 从多个来源提取结构化指标；
- 识别数据口径、时间窗口和样本差异；
- 汇总趋势并生成带证据引用的结论；
- 在格式错误、证据不足或工具失败时重试。
因此，模型升级可能从多个位置降低总成本：更少的推理轮次、更准确的工具参数、更短但信息密度更高的输出，以及更低的格式错误率。反过来，即使单次调用价格较低，只要智能体频繁重试或产生需要人工返工的结论，端到端成本仍然可能更高。
Parallel 公布的是其自身工作负载上的结果。摘要没有给出任务集规模、价格口径、质量评分方法或模型参数，因此不能直接推导出所有智能体都能获得相同的 50% 改善。更可靠的做法是用自己的任务轨迹进行复测。
用固定证据包复测速度、成本和格式成功率
下面是一个可以直接改造的基准脚本。它假设服务提供 OpenAI 兼容的 chat/completions 接口，使用同一批模拟劳动力市场证据分别调用旧模型和 GPT‑6 Astra，并记录延迟、Token 成本与 JSON 输出成功率。
示例数据是虚构的，只用于验证基准流程。运行前需要替换接口地址、模型 ID和实际价格。
# benchmark.py
import csv
import json
import os
import statistics
import time
import urllib.request
BASE_URL = os.environ["LLM_BASE_URL"].rstrip("/")
API_KEY = os.environ["LLM_API_KEY"]
ROUNDS = int(os.getenv("ROUNDS", "2"))
MODELS = {
"old": {
"id": os.getenv("OLD_MODEL", "replace-with-prior-model"),
"input_price": float(os.getenv("OLD_INPUT_USD_PER_MTOK", "0")),
"output_price": float(os.getenv("OLD_OUTPUT_USD_PER_MTOK", "0")),
},
"astra": {
"id": os.getenv("NEW_MODEL", "gpt-6-astra"),
"input_price": float(os.getenv("NEW_INPUT_USD_PER_MTOK", "0")),
"output_price": float(os.getenv("NEW_OUTPUT_USD_PER_MTOK", "0")),
},
}
# 以下数据均为虚构示例，不代表真实劳动力市场。
CASES = [
{
"case_id": "postings-trend",
"question": "比较两个季度的软件工程职位发布量，并说明变化率。",
"evidence": [
{"source_id": "s1", "period": "2025-Q1", "postings": 1200},
{"source_id": "s2", "period": "2025-Q2", "postings": 1020},
],
},
{
"case_id": "salary-regions",
"question": "比较两个地区的数据分析师薪资中位数，不要推断未提供的原因。",
"evidence": [
{"source_id": "s3", "region": "Region-A", "median_salary": 98000},
{"source_id": "s4", "region": "Region-B", "median_salary": 91000},
],
},
]
SYSTEM_PROMPT = """你是劳动力市场研究智能体。
只能使用用户提供的证据，不得补充外部事实。
每项关键结论都要引用 source_id。
证据不足时必须明确写出 unknown。
仅返回 JSON，字段为 summary、findings、citations、limitations。
"""
def call_model(model_id, case):
payload = {
"model": model_id,
"temperature": 0,
"messages": [
{"role": "system", "content": SYSTEM_PROMPT},
{
"role": "user",
"content": "分析以下证据包：\n" + json.dumps(case, ensure_ascii=False),
},
],
}
request = urllib.request.Request(
BASE_URL + "/chat/completions",
data=json.dumps(payload).encode("utf-8"),
headers={
"Authorization": "Bearer " + API_KEY,
"Content-Type": "application/json",
},
method="POST",
)
started = time.perf_counter()
with urllib.request.urlopen(request, timeout=120) as response:
data = json.loads(response.read().decode("utf-8"))
latency = time.perf_counter() - started
content = data["choices"][0]["message"]["content"]
usage = data.get("usage", {})
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
try:
parsed = json.loads(content)
valid_json = all(
key in parsed
for key in ("summary", "findings", "citations", "limitations")
)
except json.JSONDecodeError:
valid_json = False
return latency, input_tokens, output_tokens, valid_json, content
def token_cost(config, input_tokens, output_tokens):
return (
input_tokens * config["input_price"]
+ output_tokens * config["output_price"]
) / 1_000_000
rows = []
for label, config in MODELS.items():
for round_number in range(1, ROUNDS + 1):
for case in CASES:
latency, input_tokens, output_tokens, valid_json, content = call_model(
config["id"], case
)
rows.append(
{
"model": label,
"model_id": config["id"],
"round": round_number,
"case_id": case["case_id"],
"latency_seconds": round(latency, 3),
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"estimated_cost_usd": round(
token_cost(config, input_tokens, output_tokens), 8
),
"valid_json": valid_json,
"answer": content,
}
)
with open("benchmark_results.csv", "w", newline="", encoding="utf-8") as file:
writer = csv.DictWriter(file, fieldnames=rows[0].keys())
writer.writeheader()
writer.writerows(rows)
for label in MODELS:
selected = [row for row in rows if row["model"] == label]
print(
json.dumps(
{
"model": label,
"runs": len(selected),
"p50_latency_seconds": round(
statistics.median(row["latency_seconds"] for row in selected), 3
),
"total_estimated_cost_usd": round(
sum(row["estimated_cost_usd"] for row in selected), 6
),
"json_success_rate": round(
sum(row["valid_json"] for row in selected) / len(selected), 3
),
},
ensure_ascii=False,
)
)
运行方式如下。价格单位是每百万 Token 的美元价格；如果供应商使用不同计费方式，需要相应修改脚本。
export LLM_BASE_URL="https://your-provider.example/v1"
export LLM_API_KEY="replace-me"
export OLD_MODEL="your-prior-model-id"
export NEW_MODEL="your-gpt-6-astra-model-id"
export OLD_INPUT_USD_PER_MTOK="0"
export OLD_OUTPUT_USD_PER_MTOK="0"
export NEW_INPUT_USD_PER_MTOK="0"
export NEW_OUTPUT_USD_PER_MTOK="0"
export ROUNDS="5"
python benchmark.py
benchmark_results.csv 保留了每次调用的完整输出，便于后续人工评分。正式测试时，应把模拟证据替换为经过脱敏的真实任务样本，并至少覆盖简单汇总、跨来源冲突、证据缺失和长上下文四类场景。
不要让低延迟掩盖研究质量问题
劳动力市场数据尤其容易受到口径变化影响。例如，“职位发布量”可能包含重复职位，“薪资”可能是区间中点、基本工资或总薪酬，“地区”也可能按城市、都会区或远程岗位划分。更快地产生一个忽略口径差异的答案，并不构成有效优化。
除了延迟和模型账单，建议同步记录这些指标：
| 指标 | 为什么重要 |
|---|---|
| 任务完成率 | 判断智能体是否无需人工干预即可结束任务 |
| 引用覆盖率 | 检查关键结论是否能追溯到证据 |
| 数值准确率 | 捕获百分比、单位和时间窗口错误 |
| 工具调用次数 | 识别模型是否进行了多余搜索或重复读取 |
| 重试率 | 揭示格式失败和不稳定行为带来的隐性成本 |
| 人工复核时间 | 衡量自动化是否真的减少了总工作量 |
还应把搜索、抓取、模型推理和人工审核分别计时。这样才能判断改善究竟来自模型本身，还是缓存、并发、提示词缩短或数据管道调整。
从影子流量开始，而不是直接替换生产模型
可以按下面的顺序采用新模型：
- 固定任务集、提示词、工具权限和超时配置，只替换模型；
- 使用影子流量运行新旧模型，不让新模型结果直接影响用户；
- 对高风险结论设置规则校验，例如百分比重算、时间范围检查和引用存在性检查；
- 分别统计成功任务成本与失败任务成本，避免平均值掩盖重试；
- 当质量不下降且端到端成本、P50/P95 延迟都改善后，再逐步扩大流量；
- 为数据漂移、供应商故障和输出格式变化保留回滚开关。
Parallel 的结果说明，强模型可能同时改善智能体的速度和经济性，但“时间与成本减半”应被视为一个值得复现的工作负载结果，而不是通用承诺。真正稳妥的升级标准是：在自己的数据、工具和质量门槛下，用更少资源稳定完成同一项研究任务。