如何减少 AI 智能体的 Token 消耗
我们测量了同一处 Python 代码修复在两种输入下的成本:完整源文件,以及仅包含相关代码的片段。通过 VOID API 向 claude-haiku-4.5 分别发送了三次请求。每次请求的输入从 1756 个 Token 降到 359 个,六份结果均通过相同的行为检查。 下文给出任务、配置、实际扣费和复现代码。这是针对一个代码修改步骤的受控测试,展示了缩减上下文对该任务的影响;完整智能体会话的费用需要单独测量。
修复机器人无法关闭记忆的问题
最初的 bot.py 使用 [-2 * HISTORY_TURNS:] 裁剪历史。HISTORY_TURNS=3 时,它保留最近三轮问答;值为 0 时,表达式变成 [-0:],反而保留整个列表。因此,关闭记忆的配置没有正确生效。
我们要求模型将这段逻辑提取到 keep_history(history, turns)。约定如下:零返回空列表,负数抛出 ValueError,正数保留最近的若干轮问答,同时不能修改原始列表。
这个任务可以用简短的行为检查验证,因此能够在比较费用的同时,确认两种上下文都得到了符合要求的结果。
精简方案使用的代码片段
HISTORY_TURNS = 3
history = context.user_data.get("history", [])
context.user_data["history"] = (
history + [user_message, {"role": "assistant", "content": reply}]
)[-2 * HISTORY_TURNS:]比较条件
测试于 2026 年 10 月 23 日执行,通过 /v1/chat/completions 调用 claude-haiku-4.5。两种方案使用相同的系统指令和任务,设置 temperature=0、max_tokens=400,并关闭 SDK 自动重试。
完整方案在任务后附上最初的整个 bot.py。精简方案只附上 HISTORY_TURNS 配置和历史裁剪代码。原始文件和比较脚本都列在下方。
六次请求的顺序为:完整、精简、精简、完整、完整、精简。每次都开始新的对话。请求中没有包含智能体读取文件或运行命令的工具,模型返回的代码由独立的 Python 检查程序验证。
实测任务和源文件包含俄语文本。为便于复现,提示词代码块和可执行示例保留了原始字符串。上一节已用中文说明函数约定。翻译提示词会改变输入,也可能改变 Token 数量。
两种方案使用的相同任务(原始俄语提示词)
В Python-боте история ограничивается выражением
(history + [user_message, assistant_message])[-2 * HISTORY_TURNS:].
При HISTORY_TURNS=0 оно сохраняет всю историю вместо отключения памяти.
Напиши функцию keep_history(history, turns), которая исправляет это поведение.
history содержит полный список чередующихся сообщений user/assistant, только полные пары.
turns - целое число. При turns=0 верни пустой список; при turns<0 подними ValueError.
При turns>0 верни последние turns полных пар или все сообщения, если их меньше.
Всегда возвращай новый список и не меняй исходный.
Верни только одну функцию keep_history на Python без Markdown, импортов и объяснений.六次请求的 Token 数量与扣费
每组中的三次请求返回了相同的输入和输出 Token 数量。完整文件使用 1756 个输入 Token 和 59 个输出 Token;代码片段使用 359 个输入 Token 和 64 个输出 Token。六次响应中的 cached_tokens 都为 0。
表中费用来自 VOID 请求记录。实测时的价格为每百万输入 Token 0.57 美元、每百万输出 Token 2.20 美元。
六份函数均通过了八项行为检查。测试前,我们去掉了回答中的 Markdown 代码围栏,没有修改 Python 代码本身。精简方案多复制了一次列表,但仍满足所要求的行为。
本次测试的费用下降来自输入减少。模型并没有输出更短的答案:精简方案的输出反而从 59 个 Token 增加到 64 个。换一个修复任务、模型或历史上下文,结果可能不同。
实测结果
完整文件
请求次数:3
每次输入:1756 Token
每次输出:59 Token
每次费用:$0.00113072
三次总费用:$0.00339216
代码片段
请求次数:3
每次输入:359 Token
每次输出:64 Token
每次费用:$0.00034543
三次总费用:$0.00103629完整文件方案返回的函数
def keep_history(history, turns):
if turns < 0:
raise ValueError("turns must be non-negative")
if turns == 0:
return []
return history[-2 * turns:]精简上下文方案返回的函数
def keep_history(history, turns):
if turns < 0:
raise ValueError("turns must be non-negative")
if turns == 0:
return []
return (history + [])[-(2 * turns):]如何验证结果
在输入包含三轮历史问答的情况下,分别检查保留零、一、二、三和九轮。每种情况都必须返回一个新列表,且不能修改原列表。另外两项检查使用空历史,最后一项检查要求负数轮次抛出 ValueError。
自行检查生成的函数时,请先阅读代码,再去掉 Markdown 代码围栏并保存为 solution.py。将下方代码保存为 test_history.py,运行 python test_history.py。这些检查基于既定输入约定:历史由完整的 user/assistant 消息对组成。
test_history.py(保留原始测试提示文字)
from solution import keep_history
history = [
{"role": role, "content": str(turn)}
for turn in range(3)
for role in ("user", "assistant")
]
original = list(history)
for turns, expected in (
(0, []),
(1, history[-2:]),
(2, history[-4:]),
(3, history),
(9, history),
):
result = keep_history(history, turns)
assert result == expected
assert result is not history
assert history == original
assert keep_history([], 0) == []
assert keep_history([], 3) == []
try:
keep_history(history, -1)
except ValueError:
pass
else:
raise AssertionError("Отрицательное число пар должно вызывать ValueError")
print("8 проверок поведения пройдены")
如何应用到自己的智能体
进行局部修改时,先找到相关函数,再连同要求一起交给模型。如果问题涉及调用方、类型或配置,再补充那些代码。在本次任务中,Telegram 消息处理、客户端初始化和配额存储都不影响 keep_history 的结果。
对于很长的日志,可将完整内容单独保存,只向模型传入失败测试名称、出错位置和诊断信息。如果精简后又需要读取同一份日志,比较费用时必须计入这些额外调用。
使用工具时,应将工具调用和对应结果一起保留。仅按列表元素数量裁剪消息,可能留下找不到对应调用的工具响应。此类对话需要使用理解消息结构的上下文压缩方式。
修改上下文后,运行相同的结果检查。错误答案即使更便宜,也不能证明完成任务的成本降低了。
用脚本重复比较
使用机器人教程中的 Python 环境和 API 密钥。这个脚本需要 openai==3.37.0 和 python-dotenv==1.2.4。在 .env 中设置 VOID_API_KEY 和 VOID_MODEL=claude-haiku-4.5,不需要 Telegram 令牌。
将下方初始机器人源文件保存为 bot-baseline.py,将比较脚本保存为 compare_context.py。运行 python compare_context.py bot-baseline.py。脚本会把源文件当作文本读取,并发起六次付费模型请求,不会执行机器人源文件。
每次请求结束后,回答和 usage 都会保存到 comparison.json。syntax_ok 只检查语法,行为需要用上一节的代码验证。如果某次请求失败,脚本会停止,不会自动重试。
模型版本、文件格式和缓存状态都可能使重复测试得到不同数字。针对自己的任务,应比较达到验收标准的总费用,包括失败尝试和重复读取。
compare_context.py(保留实测的原始提示词)
import ast
from datetime import datetime, timezone
import json
import os
from pathlib import Path
import sys
from dotenv import load_dotenv
from openai import OpenAI
load_dotenv(Path(__file__).resolve().parent / ".env")
model = os.environ["VOID_MODEL"]
task = """В Python-боте история ограничивается выражением
(history + [user_message, assistant_message])[-2 * HISTORY_TURNS:].
При HISTORY_TURNS=0 оно сохраняет всю историю вместо отключения памяти.
Напиши функцию keep_history(history, turns), которая исправляет это поведение.
history содержит полный список чередующихся сообщений user/assistant, только полные пары.
turns - целое число. При turns=0 верни пустой список; при turns<0 подними ValueError.
При turns>0 верни последние turns полных пар или все сообщения, если их меньше.
Всегда возвращай новый список и не меняй исходный.
Верни только одну функцию keep_history на Python без Markdown, импортов и объяснений."""
short_context = """HISTORY_TURNS = 3
history = context.user_data.get("history", [])
context.user_data["history"] = (
history + [user_message, {"role": "assistant", "content": reply}]
)[-2 * HISTORY_TURNS:]"""
if len(sys.argv) != 2:
raise SystemExit("Запуск: python compare_context.py bot.py")
full_context = Path(sys.argv[1]).read_text()
report = {"model": model, "date": datetime.now(timezone.utc).isoformat(), "runs": []}
output = Path(__file__).resolve().parent / "comparison.json"
with OpenAI(
api_key=os.environ["VOID_API_KEY"],
base_url="https://void-api.tech/v1",
timeout=75,
max_retries=0,
) as client:
for variant in ("full", "focused", "focused", "full", "full", "focused"):
context = full_context if variant == "full" else short_context
response = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": "Ты исправляешь код Python. Соблюдай заданный контракт функции."},
{"role": "user", "content": task + "\n\nИсходный код:\n" + context},
],
temperature=0,
max_tokens=400,
)
code = response.choices[0].message.content or ""
if code.startswith("```python\n") and code.rstrip().endswith("```"):
code = code.strip()[len("```python\n"):-3].strip()
try:
ast.parse(code)
syntax_ok = True
except SyntaxError:
syntax_ok = False
row = {
"variant": variant,
"response_id": response.id,
"usage": response.usage.model_dump() if response.usage else None,
"syntax_ok": syntax_ok,
"answer": code,
}
report["runs"].append(row)
output.write_text(json.dumps(report, ensure_ascii=False, indent=2))
print(json.dumps({"variant": variant, "usage": row["usage"], "syntax_ok": syntax_ok}, ensure_ascii=False))
print(f"Ответы и токены сохранены в {output.name}. Проверьте поведение функций отдельно от синтаксиса.")
实测使用的原始文件
这是完整上下文方案中传给模型的初始文件。为复现任务,其中保留了 HISTORY_TURNS=0 时的错误逻辑。实际使用机器人时,请采用机器人教程中的修复版本。
将这个代码块保存为 bot-baseline.py。比较 Token 用量时,无需执行该文件。
bot-baseline.py:实验的原始输入
import logging
import os
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from dotenv import load_dotenv
from openai import APIError, AsyncOpenAI
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters
ROOT = Path(__file__).resolve().parent
load_dotenv(ROOT / ".env")
OWNER_ID = int(os.getenv("TELEGRAM_ALLOWED_USER_ID", "0"))
MODEL = os.environ["VOID_MODEL"]
DAILY_REQUESTS = 20
HISTORY_TURNS = 3
MAX_INPUT_CHARS = 3000
MAX_OUTPUT_TOKENS = 600
logging.basicConfig(level=logging.WARNING)
client = AsyncOpenAI(
api_key=os.environ["VOID_API_KEY"],
base_url="https://void-api.tech/v1",
timeout=60.0,
max_retries=0,
)
db = sqlite3.connect(ROOT / "quota.sqlite3")
db.execute("CREATE TABLE IF NOT EXISTS quota (day TEXT PRIMARY KEY, used INTEGER NOT NULL)")
db.commit()
def reserve_request():
day = datetime.now(timezone.utc).date().isoformat()
with db:
cursor = db.execute(
"INSERT INTO quota(day, used) VALUES (?, 1) "
"ON CONFLICT(day) DO UPDATE SET used = used + 1 WHERE used < ?",
(day, DAILY_REQUESTS),
)
return cursor.rowcount == 1
async def start(update, context):
await update.message.reply_text(
f"Ваш Telegram ID: {update.effective_user.id}.\n"
"Укажите его в TELEGRAM_ALLOWED_USER_ID и перезапустите бота.\n"
"Команда /reset очищает историю диалога."
)
async def reset(update, context):
if update.effective_user.id != OWNER_ID:
return
context.user_data.clear()
await update.message.reply_text("История очищена. Дневной счётчик сохранён.")
async def answer(update, context):
if update.effective_user.id != OWNER_ID:
await update.message.reply_text("Это личный бот. Доступ к модели закрыт.")
return
text = update.message.text.strip()
if not text or len(text) > MAX_INPUT_CHARS:
await update.message.reply_text(f"Отправьте от 1 до {MAX_INPUT_CHARS} символов.")
return
if not reserve_request():
await update.message.reply_text("Дневной лимит исчерпан. Обновление в 00:00 UTC.")
return
history = context.user_data.get("history", [])
user_message = {"role": "user", "content": text}
messages = [
{"role": "system", "content": "Отвечай по-русски, кратко и по существу."},
*history,
user_message,
]
try:
response = await client.chat.completions.create(
model=MODEL,
messages=messages,
max_tokens=MAX_OUTPUT_TOKENS,
)
except APIError as error:
logging.warning(
"Void API: %s, HTTP %s", type(error).__name__, getattr(error, "status_code", None)
)
await update.message.reply_text(
"Не удалось получить ответ. Проверьте ключ, баланс и модель в VOID. "
"Попытка учтена в дневном лимите."
)
return
choice = response.choices[0] if response.choices else None
reply = (choice.message.content or "").strip() if choice else ""
if not reply:
await update.message.reply_text("Модель не вернула текст. Попытка учтена в лимите.")
return
context.user_data["history"] = (
history + [user_message, {"role": "assistant", "content": reply}]
)[-2 * HISTORY_TURNS:]
for offset in range(0, len(reply), 2000):
await update.message.reply_text(reply[offset : offset + 2000])
if choice.finish_reason == "length":
await update.message.reply_text("Ответ остановлен по лимиту токенов.")
if response.usage:
print(
f"model={response.model} input={response.usage.prompt_tokens} "
f"output={response.usage.completion_tokens}",
flush=True,
)
async def on_error(update, context):
logging.warning("Ошибка обработчика: %s", type(context.error).__name__)
async def shutdown(application):
await client.close()
db.close()
def build_app():
app = (
Application.builder()
.token(os.environ["TELEGRAM_BOT_TOKEN"])
.concurrent_updates(False)
.post_shutdown(shutdown)
.build()
)
private = filters.ChatType.PRIVATE
app.add_handler(CommandHandler("start", start, filters=private))
app.add_handler(CommandHandler("reset", reset, filters=private))
app.add_handler(MessageHandler(private & filters.TEXT & ~filters.COMMAND, answer))
app.add_error_handler(on_error)
return app
if __name__ == "__main__":
build_app().run_polling(allowed_updates=[Update.MESSAGE], drop_pending_updates=True)
这个例子的适用范围
我们测试的是一个有明确约定的小函数。这个结果不能确定大型重构、跨多个文件排查错误或包含网页搜索的任务能节省多少费用。
更换模型时,应比较多项已完成任务的费用和失败次数。使用缓存时,请依据实际缓存字段及对应价格计算;cached_tokens 可能已包含在输入总量中,不能重复相加。
首次比较自己的智能体时,可以选择一项可重复的任务,记录初始文件和验收标准。随后只改变一个因素,例如传入代码的多少,再比较符合要求的结果所花费的成本。