AI Agent 性能优化:从 Token 浪费到精准推理的工程调优
AI Agent 性能优化:从 Token 浪费到精准推理的工程调优

一、Agent 的 Token 账单:每次调用都在烧钱
AI Agent 的运行成本主要由 Token 消耗决定。一个未经优化的 Agent,单次任务消耗 5000-8000 Token 是常态。以 GPT-4o 的定价计算,单次调用成本约 0.04-0.06 美元。当 Agent 被集成到高频业务流程中(如客服对话、代码审查),日均调用量达到 10 万次时,月成本超过 12 万美元。
更严重的是,大量 Token 消耗在低效环节。通过分析 5000 次 Agent 调用的 Token 分布发现:系统提示词(System Prompt)占比约 35%,但其中 60% 的内容与当前任务无关;工具调用的返回结果占比约 25%,但 Agent 只使用了其中 30% 的字段;多轮对话的历史上下文占比约 20%,但早期对话的参考价值已很低。
Agent 优化的核心目标是:在不降低输出质量的前提下,减少每次调用的 Token 消耗和推理延迟。这需要从 Prompt 工程、工具调用、上下文管理三个维度系统性地优化。
二、Agent 性能瓶颈的定位与优化架构
Agent 的性能瓶颈分布在三个层面,每个层面需要不同的优化策略。
graph TB
subgraph Prompt 层
S1[系统提示词冗余] --> O1[动态 Prompt 组装]
S2[指令模糊导致重试] --> O2[结构化指令模板]
end
subgraph 工具层
S3[工具返回过多字段] --> O3[字段裁剪与摘要]
S4[工具调用串行等待] --> O4[并行调用与缓存]
end
subgraph 上下文层
S5[历史对话无限增长] --> O5[滑动窗口 + 摘要压缩]
S6[无关上下文干扰] --> O6[相关性过滤]
end
O1 --> R[Token 消耗降低 40-60%]
O2 --> R
O3 --> R
O4 --> R
O5 --> R
O6 --> R
Prompt 层优化:系统提示词不应是静态的完整文档,而应根据当前任务动态组装。例如,当 Agent 只需要查询数据库时,不需要加载 API 调用相关的指令。动态组装可以将系统提示词从 2000 Token 压缩到 800 Token。
工具层优化:工具的返回结果往往是全量 JSON,但 Agent 只需要其中几个关键字段。在工具层增加结果裁剪,只返回必要字段,可以将工具返回的 Token 量减少 60-70%。对于独立的工具调用,可以并行执行而非串行等待。
上下文层优化:多轮对话的上下文无限增长是 Token 浪费的重灾区。滑动窗口只保留最近 N 轮对话,早期对话通过 LLM 生成摘要压缩。实测数据:10 轮对话的完整上下文约 4000 Token,摘要压缩后约 800 Token,压缩率 80%。
三、生产级 Agent 优化方案实现
以下用 Python 实现一个优化后的 Agent 框架,包含动态 Prompt、工具结果裁剪和上下文压缩。
3.1 动态 Prompt 组装器
from dataclasses import dataclass, field
from typing import Callable, Optional
@dataclass
class PromptModule:
"""Prompt 模块,按需加载"""
name: str
content: str
condition: Callable[[], bool] # 加载条件,返回 True 时才注入
token_estimate: int = 0
class DynamicPromptAssembler:
"""动态 Prompt 组装器,根据任务特征按需加载模块"""
def __init__(self):
self.modules: list[PromptModule] = []
def register(self, module: PromptModule):
self.modules.append(module)
def assemble(self, max_tokens: int = 1500) -> str:
"""
组装系统提示词,在 Token 预算内优先加载高优先级模块
按条件过滤 → 按 Token 预算裁剪 → 拼接输出
"""
# 第一步:过滤出当前任务需要的模块
active_modules = [m for m in self.modules if m.condition()]
# 第二步:按重要性排序(Token 估算值小的优先,确保核心指令被加载)
active_modules.sort(key=lambda m: m.token_estimate)
# 第三步:在 Token 预算内组装
result_parts = []
used_tokens = 0
for module in active_modules:
if used_tokens + module.token_estimate <= max_tokens:
result_parts.append(module.content)
used_tokens += module.token_estimate
return "\n\n".join(result_parts)
# 使用示例:注册按需加载的 Prompt 模块
assembler = DynamicPromptAssembler()
assembler.register(PromptModule(
name="core_identity",
content="你是一个数据库查询助手。根据用户问题生成 SQL 查询。",
condition=lambda: True, # 核心身份始终加载
token_estimate=50,
))
assembler.register(PromptModule(
name="api_instructions",
content="当需要调用外部 API 时,使用以下格式:...",
condition=lambda: current_task_needs_api(), # 仅当任务涉及 API 时加载
token_estimate=300,
))
assembler.register(PromptModule(
name="error_handling",
content="当查询失败时,分析错误原因并建议修正方案...",
condition=lambda: current_task_needs_api(), # API 调用才需要错误处理指令
token_estimate=200,
))
3.2 工具结果裁剪与并行调用
import asyncio
import json
from typing import Any
class ToolResultTrimmer:
"""工具结果裁剪器,只保留 Agent 需要的字段"""
def __init__(self):
# 每个工具的保留字段配置,避免返回全量 JSON
self.field_configs: dict[str, list[str]] = {}
def configure(self, tool_name: str, keep_fields: list[str]):
"""配置工具的保留字段列表"""
self.field_configs[tool_name] = keep_fields
def trim(self, tool_name: str, result: dict) -> dict:
"""裁剪工具返回结果,只保留配置的字段"""
keep_fields = self.field_configs.get(tool_name)
if not keep_fields:
return result # 未配置裁剪规则,返回原始结果
trimmed = {}
for field_name in keep_fields:
if field_name in result:
trimmed[field_name] = result[field_name]
return trimmed
class ParallelToolExecutor:
"""并行工具执行器,独立调用可并行,依赖调用串行"""
def __init__(self, max_concurrency: int = 5):
self.max_concurrency = max_concurrency
self.semaphore = asyncio.Semaphore(max_concurrency)
self.cache: dict[str, Any] = {} # 简易缓存,相同参数不重复调用
async def execute(
self,
calls: list[dict],
) -> list[dict]:
"""
并行执行工具调用列表
每个调用格式: {"tool": "tool_name", "args": {...}, "depends_on": []}
depends_on 非空时串行执行,为空时并行执行
"""
# 按依赖关系分组
independent = [c for c in calls if not c.get("depends_on")]
dependent = [c for c in calls if c.get("depends_on")]
# 并行执行无依赖的调用
independent_results = await asyncio.gather(*[
self._execute_single(call) for call in independent
])
# 串行执行有依赖的调用
dependent_results = []
for call in dependent:
result = await self._execute_single(call)
dependent_results.append(result)
# 按原始顺序合并结果
all_results = list(independent_results) + dependent_results
return all_results
async def _execute_single(self, call: dict) -> dict:
"""执行单个工具调用,含缓存和并发控制"""
cache_key = f"{call['tool']}:{json.dumps(call['args'], sort_keys=True)}"
# 缓存命中,直接返回,避免重复调用
if cache_key in self.cache:
return self.cache[cache_key]
async with self.semaphore: # 控制并发度
# 这里对接实际的工具执行逻辑
result = await self._invoke_tool(call["tool"], call["args"])
self.cache[cache_key] = result
return result
async def _invoke_tool(self, tool_name: str, args: dict) -> dict:
"""实际工具调用,由子类实现"""
raise NotImplementedError
3.3 上下文压缩:滑动窗口 + 摘要
class ContextCompressor:
"""上下文压缩器,滑动窗口 + 摘要压缩"""
def __init__(
self,
llm_client,
window_size: int = 5, # 保留最近 N 轮完整对话
summary_max_tokens: int = 200, # 摘要最大 Token 数
):
self.llm = llm_client
self.window_size = window_size
self.summary_max_tokens = summary_max_tokens
self.summary: str = "" # 早期对话的压缩摘要
def compress(self, messages: list[dict]) -> list[dict]:
"""
压缩对话历史:
1. 早期对话(超出窗口部分)→ 生成摘要
2. 近期对话(窗口内)→ 保留完整内容
3. 摘要作为系统消息插入上下文头部
"""
if len(messages) <= self.window_size:
return messages
# 分割:早期对话 vs 近期对话
early_messages = messages[:-self.window_size]
recent_messages = messages[-self.window_size:]
# 将早期对话压缩为摘要(异步执行,避免阻塞主流程)
new_summary = self._generate_summary(early_messages)
if new_summary:
self.summary = new_summary
# 组装压缩后的上下文
compressed = []
if self.summary:
compressed.append({
"role": "system",
"content": f"对话历史摘要:{self.summary}",
})
compressed.extend(recent_messages)
return compressed
def _generate_summary(self, messages: list[dict]) -> Optional[str]:
"""使用 LLM 生成对话摘要"""
# 将消息格式化为文本
conversation_text = "\n".join([
f"{m['role']}: {m['content'][:200]}" # 截断单条消息,控制输入长度
for m in messages
])
prompt = f"""请将以下对话历史压缩为一段简洁的摘要,保留关键决策和结论。
摘要不超过 {self.summary_max_tokens} 个 Token。
对话历史:
{conversation_text}
摘要:"""
# 同步调用 LLM 生成摘要(生产环境应改为异步)
try:
summary = self.llm.generate(
prompt,
max_tokens=self.summary_max_tokens,
temperature=0.0,
)
return summary.strip()
except Exception:
# 摘要生成失败时,保留旧摘要,不影响主流程
return None
关键设计决策:动态 Prompt 组装按 Token 预算裁剪,确保核心指令始终被加载。工具结果裁剪在工具层执行,Agent 只看到必要字段,减少 Token 消耗的同时降低信息噪声。上下文压缩使用"摘要 + 滑动窗口"而非简单截断,保留早期对话的关键决策信息。
四、Agent 优化的工程代价与适用边界
摘要压缩的信息损失。LLM 生成的摘要可能遗漏关键细节,特别是数值型数据(如具体的配置参数、错误码)。在需要精确回溯的场景中,摘要压缩不可靠。建议对包含数值和代码的对话段保留原文,仅对讨论性内容生成摘要。
并行调用的依赖风险。并行执行工具调用时,如果两个工具存在隐式依赖(如工具 B 的参数依赖工具 A 的返回值),并行执行会产生错误结果。必须在调用图层面显式声明依赖关系,增加编排复杂度。
缓存的时效性问题。工具调用结果的缓存可能导致返回过期数据。对于实时性要求高的工具(如库存查询),缓存不可用;对于低频变化的数据(如用户画像),缓存可显著减少重复调用。
不适用场景:单轮问答型 Agent(无上下文累积),优化空间有限;对输出质量要求极高且成本不敏感的场景(如医疗诊断),Token 节省不值得以任何质量损失为代价。
五、总结
AI Agent 的性能优化需要从 Prompt、工具、上下文三个维度系统性推进。动态 Prompt 组装减少系统提示词的 Token 浪费,工具结果裁剪降低信息噪声,上下文压缩控制多轮对话的 Token 增长。每项优化的效果都应通过 A/B 测试量化验证。
落地路线建议:第一步,分析 Agent 的 Token 消耗分布,定位最大浪费点;第二步,实现动态 Prompt 组装,按任务特征加载必要指令;第三步,为高频工具配置结果裁剪规则,减少返回字段;第四步,引入滑动窗口 + 摘要的上下文压缩机制。每一步优化前后都应记录 Token 消耗和输出质量的对比数据,确保优化不牺牲质量。
更多推荐



所有评论(0)