OPC中国智能体成本控制：从 Token 预算到可观测性的工程实践
一、问题背景：为什么多步骤智能体容易成本失控？
在OPC中国所代表的 AI 协同型 One Person Company（一人公司）实践中，一个人可能维护研究、写作和客户支持等多条工作流。本文讨论这类小型项目的调用成本，不涉及工业自动化领域的 OPC/OPC UA 协议。
在一个简单聊天应用中，一次请求通常对应一次模型调用；在智能体工作流中，一次用户任务可能被拆成意图识别、资料检索、计划生成、工具执行、结果审查和格式整理等多个步骤。如果某个节点失败后自动重试，或者每一步都携带完整会话历史，一次任务可能产生远超预期的调用量。
成本失控通常不是某一次请求特别昂贵，而是以下因素叠加：
- 所有任务都调用能力最强、成本更高的模型；
- 系统提示词、工具说明与历史消息被重复发送；
- 检索返回过多无关片段，扩大输入上下文；
- 输出长度没有上限，模型生成大量不必要内容；
- 工具或模型失败后无限重试；
- 多智能体相互讨论，却没有停止条件；
- 日志只记录“成功/失败”，没有记录 Token、耗时和调用链。
对 OPC中国项目而言，资源通常更有限，因此成本治理的目标不是单纯压低每次调用，而是让每一笔消耗都能回答三个问题：花在什么任务上、产生了什么结果、是否值得继续。
二、先建立成本公式：不要只看模型单价
大模型服务通常按输入与输出 Token 计费，部分能力还可能产生知识库、工具或其他资源消耗。阿里云百炼的模型调用价格页面按模型列出输入、输出 Token 单价，并说明部分模型存在阶梯计费、Batch 调用或上下文缓存等规则。由于价格和活动可能调整，生产系统应通过配置维护单价，不要把数字写死在业务代码中。阿里云百炼模型调用价格
一次任务的估算成本可以表示为：
[C{task}=\sum{i=1}^{n}(T{in,i}P{in,i}+T{out,i}P{out,i}+C{tool,i})+C{infra}]
其中：
- (T_{in,i})：第 i 次模型调用的输入 Token；
- (T_{out,i})：第 i 次模型调用的输出 Token；
- (P{in,i})、(P{out,i})：对应模型的输入、输出单价；
- (C_{tool,i})：搜索、代码执行或其他工具的费用；
- (C_{infra})：数据库、日志、队列等基础设施成本。
但仅计算金额仍然不够。更有用的单位是“每个有效交付的成本”：
[C{delivery}=\frac{C{total}}{N_{accepted}}]
如果模型调用成本下降，但返工率上升、人工审核时间增加，最终每个有效交付可能反而更贵。
三、在调用之前设置三层预算
不要等账单出现异常后再限制使用。建议在请求进入智能体工作流时，就设置三层预算。
1. 单次调用预算
约束某个节点的最大输入、最大输出和超时时间。例如分类节点只需要输出固定 JSON，就不应允许生成几千 Token。
2. 单任务预算
限制一个用户任务最多调用多少次模型、多少次工具以及最多重试几次。即使每次调用都很便宜，无限循环仍会造成资源浪费。
3. 时间窗口预算
按用户、工作流或项目设置每日/每周预算。当异常流量、错误配置或恶意请求出现时，系统可以降级或暂停，而不是持续消耗。
下面是一个与具体模型 SDK 无关的 Python 预算对象：
from dataclasses import dataclass
class BudgetExceeded(RuntimeError):
pass
@dataclass
class TaskBudget:
max_calls: int = 8
max_input_tokens: int = 30_000
max_output_tokens: int = 8_000
max_tool_calls: int = 4
used_calls: int = 0
used_input_tokens: int = 0
used_output_tokens: int = 0
used_tool_calls: int = 0
def reserve_model_call(self, estimated_input: int, max_output: int) -> None:
next_calls = self.used_calls + 1
next_input = self.used_input_tokens + estimated_input
next_output = self.used_output_tokens + max_output
if next_calls > self.max_calls:
raise BudgetExceeded("模型调用次数超过任务预算")
if next_input > self.max_input_tokens:
raise BudgetExceeded("输入 Token 超过任务预算")
if next_output > self.max_output_tokens:
raise BudgetExceeded("输出 Token 超过任务预算")
self.used_calls = next_calls
self.used_input_tokens = next_input
self.used_output_tokens = next_output
def reserve_tool_call(self) -> None:
if self.used_tool_calls + 1 > self.max_tool_calls:
raise BudgetExceeded("工具调用次数超过任务预算")
self.used_tool_calls += 1
示例中的数值只是说明结构，生产参数必须根据真实任务、模型限制和评测结果配置。预留预算后，还应使用响应中返回的实际用量修正统计，不能只依赖输入前估算。
四、模型分级路由：把复杂模型用在真正复杂的节点
智能体工作流中的任务复杂度并不相同。意图分类、格式校验和字段抽取通常目标明确；方案权衡、复杂代码分析和多来源综合则需要更强推理能力。
可以建立三级路由：
| 任务级别 | 典型任务 | 路由原则 |
| L1：确定性强 | 分类、抽取、格式转换、敏感词初筛 | 优先轻量模型或规则 |
| L2：需要理解 | 摘要、改写、一般问答、草稿生成 | 使用均衡型模型 |
| L3：复杂推理 | 架构权衡、疑难代码、跨文档综合 | 使用能力更强的模型并人工审核 |
路由不能只靠用户一句话的长度判断。至少要结合任务类型、是否需要工具、输入规模、输出风险和历史失败率。
一个实用策略是“先低后高”：先用成本更低的路径处理；若输出不满足结构校验、证据不足或置信度低，再升级到更强模型。升级必须有明确触发条件，不能让模型自行无限升级。
五、压缩上下文：最便宜的 Token 是不发送的 Token
长上下文并不自动带来更好结果。无关历史、重复工具说明和过量检索内容会增加成本，也可能分散模型注意力。
1. 固定提示与动态数据分离
将角色、规则和输出格式写成版本化模板，仅注入当前任务所需的变量。阿里云百炼 Prompt 模板支持把固定结构与动态变量分离，并通过模板 ID 管理和调用，有助于统一版本与复用。Prompt 模板概述
2. 会话历史分层
不要每次发送完整聊天记录。可以保留：最近若干轮原文、较早对话摘要、不可丢失的用户约束。摘要更新时要记录版本，避免重要条件被模型压缩掉。
3. 检索结果限量
只传递能回答问题的前 K 个片段，并设置相似度阈值与最大总字符数。检索片段越多，不代表证据越充分。
4. 工具说明按需加载
当工作流有大量工具时，不要把全部工具描述发给每个节点。先根据任务选择候选工具，再向执行节点提供必要接口。
六、上下文缓存：先判断是否真的存在稳定公共前缀
如果大量请求共享较长的系统说明、工具定义或同一份文档，上下文缓存可以减少重复计算。阿里云百炼官方文档说明，上下文缓存通过匹配请求中的公共前缀降低延迟和成本，并提供显式、隐式两种模式；支持范围、最少 Token、有效期和计费规则应以当前文档为准。上下文缓存文档
缓存更适合：
- 多个问题共享同一长文档；
- 系统提示和工具定义稳定且较长；
- 短时间内存在连续批量任务；
- Prompt 公共内容位于前缀，动态内容放在后部。
不适合：
- 每次请求的前缀都不同；
- 公共内容很短；
- 模板频繁变更；
- 为命中缓存而加入本来不需要的长文本。
缓存优化应看“命中后的实际 Token 与延迟”，不能只看是否开启。还要注意显式缓存的创建也可能产生额外成本，低复用场景未必划算。
七、重试必须有边界：区分可重试和不可重试错误
最危险的成本问题之一是无条件重试。若参数错误、权限不足或内容超出上下文限制，重复请求通常不会自动成功。
推荐将错误分为三类：
| 类型 | 示例 | 处理方式 |
| 瞬时错误 | 网络抖动、临时服务不可用 | 指数退避并限制次数 |
| 限流错误 | 请求过快、并发超限 | 读取服务提示，延迟或排队 |
| 永久错误 | 参数错误、认证失败、上下文过长 | 不重试，直接修正或降级 |
带抖动的指数退避示例：
import random
import time
RETRYABLE = {"timeout", "temporary_unavailable", "rate_limited"}
def retry_delay(attempt: int, base: float = 0.5, cap: float = 8.0) -> float:
delay = min(cap, base * (2 ** attempt))
return delay * random.uniform(0.8, 1.2)
def call_with_retry(call, max_attempts: int = 3):
for attempt in range(max_attempts):
result = call()
if result.ok:
return result
if result.error_code not in RETRYABLE:
raise RuntimeError(result.error_code)
if attempt == max_attempts - 1:
raise RuntimeError("retry_exhausted")
time.sleep(retry_delay(attempt))
生产代码还应根据官方接口的实际错误码、Retry-After 信息和幂等要求实现。对写入类工具调用，必须使用幂等键，避免重试造成重复发布、重复发送或重复扣款。
八、本地最小验证：预算如何阻断第4次调用
为了验证预算对象不会只停留在伪代码，我将完整示例保存为 agent_budget_demo.py。脚本仅使用 Python 标准库，模拟一个包含分类、摘要和草稿生成的任务。
运行环境与命令：
Python 3.12
python examples/agent_budget_demo.py
任务预算设置为：最多3次模型调用、累计输入不超过10000 Token、累计预留输出不超过2000 Token。实际输出为：
request=1 route=lightweight calls=1 input=800 output_reserved=120
request=2 route=balanced calls=2 input=3800 output_reserved=720
request=3 route=balanced calls=3 input=8800 output_reserved=1720
request=4 blocked=call_limit
结果表明，前三次调用完成预留，第4次在真正请求模型前被 call_limit 阻断。这是本地逻辑验证，不包含真实模型调用，因此不能用来推算阿里云账单。接入实际接口后，还需要使用响应中的真实 Token 用量回填记录，并对估算值与实际值进行校正。
这个实验还暴露一个实现细节：如果只在请求完成后累计用量，超限请求已经发生；预算应在调用前预留，在响应后再以实际用量结算差额。
九、可观测性：让每个 Token 都能找到它属于哪次交付
仅查看总用量无法回答“哪个节点最贵、为什么重试、成本是否换来了质量”。需要为每个任务建立统一 trace，并让每次模型与工具调用成为一个 span。
建议记录以下字段：
{
"trace_id": "task-20260722-001",
"span_id": "draft-generator-03",
"workflow": "technical_article",
"workflow_version": "v1.4",
"prompt_version": "draft-v7",
"model": "configured-model-id",
"input_tokens": 0,
"output_tokens": 0,
"cached_tokens": 0,
"latency_ms": 0,
"retry_count": 0,
"tool_calls": [],
"status": "success",
"quality_result": "pending_review"
}
不要在日志中原样保存密钥、个人信息、客户机密或完整 Prompt。可以记录模板版本、内容哈希、脱敏后的错误摘要和必要的用量元数据。
四组核心指标
- 用量：输入/输出/缓存 Token、工具调用次数；
- 性能：总耗时、首 Token 延迟、各节点耗时；
- 可靠性：错误率、重试率、超时率、降级率；
- 业务质量：人工通过率、返工率、有效交付成本。
只有把技术用量与业务结果关联起来，才能判断一次高成本调用是否合理。
十、成本异常如何排查？
当每日成本突然升高时，按以下顺序排查：
- 请求数量是否上涨，是否存在重复任务或异常调用方；
- 单任务模型调用次数是否增加；
- 输入 Token 是否因历史消息或检索片段膨胀；
- 输出上限或停止条件是否发生变化；
- 重试率、工具失败率是否异常；
- 模型路由是否把大量 L1 任务送到 L3；
- 缓存命中率是否下降；
- Prompt、工作流或模型版本最近是否变更。
每次发布 Prompt 或工作流新版本时，应使用固定样本回归：比较质量、Token、耗时、工具次数和人工修改量。如果质量只提高很少，成本却大幅增加，就需要重新评估变更。
十一、小型智能体项目上线前的成本护栏清单
[ ] 模型单价通过配置维护，未写死在业务逻辑中
[ ] 每个节点有最大输入、最大输出和超时
[ ] 每个任务有模型次数、工具次数和重试预算
[ ] 低、中、高复杂度任务有明确路由规则
[ ] 历史消息、检索内容和工具说明按需加载
[ ] 缓存效果使用实际命中数据验证
[ ] 永久错误不会进入自动重试
[ ] 写入类工具具有幂等键和人工审批
[ ] 模型与工具调用均记录 trace/span 和用量
[ ] 日志完成敏感信息脱敏
[ ] 配置变更会运行质量与成本回归测试
[ ] 时间窗口预算超限时有降级或暂停策略
十二、结语：成本治理不是少用 AI，而是让调用服务于交付
小型智能体项目的效率，不应只用调用了多少模型衡量。更有意义的指标是，在可控成本内完成了多少通过验收的交付，并且每次异常都能够定位和修复。
从预算开始，在调用前限制输入、输出、次数和工具；通过模型分级、上下文压缩和缓存减少重复计算；用有边界的重试避免隐性循环；最后把 Token、延迟、错误和人工验收统一到同一条调用链中。
当每笔资源消耗都有任务、有负责人、有质量结果，AI 才真正成为 OPC 可持续运营的一部分。
