阅读时间：约 16 分钟
前置知识：理解 Agent 决策循环（P01）、工具调用（P02）、重试策略（P04）、工作流（P08）

---

P01-P08 把 Agent 搭出来了。但上线第一个礼拜，账单来了。一个 Agent 跑一天花了 40 块钱。正常吗？正常。怎么让同样的 Agent 跑一天只花 8 块？这才是工程。

---

前言：Agent 的成本长什么样

一个 Agent 跑一次，不是"调用一次 LLM 多少钱"。它是一个决策循环：

用户输入 → [LLM 判断] → [可能调工具1] → [可能调工具2] → [可能重试] → [最终回复]

把每一步的 Token 加起来：

步骤	输入 Token	输出 Token	说明
用户消息	100	0	用户说了什么
System Prompt	2000	0	固定开销，每次都带
上下文历史	1500	0	P03 的上下文管理
LLM 推理	3600	800	第一次决策
工具调用	800	400	调工具的描述 + 返回
再次推理	4800	600	基于工具结果再判断
最终回复	5400	200	给用户的答案
合计	16300	2000

一次运行 18300 Token。跑 1000 次就是 1830 万 Token。按当前主流 API 价格算，大约 3-10 块钱。

1000 次运行看起来不多。一个中型网站的日活用户，随便跑几次就是这么多。

📌 本章核心：Agent 成本 = 固定开销（System Prompt + 工具描述）+ 可变开销（上下文 + 推理 + 工具调用 + 重试）。优化每一层，整体降 60-80%。

---

第一部分：固定开销：每次都付的钱，可以省

System Prompt 和工具描述，每次运行都要带。2000 Token 的 Prompt，跑 10 万次就是 2 亿 Token。

1.1 System Prompt 压缩

P01 的 System Prompt 大概长这样：

system_prompt = """
你是技术问答助手。你叫 TechBot。
你擅长 Python、JavaScript、Go 等技术栈。
你有以下工具可用：
- get_weather: 查询天气
- search_code: 搜索代码
- book_flight: 预订航班
...（还有 50 个工具的描述）
"""

问题：用户问"今天天气怎么样"，你却把 50 个工具的描述全带上了。

优化方案：按上下文动态选择工具描述。

class ToolDescriptionFilter:
    """按需加载工具描述，只带可能用到的"""

    def __init__(self, tools):
        self.tools = tools  # [{name, description, keywords}, ...]

    def filter_for_task(self, user_message, n=3):
        """
        根据用户消息，选择最相关的 N 个工具描述
        简单实现用关键词匹配，生产环境用向量检索
        """
        msg_lower = user_message.lower()
        scored = []
        for tool in self.tools:
            score = 0
            for kw in tool.get("keywords", []):
                if kw.lower() in msg_lower:
                    score += 1
            if score > 0:
                scored.append((score, tool))

        scored.sort(key=lambda x: x[0], reverse=True)
        selected = [t[1] for t in scored[:n]]

        # 始终带上系统描述
        tools_desc = "\n\n".join(
            f"- {t['name']}: {t['description']}" for t in selected
        )
        return f"""你是技术问答助手。
可用工具：
{tools_desc}

如果工具描述不够，再查看完整工具列表。"""

效果：System Prompt 从 2000 Token 降到 500 Token，固定开销降 75%。

1.2 工具描述按场景分组

不只是按当前任务选，还要按用户类型、业务场景预分组。

场景	常用工具	工具描述 Token
日常聊天	无	0
天气查询	get_weather	150
代码问题	search_code, analyze_code	400
旅行规划	get_weather, search_flights, book_flight	700

📌 本章要点：System Prompt 压缩 + 工具描述按需加载，固定开销从 2000 降到 500 Token。不用的工具描述，一次都不要带。

---

固定开销管住了。现在看更大的那块：上下文历史。

第二部分：上下文管理：越聊越贵的秘密

P03 讲过上下文管理，但当时侧重"怎么管得全"。现在说"怎么管得省"。

2.1 上下文压缩

对话超过 10 轮，历史消息占 8000+ Token。怎么压缩？

class ContextCompressor:
    """上下文压缩：把历史对话压缩为摘要"""

    def __init__(self, llm_client, max_tokens=4000):
        self.llm = llm_client
        self.max_tokens = max_tokens

    def compress(self, messages):
        """
        保留最近 3 轮对话，其余压缩为摘要
        效果：20 轮对话的上下文从 12000 Token 降到 4000 Token
        """
        if len(messages) <= 6:  # 3 轮对话以内，不压缩
            return messages

        recent = messages[-6:]  # 最近 3 轮
        older = messages[:-6]

        # 调用 LLM 压缩
        summary_prompt = f"""
        将以下对话压缩为一句话摘要，保留关键信息：
        {"".join(str(m) for m in older)}

        用以下格式返回：
        [对话摘要] 用户询问了天气和代码问题，助手提供了北京天气信息和 Python 调试建议。
        """
        summary = self.llm.generate(summary_prompt, model="deepseek-v4-flash").content

        return [
            {"role": "system", "content": summary},
            *recent
        ]

2.2 滑动窗口 + 摘要混合

P03 讲过滑动窗口。这里补充一个关键数据：

策略	上下文 Token	信息保留率	适用场景
全量保留	12000+	100%	短对话（<10 轮）
滑动窗口（最近 4 轮）	3000-5000	60%	中等对话
滑动窗口 + 摘要	4000-6000	80%	长对话（推荐）

最佳实践：短对话全量，中等对话滑动窗口，长对话滑动窗口 + 摘要。

2.3 工具调用结果缓存

工具调用是最贵的部分之一：调用外部 API + LLM 推理。同一个工具调用，100 次有 80 次结果一样。

import time
import hashlib
import json

class ToolCache:
    """工具调用缓存：同一个调用不重复执行"""

    def __init__(self, ttl=300):  # 默认 5 分钟过期
        self.cache = {}
        self.ttl = ttl

    def _key(self, tool_name, args):
        """用工具名 + 参数生成缓存键"""
        raw = f"{tool_name}:{json.dumps(args, sort_keys=True)}"
        return hashlib.md5(raw.encode()).hexdigest()

    def get(self, tool_name, args):
        key = self._key(tool_name, args)
        entry = self.cache.get(key)
        if entry and time.time() - entry["time"] < self.ttl:
            return entry["result"]
        return None

    def set(self, tool_name, args, result):
        key = self._key(tool_name, args)
        self.cache[key] = {
            "result": result,
            "time": time.time()
        }

    def clear(self, tool_name=None):
        if tool_name:
            self.cache = {
                k: v for k, v in self.cache.items()
                if not k.startswith(tool_name)
            }
        else:
            self.cache.clear()

缓存覆盖策略：

工具类型	缓存策略	TTL	预期命中率
天气查询	按城市缓存	30 分钟	40-60%
搜索 API	按查询词缓存	5 分钟	20-40%
实时数据	不缓存	0	0%
用户偏好	持久化缓存	永久	60-80%

📌 本章要点：上下文压缩（短对话全量、长对话摘要）、滑动窗口 + 摘要混合、工具调用缓存。上下文优化是单次运行成本降 30-50% 的关键。

---

上下文管住了。接下来是最大的成本块：LLM 推理。

第三部分：模型路由：简单任务别用贵的模型

3.1 复杂度分级

不是每个问题都需要最强的模型。

class SmartRouter:
    """智能模型路由：按任务复杂度选模型"""

    def __init__(self):
        # 模型配置：模型名 + 能力等级 + 价格（每 1K Token）
        self.models = {
            "tiny": {
                "name": "deepseek-v4-flash",
                "cost_per_1k": 0.001,  # 最便宜
                "max_tokens": 4000,
                "capabilities": ["simple_classification", "intent_detection"]
            },
            "standard": {
                "name": "qwen3-max",
                "cost_per_1k": 0.008,
                "max_tokens": 8000,
                "capabilities": ["reasoning", "tool_calling", "summarization"]
            },
            "powerful": {
                "name": "deepseek-v4-pro",
                "cost_per_1k": 0.032,
                "max_tokens": 20000,
                "capabilities": ["complex_reasoning", "multi_tool", "creativity"]
            }
        }

    def classify_complexity(self, user_message):
        """
        用最小成本判断任务复杂度
        这是一个"判断派给谁"的任务，本身用便宜模型就能做
        """
        system_prompt = """
        判断用户任务的复杂度，返回 1-5 的整数：
        1 = 简单问答/打招呼
        2 = 单步操作（查天气、问事实）
        3 = 需要 1-2 步工具调用
        4 = 需要 3+ 步工具调用或推理
        5 = 多步骤复杂任务（分析、对比、创作）
        
        只返回数字，不要解释。
        用户输入：{msg}
        """
        resp = call_llm(system_prompt.format(msg=user_message),
                        model="deepseek-v4-flash")
        return int(resp.content.strip())

    def route(self, user_message, complexity=None):
        """
        根据复杂度路由到合适模型
        复杂度 1-2 → 便宜模型
        复杂度 3-4 → 标准模型
        复杂度 5 → 最强模型
        """
        if complexity is None:
            complexity = self.classify_complexity(user_message)

        if complexity <= 2:
            return self.models["tiny"]
        elif complexity <= 4:
            return self.models["standard"]
        else:
            return self.models["powerful"]

3.2 路由的效果

用户说："今天天气怎么样？"
→ 复杂度 1 → deepseek-v4-flash（0.001/1K Token）
→ 成本：约 0.0003 元

用户说："帮我分析一下这个架构方案的优缺点"
→ 复杂度 4 → qwen3-max（0.008/1K Token）
→ 成本：约 0.005 元

用户说："对比 LangGraph、CrewAI、AutoGen，写一份选型报告"
→ 复杂度 5 → deepseek-v4-pro（0.032/1K Token）
→ 成本：约 0.02 元

实际数据：如果业务场景中 60% 的查询复杂度在 1-2 级，路由后总成本降低 50-70%。

3.3 降级策略：Token 预算耗尽怎么办

跑着跑着发现 Token 快超预算了，不能直接报错，要优雅降级。

class CostGuard:
    """成本守卫：Token 预算控制 + 优雅降级"""

    def __init__(self, budget_per_run=10000, max_runs=1000):
        self.budget_per_run = budget_per_run
        self.max_runs = max_runs
        self.current_budget = budget_per_run
        self.run_count = 0

    def check_budget(self, estimated_tokens):
        """执行前检查预算"""
        if self.run_count >= self.max_runs:
            return False, "今日运行次数已达上限"
        if estimated_tokens > self.current_budget:
            return False, "预算不足"
        return True, "OK"

    def estimate_tokens(self, messages, tool_defs=None):
        """估算总 Token 消耗"""
        input_tokens = sum(len(str(m.get("content", ""))) // 2
                          for m in messages)
        if tool_defs:
            input_tokens += sum(len(str(t)) for t in tool_defs)

        # 预估输出：推理约 500 Token，重试 2 次
        output_tokens = 500 * (1 + 0.3 * 2)  # 30% 概率重试

        return input_tokens + output_tokens

    def degrade(self, task, available_budget):
        """
        预算不足时的降级策略
        优先级：简单回复 > 工具调用 > 完整推理
        """
        if available_budget < 2000:
            # 最低预算：给一个简短回复
            return {
                "mode": "simple_reply",
                "max_tokens": 200,
                "model": "deepseek-v4-flash"
            }
        elif available_budget < 5000:
            # 中等预算：只做工具调用，不做深度推理
            return {
                "mode": "tool_only",
                "max_tokens": 500,
                "model": "deepseek-v4-flash"
            }
        else:
            # 正常预算：完整推理
            return {
                "mode": "full_reasoning",
                "max_tokens": available_budget,
                "model": "qwen3-max"
            }

📌 本章要点：模型路由（按复杂度选模型）降 50-70% 成本。降级策略（简单回复 → 工具调用 → 完整推理）确保预算耗尽时不直接崩溃。

---

路由和降级都做了。但还有最后一个成本杀手：重试。

第四部分：重试策略：越重试越贵

P04 讲过重试策略，但当时侧重"怎么不崩"。现在说"怎么不花冤枉钱"。

4.1 重试预算

不是每个错误都值得重试。

class RetryBudget:
    """
    重试预算：每个 Agent 运行限制总重试次数
    防止无限重试烧 token
    """

    def __init__(self, max_retries=3, budget_per_retry=2000):
        self.max_retries = max_retries
        self.budget_per_retry = budget_per_retry

    def should_retry(self, error_type, retry_count, estimated_cost):
        """
        判断是否应该重试
        - 工具超时：值得重试（3 次内）
        - 工具返回空：不值得重试（再试也是空）
        - 模型幻觉：不值得重试（幻觉是概率性的，再试一样）
        - 网络错误：值得重试（5 次内）
        """
        if retry_count >= self.max_retries:
            return False

        retry_worthiness = {
            "timeout": True,
            "empty_result": False,   # 空结果重试无意义
            "hallucination": False,  # 幻觉再试大概率还是错
            "network_error": True,
            "invalid_params": False, # 参数错了重试也没用
            "rate_limit": True,
        }

        return retry_worthiness.get(error_type, False)

    def estimate_retry_cost(self, base_tokens):
        """估算重试的总成本（指数退避重试，每次 token 量可能增加）"""
        total = 0
        for i in range(self.max_retries):
            # 每次重试上下文变长（多了错误信息）
            tokens = base_tokens + (i * 500)
            total += tokens
        return total

4.2 熔断机制

当连续失败超过阈值，直接熔断：不再调用 LLM，返回缓存结果或友好提示。

class CircuitBreaker:
    """熔断器：连续失败 N 次后停止调用"""

    def __init__(self, failure_threshold=5, reset_timeout=60):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "closed"  # closed | open | half_open

    def record_result(self, success):
        if success:
            self.failure_count = 0
            self.state = "closed"
        else:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = "open"

    def allow_request(self):
        if self.state == "closed":
            return True
        if self.state == "open":
            if (time.time() - self.last_failure_time
                    > self.reset_timeout):
                self.state = "half_open"
                return True
            return False
        # half_open: 允许一次请求
        return True

效果：服务异常时，熔断在 5 次失败后停止调用 LLM，避免每次重试都在烧钱。

📌 本章要点：重试预算（哪些错误值得重试）+ 熔断（连续失败后停止调用）。重试不是免费的，每次重试都在烧 Token。

---

第五部分：成本监控：看不见就管不好

class TokenCostEstimator:
    """Token 成本估算器：每次运行都记账"""

    def __init__(self):
        self.costs = []
        self.price_map = {
            "deepseek-v4-flash": {"input": 0.001, "output": 0.002},
            "qwen3-max": {"input": 0.008, "output": 0.016},
            "deepseek-v4-pro": {"input": 0.032, "output": 0.064},
        }

    def estimate(self, model_name, input_tokens, output_tokens):
        """估算成本"""
        prices = self.price_map.get(model_name, {
            "input": 0.01, "output": 0.02
        })
        cost = (input_tokens / 1000) * prices["input"] + \
               (output_tokens / 1000) * prices["output"]

        record = {
            "model": model_name,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cost": round(cost, 6),
            "timestamp": time.time()
        }
        self.costs.append(record)
        return record

    def daily_report(self):
        """生成日成本报告"""
        if not self.costs:
            return {}

        total_cost = sum(r["cost"] for r in self.costs)
        model_breakdown = {}
        for r in self.costs:
            model = r["model"]
            if model not in model_breakdown:
                model_breakdown[model] = {"runs": 0, "cost": 0}
            model_breakdown[model]["runs"] += 1
            model_breakdown[model]["cost"] += r["cost"]

        return {
            "total_cost": round(total_cost, 2),
            "total_runs": len(self.costs),
            "avg_cost_per_run": round(total_cost / len(self.costs), 6),
            "model_breakdown": model_breakdown
        }

    def find_hotspots(self, top_n=5):
        """找出最烧钱的运行记录"""
        sorted_runs = sorted(self.costs, key=lambda r: r["cost"], reverse=True)
        return sorted_runs[:top_n]

关键监控指标

指标	正常范围	需要关注
单次运行成本	0.001-0.01 元	> 0.05 元
日总成本	根据预算	超预算
重试率	< 10%	> 20%
缓存命中率	> 40%	< 10%
最贵 5% 的运行	-	占总量 50%+

📌 本章要点：看不见的成本就是失控的成本。每次运行都记账，每天出报告，每周看趋势。

---

第六部分：优化效果总览

把前面所有的优化串起来：

原始状态：
├── System Prompt: 2000 Token (固定)
├── 上下文历史: 4000 Token
├── LLM 推理: 6000 Token
├── 工具调用: 4000 Token
├── 重试 (平均 1 次): 4000 Token
└── 单次总计: ~20000 Token

优化后：
├── System Prompt: 500 Token (按需加载，-75%)
├── 上下文历史: 2000 Token (压缩 + 缓存，-50%)
├── LLM 推理: 2000 Token (模型路由，-67%)
├── 工具调用: 1500 Token (缓存，-62.5%)
├── 重试: 1000 Token (预算 + 熔断，-75%)
└── 单次总计: ~7000 Token

总成本降低：20000 → 7000，约 65%。

如果配合：

• 高缓存命中率（60%+）→ 再降 20%
• 路由后 60% 任务走便宜模型 → 再降 30%

最终可能降到原成本的 15-20%。

📌 本章要点：每层优化 30-70%，组合起来 65%。加缓存 + 路由，可能降到原成本的 15-20%。

---

总结：读完这篇，你应该带走这几件事

1. Agent 成本 = 固定开销 + 可变开销 + 重试开销。固定开销每次都不变（System Prompt），可变开销随对话增长（上下文 + 推理），重试开销是失败时产生的额外成本。
2. 固定开销靠压缩和按需加载。System Prompt 从 2000 降到 500 Token，工具描述按需加载。
3. 上下文靠压缩、缓存、滑动窗口。短对话全量、长对话摘要，工具结果缓存命中率 40-80%。
4. 推理靠模型路由。按复杂度选模型，60% 的任务走便宜模型，成本降 50-70%。
5. 重试靠预算和熔断。不是每个错误都值得重试，连续失败要熔断。
6. 成本必须可观测。每次运行记账，每天出报告，找不到最烧钱的环节就没法优化。

🤔 思考一下：你的 Agent 跑一次大概花多少 Token？有没有算过？

---

思维导图

• Agent 成本优化
  • 成本结构
    • 固定开销 20-30%（System Prompt + 工具描述）
    • 可变开销 50-60%（上下文 + 推理 + 工具调用）
    • 重试开销 10-20%（失败重试）
  • 固定开销优化
    • System Prompt 压缩（2000 → 500）
    • 工具描述按需加载
  • 上下文优化
    • 压缩 + 滑动窗口 + 摘要
    • 工具调用缓存（命中率 40-80%）
  • 推理优化
    • 模型路由（按复杂度选模型）
    • 降级策略（简单回复 → 工具调用 → 完整推理）
  • 重试优化
    • 重试预算（哪些错误值得重试）
    • 熔断（连续失败后停止）
  • 成本监控
    • 每次运行记账
    • 日报告 + 周趋势
    • 最烧钱环节定位
  • 优化效果
    • 单层优化 30-70%
    • 组合优化 65%
    • 配合缓存 + 路由降到 15-20%

---

下一篇预告

S02. 监控与可观测：从"不知道发生了什么"到"全链路追踪"

成本优化了，但怎么知道 Agent 在跑什么？出了错怎么定位？用户投诉说"这 Agent 越来越智障了"，你怎么证明是模型的问题还是你代码的问题？全链路追踪 + 指标 Dashboard + 异常告警。

🤔 思考一下：你的 Agent 现在出了问题，你是"猜"哪里错了，还是有日志能看到每一步发生了什么？