Agent 开发实战：成本失控的 5 个解决方案
·
Agent 开发实战：成本失控的 5 个解决方案
0. 痛点：一天烧掉 500 块，还不知道钱花哪了
你有没有遇到过这种情况：
早上 9:00 - 检查账单
Azure OpenAI：昨日消费 ¥512.00
Token 用量：2,340,000 Token（输入 1.2M，输出 1.14M）
问题：
- 哪个 Agent 花的？
- 哪个用户花的？
- 花在哪次对话？
- 有没有浪费？
或者更糟：
用户：帮我写个 10 万字的武侠小说
Agent：好的，让我开始创作...
[循环调用 LLM 100 次]
[每次 2000 Token 输出]
[总成本：100 × ¥0.06 = ¥6.00]
[用户只问了 8 个字]
这就是成本失控（Cost Runaway）。
1. 问题根源：为什么 Agent 成本会失控？
原因 1：Token 消耗无监控
# 没有 Token 计数
def call_llm(prompt: str) -> str:
"""调用 LLM（没有记录 Token 消耗）"""
response = openai.ChatCompletion.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
# 问题：不知道这次调用花了多少 Token
原因 2：没有成本预算
# 没有预算限制
def process_user_query(user_id: str, query: str) -> str:
"""处理用户问题（没有预算检查）"""
# 不管用户已经花了多少钱，继续调用
return call_llm(query)
# 问题：恶意用户可能无限调用，刷爆账单
原因 3：模型选择不合理
# 所有任务都用最贵的模型
def simple_task() -> str:
"""简单任务（用 GPT-4）"""
return call_gpt4("将这段中文翻译成英文：" + text)
# 问题：简单任务用 GPT-3.5 就够了，浪费钱
解决方案 1：Token 计数和成本追踪（Token Counting & Cost Tracking）
原理
每次调用 LLM 都记录 Token 消耗，计算成本。
from typing import Dict, Any, List
import time
class TokenCounter:
"""
Token 计数器
支持模型：
- GPT-4: ¥0.06/1K Token（输入），¥0.12/1K Token（输出）
- GPT-3.5: ¥0.0015/1K Token（输入），¥0.002/1K Token（输出）
- Claude-3.5-Sonnet: ¥0.045/1K Token（输入），¥0.225/1K Token（输出）
"""
# 模型价格（人民币/1K Token）
PRICES = {
"gpt-4": {"input": 0.06, "output": 0.12},
"gpt-3.5-turbo": {"input": 0.0015, "output": 0.002},
"claude-3.5-sonnet": {"input": 0.045, "output": 0.225},
}
def __init__(self):
self.records: List[Dict] = [] # 调用记录
self.total_cost: float = 0.0
def count(self, model: str, input_tokens: int, output_tokens: int) -> Dict[str, Any]:
"""
计算一次调用的 Token 消耗和成本
Args:
model: 模型名
input_tokens: 输入 Token 数
output_tokens: 输出 Token 数
Returns:
{
"input_tokens": 1234,
"output_tokens": 567,
"input_cost": 0.07404,
"output_cost": 0.06804,
"total_cost": 0.14208
}
"""
# 1. 获取模型价格
if model not in self.PRICES:
raise ValueError(f"未知模型：{model}")
price = self.PRICES[model]
# 2. 计算成本
input_cost = (input_tokens / 1000) * price["input"]
output_cost = (output_tokens / 1000) * price["output"]
total_cost = input_cost + output_cost
# 3. 记录
record = {
"timestamp": time.time(),
"model": model,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"input_cost": input_cost,
"output_cost": output_cost,
"total_cost": total_cost
}
self.records.append(record)
self.total_cost += total_cost
return {
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"input_cost": input_cost,
"output_cost": output_cost,
"total_cost": total_cost
}
def get_stats(self) -> Dict[str, Any]:
"""获取统计信息"""
if not self.records:
return {"total_cost": 0.0, "total_calls": 0}
total_input_tokens = sum(r["input_tokens"] for r in self.records)
total_output_tokens = sum(r["output_tokens"] for r in self.records)
return {
"total_cost": self.total_cost,
"total_calls": len(self.records),
"total_input_tokens": total_input_tokens,
"total_output_tokens": total_output_tokens,
"avg_cost_per_call": self.total_cost / len(self.records)
}
def reset(self):
"""重置计数器"""
self.records = []
self.total_cost = 0.0
# 使用示例
counter = TokenCounter()
# 模拟几次调用
counter.count("gpt-4", 1000, 500) # ¥0.12
counter.count("gpt-4", 2000, 1000) # ¥0.24
counter.count("gpt-3.5-turbo", 1000, 500) # ¥0.0025
# 查看统计
stats = counter.get_stats()
print(f"总成本：¥{stats['total_cost']:.4f}")
print(f"总调用次数：{stats['total_calls']}")
print(f"平均每次调用成本：¥{stats['avg_cost_per_call']:.4f}")
集成到 LLM 调用
class LLMWithTokenCounting:
"""带 Token 计数的 LLM 调用"""
def __init__(self, model: str, counter: TokenCounter):
self.model = model
self.counter = counter
self.client = openai.OpenAI(api_key="your-api-key")
def call(self, prompt: str) -> str:
"""调用 LLM（带 Token 计数）"""
# 1. 调用 LLM
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}]
)
# 2. 统计 Token
input_tokens = response.usage.prompt_tokens
output_tokens = response.usage.completion_tokens
cost_info = self.counter.count(self.model, input_tokens, output_tokens)
print(f"本次调用成本：¥{cost_info['total_cost']:.4f}")
return response.choices[0].message.content
# 使用
counter = TokenCounter()
llm = LLMWithTokenCounting(model="gpt-4", counter=counter)
# 调用 LLM
response = llm.call("今天天气怎么样？")
# 查看总成本
stats = counter.get_stats()
print(f"总成本：¥{stats['total_cost']:.4f}")
优点：精准追踪成本，发现问题
缺点：增加少量开销（可忽略）
解决方案 2：成本预算和限流（Cost Budget & Rate Limiting）
原理
给用户或 Agent 设置成本预算，超过就拒绝服务。
from typing import Dict, Any
import time
class CostBudgetManager:
"""
成本预算管理
策略：
1. 用户级别预算：每个用户每天最多花 ¥10
2. Agent 级别预算：每个 Agent 每天最多花 ¥50
3. 全局预算：所有用户每天最多花 ¥500
"""
def __init__(self, user_daily_budget: float = 10.0, agent_daily_budget: float = 50.0, global_daily_budget: float = 500.0):
"""
Args:
user_daily_budget: 用户每日预算（元）
agent_daily_budget: Agent 每日预算（元）
global_daily_budget: 全局每日预算（元）
"""
self.user_daily_budget = user_daily_budget
self.agent_daily_budget = agent_daily_budget
self.global_daily_budget = global_daily_budget
# 成本记录：{user_id: [(timestamp, cost), ...]}
self.user_costs: Dict[str, List[tuple]] = {}
# 成本记录：{agent_id: [(timestamp, cost), ...]}
self.agent_costs: Dict[str, List[tuple]] = {}
# 全局成本记录：[(timestamp, cost), ...]
self.global_costs: List[tuple] = []
def check_budget(self, user_id: str, agent_id: str, estimated_cost: float) -> bool:
"""
检查预算是否充足
Returns:
True: 预算充足，可以调用
False: 预算不足，拒绝调用
"""
# 1. 检查用户预算
user_cost_today = self._get_cost_today(self.user_costs.get(user_id, []))
if user_cost_today + estimated_cost > self.user_daily_budget:
print(f"【预算不足】用户 {user_id} 今日预算已用尽（已花 ¥{user_cost_today:.4f}，预算 ¥{self.user_daily_budget}）")
return False
# 2. 检查 Agent 预算
agent_cost_today = self._get_cost_today(self.agent_costs.get(agent_id, []))
if agent_cost_today + estimated_cost > self.agent_daily_budget:
print(f"【预算不足】Agent {agent_id} 今日预算已用尽（已花 ¥{agent_cost_today:.4f}，预算 ¥{self.agent_daily_budget}）")
return False
# 3. 检查全局预算
global_cost_today = self._get_cost_today(self.global_costs)
if global_cost_today + estimated_cost > self.global_daily_budget:
print(f"【预算不足】全局今日预算已用尽（已花 ¥{global_cost_today:.4f}，预算 ¥{self.global_daily_budget}）")
return False
return True
def record_cost(self, user_id: str, agent_id: str, cost: float):
"""记录一次成本"""
timestamp = time.time()
# 记录到用户
if user_id not in self.user_costs:
self.user_costs[user_id] = []
self.user_costs[user_id].append((timestamp, cost))
# 记录到 Agent
if agent_id not in self.agent_costs:
self.agent_costs[agent_id] = []
self.agent_costs[agent_id].append((timestamp, cost))
# 记录到全局
self.global_costs.append((timestamp, cost))
def _get_cost_today(self, records: List[tuple]) -> float:
"""获取今日的成本（UTC+8 时区）"""
today_start = self._get_today_start_timestamp()
return sum(cost for timestamp, cost in records if timestamp >= today_start)
def _get_today_start_timestamp(self) -> float:
"""获取今天 0 点的时间戳（UTC+8）"""
import datetime
now = datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=8)))
today_start = datetime.datetime(now.year, now.month, now.day, tzinfo=now.tzinfo)
return today_start.timestamp()
def get_budget_status(self) -> Dict[str, Any]:
"""获取预算状态"""
return {
"user_budgets": {
user_id: {
"spent": self._get_cost_today(records),
"budget": self.user_daily_budget,
"remaining": self.user_daily_budget - self._get_cost_today(records)
}
for user_id, records in self.user_costs.items()
},
"agent_budgets": {
agent_id: {
"spent": self._get_cost_today(records),
"budget": self.agent_daily_budget,
"remaining": self.agent_daily_budget - self._get_cost_today(records)
}
for agent_id, records in self.agent_costs.items()
},
"global_budget": {
"spent": self._get_cost_today(self.global_costs),
"budget": self.global_daily_budget,
"remaining": self.global_daily_budget - self._get_cost_today(self.global_costs)
}
}
# 使用示例
budget_manager = CostBudgetManager(
user_daily_budget=10.0,
agent_daily_budget=50.0,
global_daily_budget=500.0
)
# 模拟调用
user_id = "user_001"
agent_id = "agent_001"
# 检查预算
estimated_cost = 0.12 # 预估成本 ¥0.12
if budget_manager.check_budget(user_id, agent_id, estimated_cost):
# 调用 LLM
cost = 0.12 # 实际成本
budget_manager.record_cost(user_id, agent_id, cost)
print(f"调用成功，花费 ¥{cost:.4f}")
else:
print("预算不足，拒绝调用")
# 查看预算状态
status = budget_manager.get_budget_status()
print(json.dumps(status, indent=2, ensure_ascii=False))
集成到 Agent
class AgentWithCostBudget:
"""带成本预算的 Agent"""
def __init__(self, agent_id: str, budget_manager: CostBudgetManager):
self.agent_id = agent_id
self.budget_manager = budget_manager
self.llm = LLMWithTokenCounting(model="gpt-4", counter=TokenCounter())
def process(self, user_id: str, query: str) -> str:
"""处理用户问题（带预算检查）"""
# 1. 预估成本（根据历史平均）
estimated_cost = self._estimate_cost(query)
# 2. 检查预算
if not self.budget_manager.check_budget(user_id, self.agent_id, estimated_cost):
return "【预算不足】今日预算已用尽，请明天再试"
# 3. 调用 LLM
response = self.llm.call(query)
# 4. 记录成本
actual_cost = self.llm.counter.records[-1]["total_cost"]
self.budget_manager.record_cost(user_id, self.agent_id, actual_cost)
return response
def _estimate_cost(self, query: str) -> float:
"""预估成本（简化：按 Token 数估算）"""
# 假设平均输入 500 Token，输出 300 Token
input_tokens = 500
output_tokens = 300
price = TokenCounter.PRICES[self.llm.model]
input_cost = (input_tokens / 1000) * price["input"]
output_cost = (output_tokens / 1000) * price["output"]
return input_cost + output_cost
# 使用
budget_manager = CostBudgetManager(user_daily_budget=10.0)
agent = AgentWithCostBudget(agent_id="agent_001", budget_manager=budget_manager)
response = agent.process("user_001", "今天天气怎么样？")
print(response)
优点：防止成本失控，保护预算
缺点：可能误杀合理请求
解决方案 3：模型选择优化（Model Selection Optimization）
原理
根据任务难度，选择合适的模型（简单任务用便宜的模型）。
from typing import Literal
class SmartModelSelector:
"""
智能模型选择器
策略：
1. 简单任务 → GPT-3.5（便宜）
2. 中等任务 → GPT-4-turbo（性价比）
3. 复杂任务 → GPT-4（贵但准）
"""
# 模型价格（人民币/1K Token）
MODEL_PRICES = {
"gpt-3.5-turbo": {"input": 0.0015, "output": 0.002},
"gpt-4-turbo": {"input": 0.03, "output": 0.06},
"gpt-4": {"input": 0.06, "output": 0.12},
}
def __init__(self):
# 任务分类规则（简化：根据关键词）
self.rules = {
"simple": ["翻译", "总结", "提取关键词", "简单计算"],
"medium": ["写代码", "分析数据", "回答问题"],
"complex": ["写文章", "复杂推理", "多步骤任务"]
}
def select_model(self, task: str) -> Literal["gpt-3.5-turbo", "gpt-4-turbo", "gpt-4"]:
"""
根据任务选择合适的模型
Returns:
模型名
"""
# 1. 分类任务
task_type = self._classify_task(task)
# 2. 选择模型
if task_type == "simple":
return "gpt-3.5-turbo"
elif task_type == "medium":
return "gpt-4-turbo"
else: # complex
return "gpt-4"
def _classify_task(self, task: str) -> str:
"""分类任务（简化：关键词匹配）"""
for task_type, keywords in self.rules.items():
if any(kw in task for kw in keywords):
return task_type
# 默认：中等任务
return "medium"
def estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""估算成本"""
price = self.MODEL_PRICES[model]
input_cost = (input_tokens / 1000) * price["input"]
output_cost = (output_tokens / 1000) * price["output"]
return input_cost + output_cost
# 使用示例
selector = SmartModelSelector()
# 测试不同任务
tasks = [
"翻译这段中文：Hello World",
"写一段 Python 代码计算斐波那契数列",
"写一篇 5000 字的关于 Agent 开发的深度文章"
]
for task in tasks:
model = selector.select_model(task)
estimated_cost = selector.estimate_cost(model, input_tokens=500, output_tokens=300)
print(f"任务：{task[:30]}...")
print(f" 选择模型：{model}")
print(f" 预估成本：¥{estimated_cost:.4f}")
print()
更先进的模型选择：用 LLM 做 Judge
class LLMAsJudgeModelSelector:
"""用 LLM 做 Judge 选择模型"""
def __init__(self, judge_llm):
self.judge_llm = judge_llm
def select_model(self, task: str) -> str:
"""用 LLM 判断任务难度，选择模型"""
prompt = f"""
请判断以下任务的难度，并选择合适的模型：
任务：{task}
可选模型：
- gpt-3.5-turbo：简单任务（翻译、总结、提取关键词），价格 ¥0.0015/1K Token（输入）
- gpt-4-turbo：中等任务（写代码、分析数据），价格 ¥0.03/1K Token（输入）
- gpt-4：复杂任务（写文章、复杂推理），价格 ¥0.06/1K Token（输入）
输出格式（JSON）：
```json
{{
"difficulty": "simple/medium/complex",
"model": "gpt-3.5-turbo/gpt-4-turbo/gpt-4",
"reason": "选择理由"
}}
“”"
# 调用 Judge LLM（用便宜的模型）
response = self.judge_llm.call(prompt, model="gpt-3.5-turbo")
# 解析响应
try:
result = json.loads(response)
return result["model"]
except json.JSONDecodeError:
# 解析失败，返回默认模型
return "gpt-4-turbo"
**优点**：降低成本，提高性价比
**缺点**：分类错误可能导致质量下降
---
## 解决方案 4：缓存策略（Caching Strategy）
### 原理
缓存相同问题的回答，避免重复调用 LLM。
```python
from typing import Dict, Any, Optional
import hashlib
import json
class ResponseCache:
"""
回答缓存
策略：
1. 精确匹配：完全相同的问题，直接返回缓存
2. 语义匹配：相似的问题，返回最相似的缓存
"""
def __init__(self, similarity_threshold: float = 0.95):
"""
Args:
similarity_threshold: 语义相似度阈值（≥ 此值视为相同问题）
"""
self.cache: Dict[str, Any] = {} # {question_hash: response}
self.similarity_threshold = similarity_threshold
def get(self, question: str) -> Optional[str]:
"""
获取缓存的回答
Returns:
缓存的回答（如果有），否则 None
"""
# 1. 精确匹配
question_hash = self._hash(question)
if question_hash in self.cache:
print(f"【缓存命中】精确匹配")
return self.cache[question_hash]
# 2. 语义匹配（简化：用编辑距离）
for cached_question, cached_response in self.cache.items():
similarity = self._calculate_similarity(question, cached_question)
if similarity >= self.similarity_threshold:
print(f"【缓存命中】语义匹配（相似度：{similarity:.2f}）")
return cached_response
return None
def set(self, question: str, response: str):
"""缓存回答"""
question_hash = self._hash(question)
self.cache[question_hash] = response
def _hash(self, text: str) -> str:
"""计算哈希值"""
return hashlib.md5(text.encode("utf-8")).hexdigest()
def _calculate_similarity(self, text1: str, text2: str) -> float:
"""计算语义相似度（简化：用编辑距离）"""
# 简化：用 Levenshtein 距离
distance = self._levenshtein_distance(text1, text2)
max_len = max(len(text1), len(text2))
similarity = 1 - (distance / max_len)
return similarity
def _levenshtein_distance(self, s1: str, s2: str) -> int:
"""计算 Levenshtein 距离"""
if len(s1) < len(s2):
return self._levenshtein_distance(s2, s1)
if len(s2) == 0:
return len(s1)
previous_row = range(len(s2) + 1)
for i, c1 in enumerate(s1):
current_row = [i + 1]
for j, c2 in enumerate(s2):
insertions = previous_row[j + 1] + 1
deletions = current_row[j] + 1
substitutions = previous_row[j] + (c1 != c2)
current_row.append(min(insertions, deletions, substitutions))
previous_row = current_row
return previous_row[-1]
# 使用示例
cache = ResponseCache(similarity_threshold=0.9)
# 第一次调用（未命中缓存）
question1 = "今天天气怎么样？"
cached_response1 = cache.get(question1)
if cached_response1 is None:
print("【缓存未命中】调用 LLM...")
response1 = "今天天气晴朗，温度 25°C" # 模拟 LLM 调用
cache.set(question1, response1)
print(f"回答：{response1}")
else:
print(f"回答：{cached_response1}")
# 第二次调用（精确匹配）
question2 = "今天天气怎么样？"
cached_response2 = cache.get(question2)
if cached_response2 is None:
print("【缓存未命中】调用 LLM...")
else:
print(f"回答：{cached_response2}")
# 第三次调用（语义匹配）
question3 = "今天天气如何？"
cached_response3 = cache.get(question3)
if cached_response3 is None:
print("【缓存未命中】调用 LLM...")
else:
print(f"回答：{cached_response3}")
集成到 Agent
class AgentWithCache:
"""带缓存的 Agent"""
def __init__(self, cache: ResponseCache):
self.cache = cache
self.llm = LLMWithTokenCounting(model="gpt-4", counter=TokenCounter())
def process(self, question: str) -> str:
"""处理用户问题（带缓存）"""
# 1. 检查缓存
cached_response = self.cache.get(question)
if cached_response:
return cached_response
# 2. 调用 LLM
response = self.llm.call(question)
# 3. 缓存回答
self.cache.set(question, response)
return response
# 使用
cache = ResponseCache()
agent = AgentWithCache(cache=cache)
# 第一次调用（未命中缓存）
response1 = agent.process("今天天气怎么样？")
print(f"回答 1：{response1}")
# 第二次调用（命中缓存）
response2 = agent.process("今天天气怎么样？")
print(f"回答 2：{response2}")
优点：大幅降低成本，提高响应速度
缺点：可能返回过时信息
解决方案 5：批量调用（Batch Processing）
原理
将多个请求合并成一个批量请求，减少 API 调用次数。
from typing import List, Dict, Any
class BatchProcessor:
"""
批量处理器
适用场景：
- 多个用户输入（聊天机器人）
- 多个文档需要总结
- 多个翻译请求
"""
def __init__(self, max_batch_size: int = 10, max_wait_time: float = 5.0):
"""
Args:
max_batch_size: 最大批量大小（请求数）
max_wait_time: 最大等待时间（秒）
"""
self.max_batch_size = max_batch_size
self.max_wait_time = max_wait_time
self.batch: List[Dict] = []
def add_request(self, request: Dict) -> None:
"""添加一个请求到批量"""
self.batch.append(request)
def process_batch(self) -> List[Dict]:
"""
处理批量请求
Returns:
响应列表（与请求顺序一致）
"""
if not self.batch:
return []
# 1. 合并 Prompt
merged_prompt = self._merge_prompts(self.batch)
# 2. 调用 LLM（一次调用）
merged_response = self._call_llm(merged_prompt)
# 3. 拆分响应
responses = self._split_response(merged_response, len(self.batch))
# 4. 清空批量
self.batch = []
return responses
def _merge_prompts(self, batch: List[Dict]) -> str:
"""合并多个请求的 Prompt"""
prompts = []
for i, request in enumerate(batch):
prompts.append(f"【请求 {i+1}】{request['prompt']}")
return "\n\n".join(prompts)
def _split_response(self, merged_response: str, num_requests: int) -> List[str]:
"""拆分合并的响应"""
# 简化：假设 LLM 返回格式为：
# 【响应 1】...
# 【响应 2】...
responses = []
for i in range(num_requests):
# 用正则提取每个响应
pattern = rf"【响应 {i+1}】(.*?)(?=【响应 {i+2}】|$)"
match = re.search(pattern, merged_response, re.DOTALL)
if match:
responses.append(match.group(1).strip())
else:
responses.append("（解析失败）")
return responses
def _call_llm(self, prompt: str) -> str:
"""调用 LLM"""
# 模拟调用
return f"【响应 1】天气晴朗\n\n【响应 2】股价 123.45\n\n【响应 3】翻译完成"
# 使用示例
processor = BatchProcessor(max_batch_size=5, max_wait_time=5.0)
# 添加多个请求
processor.add_request({"prompt": "今天天气怎么样？"})
processor.add_request({"prompt": "腾讯股价是多少？"})
processor.add_request({"prompt": "翻译：Hello World"})
# 批量处理
responses = processor.process_batch()
for i, response in enumerate(responses):
print(f"响应 {i+1}：{response}")
优点：减少 API 调用次数，降低成本
缺点：增加延迟（需要等待批量满或超时）
效果对比
| 方案 | 成本降低 | 实现难度 | 性能影响 | 适用场景 |
|---|---|---|---|---|
| Token 计数和成本追踪 | N/A（监控） | ⭐⭐ | 低 | 所有场景（基础） |
| 成本预算和限流 | ⭐⭐⭐⭐ | ⭐⭐⭐ | 低 | 多用户系统 |
| 模型选择优化 | ⭐⭐⭐ | ⭐⭐ | 低 | 任务类型多样 |
| 缓存策略 | ⭐⭐⭐⭐⭐ | ⭐⭐ | 中（缓存查找） | 重复问题多 |
| 批量调用 | ⭐⭐⭐ | ⭐⭐⭐ | 高（增加延迟） | 异步任务 |
避坑指南
1. 预算不是越严越好
错误做法：
# 预算太严（用户体验差）
budget_manager = CostBudgetManager(
user_daily_budget=0.1, # 每天只能花 1 毛钱
...
)
正确做法：
# 预算合理（平衡成本和用户体验）
budget_manager = CostBudgetManager(
user_daily_budget=10.0, # 每天 ¥10
...
)
2. 缓存不是越久越好
错误做法：
# 缓存过期时间太长（返回过时信息）
cache = ResponseCache(expiration_days=365) # 缓存 1 年
正确做法：
# 缓存过期时间合理
cache = ResponseCache(expiration_days=7) # 缓存 7 天
3. 模型选择不是越便宜越好
错误做法：
# 所有任务都用最便宜的模型（质量差）
def process(task: str):
return call_gpt35(task) # 即使是复杂任务也用 GPT-3.5
正确做法：
# 根据任务难度选择模型
def process(task: str):
model = selector.select_model(task)
return call_llm(task, model=model)
延伸思考
1. 如何自动优化成本？
方案：用强化学习（RL）学习最优成本策略。
# 伪代码
class CostOptimizationRL:
"""用强化学习优化成本"""
def __init__(self):
self.policy = self._init_policy()
def optimize(self, cost_history: List[Dict]) -> Dict:
"""优化成本策略"""
# 用 RL 学习最优策略
optimal_policy = self.policy.learn(cost_history)
return optimal_policy
2. 如何预测未来成本？
方案：用时间序列预测（如 ARIMA、LSTM）。
def predict_future_cost(cost_history: List[float]) -> float:
"""预测未来 7 天的成本"""
# 用 ARIMA 模型预测
model = ARIMA(cost_history, order=(1, 1, 1))
model_fit = model.fit()
forecast = model_fit.forecast(steps=7)
return forecast
3. 如何平衡成本和质量？
方案：定义成本-质量权衡函数。
def cost_quality_tradeoff(cost: float, quality: float) -> float:
"""
成本-质量权衡函数
Returns:
得分（越高越好）
"""
# 简化：质量权重 0.7，成本权重 0.3
score = 0.7 * quality - 0.3 * cost
return score
总结
成本失控是 Agent 开发的核心痛点，需要组合多种方案：
基础方案（必做） → Token 计数和成本追踪 + 成本预算和限流
进阶方案（推荐） → 模型选择优化 + 缓存策略
高级方案（可选） → 批量调用（异步任务用）
关键原则：
- 优先用 Token 计数，精准追踪成本
- 设置合理预算，防止成本失控
- 根据任务难度选择模型，降低成本
- 缓存重复问题，减少 LLM 调用
- 批量处理异步任务，提高性价比
系列总结：
- 《Agent 开发实战：上下文窗口溢出的 5 个解决方案》
- 《Agent 开发实战：工具调用失败的 5 个解决方案》
- 《Agent 开发实战：幻觉和工具选择错误的 5 个解决方案》
- 《Agent 开发实战：多 Agent 死循环的 5 个解决方案》
- 《Agent 开发实战：成本失控的 5 个解决方案》
下一篇预告：《Agent 开发实战：延迟过高的 5 个解决方案》
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
5
0- 0
已为社区贡献31条内容
所有评论(0)