5.2 性能优化与成本控制

5.2 性能优化与成本控制 — LangChain 框架精通省钱提速指南

本节导读：学完本节，你将掌握 LangChain Agent 在生产环境中的性能优化和成本控制方法，包括模型选择策略、Prompt 压缩、缓存机制、并发控制和 Token 用量监控，把每次请求的成本控制在可预期范围内。

学习目标

• 理解 LLM 应用的成本构成和关键优化维度
• 掌握 Prompt 优化、缓存策略和模型路由三大成本控制手段
• 学会用流式输出和并发请求提升用户感知性能
• 建立基于数据的成本监控体系

核心概念

LLM 应用的成本结构与传统软件有本质区别：运行成本几乎全部由 Token 消耗决定。一次 Agent 调用可能涉及 3-10 轮 LLM 请求，每轮都带着完整的对话历史。如果不加控制，一个看起来简单的"帮我查一下明天天气"请求，背后可能消耗掉 5000-20000 个 Token。

成本优化的核心思路不是"少用"——而是"用对"。具体来说，有三个维度：

模型选择是最粗暴也最有效的手段：gpt-4o-mini 的价格大约是 gpt-4o 的 1/10，对于 80% 的场景质量足够。输入压缩关注的是每轮传入模型的 Token 数——对话历史冗余、工具描述过长、检索结果不精都是浪费。调用控制则是从架构层面减少 LLM 调用次数——能缓存的缓存、能并行的并行、能预判的预判。

这三者不是独立的，而是一个决策链：先选对模型，再压输入，最后从架构上减少调用。

环境准备 / 前置知识

• 已掌握 create_agent 的基本用法（本教程 4.1 节）
• 了解 Token 的基本概念和计费方式
• Python 3.10+ 环境

分步实战

步骤 1：量化你的成本——先有数据再优化

在优化之前，你必须先知道钱花在哪里。以下是一个简单的成本追踪器，每次 Agent 调用后记录 Token 消耗：

import time
from dataclasses import dataclass, field

@dataclass
class CostTracker:
    """追踪 Agent 调用的 Token 消耗和成本"""
    total_input_tokens: int = 0
    total_output_tokens: int = 0
    call_count: int = 0
    costs: list[float] = field(default_factory=list)

    # 模型价格表（每百万 Token，美元）
    PRICING = {
        "gpt-4o": (2.5, 10.0),       # (input, output)
        "gpt-4o-mini": (0.15, 0.6),
        "claude-sonnet-4-6": (3.0, 15.0),
        "claude-haiku-4": (0.8, 4.0),
        "gemini-2.5-flash-lite": (0.075, 0.3),
    }

    def record(self, model: str, input_tokens: int, output_tokens: int):
        self.call_count += 1
        self.total_input_tokens += input_tokens
        self.total_output_tokens += output_tokens
        pricing = self.PRICING.get(model, (1.0, 3.0))  # 默认价格
        cost = (input_tokens * pricing[0] + output_tokens * pricing[1]) / 1_000_000
        self.costs.append(cost)

    def summary(self) -> str:
        total_cost = sum(self.costs)
        avg_cost = total_cost / self.call_count if self.call_count else 0
        return (
            f"调用次数: {self.call_count}\n"
            f"输入 Token: {self.total_input_tokens:,}\n"
            f"输出 Token: {self.total_output_tokens:,}\n"
            f"总成本: ${total_cost:.4f}\n"
            f"平均每次: ${avg_cost:.4f}"
        )

tracker = CostTracker()
​

配合 LangSmith（本教程 5.1 节），你可以得到更精确的按模型、按工具分类的成本数据。但上面这段代码作为轻量级本地追踪已经足够用来发现问题。

步骤 2：Prompt 压缩——减少无效 Token

Agent 应用中 Token 浪费最严重的几个地方：

对话历史占了最大头（约 40%）。Agent 每轮推理都把之前所有对话传给模型，5 轮之后历史就可能有 3000-5000 Token。解决方案是限制历史长度：

def trim_messages(messages: list, max_tokens: int = 4000) -> list:
    """保留最近的消息，控制总 Token 数"""
    # 粗略估算：1 个中文字约 1.5 Token，1 个英文词约 1.3 Token
    # 这里用字符数的 0.7 倍作为近似 Token 数
    trimmed = []
    total_chars = 0
    for msg in reversed(messages):
        msg_chars = len(msg.get("content", ""))
        if total_chars + msg_chars > max_tokens * 1.4:  # 字符/token 比率
            break
        trimmed.insert(0, msg)
        total_chars += msg_chars
    return trimmed
​

工具描述是另一个常见浪费点。工具的 docstring 就是模型看到的工具说明，写得越详细模型理解越好，但 Token 消耗也越大。我的建议是：

# 坏例子：工具描述过于冗长
def search_database(query: str) -> str:
    """
    这个函数会在我们的 PostgreSQL 数据库中执行全文搜索。
    数据库名叫 production_db，表名叫 documents，包含 id、title、content、
    created_at 四个字段。搜索使用的是 tsvector 类型的全文索引，
    支持中英文混合搜索，会返回前 10 条按相关度排序的结果……
    """
    # 实际搜索逻辑
    pass

# 好例子：精炼但信息完整
def search_database(query: str) -> str:
    """搜索内部文档库，返回最相关的 10 条结果。支持中英文查询。"""
    # 实际搜索逻辑
    pass
​

模型不需要知道你的数据库叫什么、表结构是什么。它只需要知道：这个工具干什么、输入是什么、输出大概长什么样。

步骤 3：缓存策略——避免重复计算

LLM 的输出在相同输入下是确定性的（temperature=0 时）。这意味着对于完全相同的输入，你可以直接返回缓存结果而不调用模型。

import hashlib
import json
from functools import lru_cache

# 简单的内存缓存
_agent_cache: dict[str, dict] = {}

def cached_agent_invoke(agent, user_message: str, cache_ttl: int = 3600) -> dict:
    """带缓存的 Agent 调用"""
    cache_key = hashlib.md5(user_message.encode()).hexdigest()

    if cache_key in _agent_cache:
        cached = _agent_cache[cache_key]
        if time.time() - cached["timestamp"] < cache_ttl:
            print(f"[CACHE HIT] {user_message[:50]}...")
            return cached["result"]

    # 缓存未命中，执行真实调用
    result = agent.invoke(
        {"messages": [{"role": "user", "content": user_message}]}
    )

    _agent_cache[cache_key] = {
        "result": result,
        "timestamp": time.time()
    }
    return result
​

对于更复杂的场景，建议使用 Redis 作为缓存后端。缓存策略的关键决策点：

工具结果缓存是投入产出比最高的优化。Agent 可能会反复调用同一个工具（比如多次搜索同个关键词），工具结果缓存可以直接砍掉这些重复的 LLM 调用。

步骤 4：模型路由——不同任务用不同模型

不是每个步骤都需要最强的模型。典型的路由策略：

在 LangChain 中实现模型路由：

from langchain.agents import create_agent

# 轻量级 Agent 用于简单任务
simple_agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[simple_search, get_time],
    system_prompt="回答简单的事实性问题，保持简洁。",
)

# 高级 Agent 用于复杂任务
advanced_agent = create_agent(
    model="openai:gpt-4o",
    tools=[deep_analysis, code_execute, web_search],
    system_prompt="你是一个高级分析助手，可以执行复杂推理和代码。",
)

def route_request(user_message: str) -> dict:
    """根据消息复杂度路由到不同 Agent"""
    # 简单规则路由（生产环境可以用小模型做意图分类）
    simple_keywords = ["几点", "天气", "计算", "翻译"]
    is_simple = any(kw in user_message for kw in simple_keywords)

    agent = simple_agent if is_simple else advanced_agent
    return agent.invoke(
        {"messages": [{"role": "user", "content": user_message}]}
    )
​

价格对比（以 1000 次调用、每次平均 2000 输入 + 500 输出 Token 估算）：

我强烈建议：先用 gpt-4o-mini 开发和调试，确认流程跑通后再评估是否需要升级模型。90% 的场景 mini 的质量够用，而且成本差一个数量级。

步骤 5：流式输出——提升用户感知性能

用户等待 5 秒看到完整回答，和 0.5 秒开始看到第一个字，体验差异巨大。流式输出不减少总 Token 消耗，但大幅改善用户体感。

# create_agent 支持 astream 流式调用
async def stream_response(agent, user_message: str):
    """流式输出 Agent 响应"""
    async for event in agent.astream_events(
        {"messages": [{"role": "user", "content": user_message}]},
        version="v2"
    ):
        kind = event.get("event")
        if kind == "on_chat_model_stream":
            # 逐 token 输出
            chunk = event["data"]["chunk"]
            if hasattr(chunk, "content") and chunk.content:
                print(chunk.content, end="", flush=True)
        elif kind == "on_tool_start":
            print(f"\n[调用工具: {event['name']}...]", flush=True)

# 运行
import asyncio
asyncio.run(stream_response(agent, "解释一下量子计算的基本原理"))
​

流式输出的另一个好处是：用户可以在看到部分结果后提前中断（比如"这不是我要的"），节省后续 Token。

完整示例：带成本控制的 Agent 包装器

把上面的优化策略整合起来，写一个生产级的 Agent 包装器：

import time
import hashlib
from dataclasses import dataclass, field

@dataclass
class CostAwareAgent:
    """带成本追踪和缓存的 Agent 包装器"""

    def __init__(self, agent, model_name: str = "gpt-4o-mini", cache_ttl: int = 1800):
        self.agent = agent
        self.model_name = model_name
        self.cache_ttl = cache_ttl
        self._cache: dict[str, tuple[dict, float]] = {}
        self.stats = {
            "total_calls": 0,
            "cache_hits": 0,
            "total_input_tokens": 0,
            "total_output_tokens": 0,
        }

    def invoke(self, user_message: str, use_cache: bool = True) -> dict:
        self.stats["total_calls"] += 1
        cache_key = hashlib.md5(user_message.encode()).hexdigest()

        # 缓存检查
        if use_cache and cache_key in self._cache:
            result, timestamp = self._cache[cache_key]
            if time.time() - timestamp < self.cache_ttl:
                self.stats["cache_hits"] += 1
                print(f"[缓存命中] 节省了一次 LLM 调用")
                return result

        # 实际调用
        start = time.time()
        result = self.agent.invoke(
            {"messages": [{"role": "user", "content": user_message}]}
        )
        elapsed = time.time() - start

        # 估算 Token（粗略：中文字符数 * 1.5）
        input_text = user_message
        output_text = result["messages"][-1].content if result["messages"] else ""
        self.stats["total_input_tokens"] += int(len(input_text) * 1.5)
        self.stats["total_output_tokens"] += int(len(output_text) * 1.5)

        print(f"[调用完成] 耗时 {elapsed:.2f}s")

        # 写入缓存
        if use_cache:
            self._cache[cache_key] = (result, time.time())

        return result

    def report(self) -> str:
        hit_rate = (
            self.stats["cache_hits"] / self.stats["total_calls"] * 100
            if self.stats["total_calls"] else 0
        )
        return (
            f"=== 成本报告 ===\n"
            f"总调用: {self.stats['total_calls']}\n"
            f"缓存命中: {self.stats['cache_hits']} ({hit_rate:.1f}%)\n"
            f"估算输入 Token: {self.stats['total_input_tokens']:,}\n"
            f"估算输出 Token: {self.stats['total_output_tokens']:,}"
        )

# 使用
from langchain.agents import create_agent

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[search_docs],
    system_prompt="你是一个技术助手。",
)
cost_agent = CostAwareAgent(agent, model_name="gpt-4o-mini")

# 第一次调用（缓存未命中）
cost_agent.invoke("什么是 LangChain？")
# 第二次相同调用（缓存命中）
cost_agent.invoke("什么是 LangChain？")

print(cost_agent.report())
​

常见问题 FAQ

Q1：Agent 多轮循环导致 Token 暴增怎么办？

A：这是 Agent 应用最常见的成本问题。每轮循环都传完整对话历史，10 轮之后可能就是上万 Token。解决思路有三个：（1）限制最大循环次数（create_agent 的 max_iterations 参数）；（2）使用本节提到的 trim_messages 压缩历史；（3）让 Agent 更快做出决策——优化工具描述和系统提示，减少不必要的工具调用。

Q2：gpt-4o-mini 和 gpt-4o 质量差距有多大？

A：对于绝大多数 Agent 场景（信息检索、简单推理、代码解释、翻译），差距在 5%-10% 以内。但对于复杂多步推理、数学证明、创意写作，gpt-4o 明显更强。我的建议是：先用 mini，只在 mini 明显不够时才升级。

Q3：如何估算月度 LLM 成本？

A：用日均调用量 × 平均每次 Token 消耗 × 单价。例如：日均 500 次调用，每次平均 3000 输入 + 800 输出 Token，用 gpt-4o-mini，月成本约为 500 × 30 × (3000×0.15 + 800×0.6) / 1,000,000 = $10.35。建议在上线前用 CostTracker 跑一周真实数据，不要凭估算。

Q4：缓存会导致回答过时吗？怎么处理？

A：会。设合理的 TTL（我建议 30 分钟到 2 小时）可以平衡新鲜度和成本。对于时效性强的数据（如实时股价、天气），不要缓存或用极短的 TTL（1-5 分钟）。对于知识库类查询，可以设较长 TTL，因为文档本身不会频繁变化。

Q5：LangChain 有内置的成本追踪功能吗？

A：LangChain 的回调系统（Callbacks）可以在每次 LLM 调用后获取 Token 用量。配合 LangSmith（本教程 5.1 节），你可以得到完整的按项目、按模型的成本 Dashboard。比手写 CostTracker 更准确，因为能拿到模型返回的真实 Token 数。

最佳实践与避坑

• 先量化再优化：没有数据之前不要猜瓶颈在哪，用 CostTracker 或 LangSmith 跑一周基线
• 模型选择是最大的杠杆：gpt-4o-mini 替代 gpt-4o 可以直接降低 90% 的模型成本
• 工具描述要精炼：每个工具的 docstring 控制在 50 字以内，信息完整但不冗余
• 缓存工具结果比缓存 LLM 输出更划算：工具调用的结果通常更确定、更值得缓存
• 设 max_iterations 上限：防止 Agent 陷入死循环，同时也控制了最坏情况的成本
• 流式输出是低成本高收益的优化：不省 Token 但极大改善用户体验
• 坑点：temperature=0 不代表完全确定：同一输入多次调用仍可能有微小差异，缓存 key 要考虑这点
• 坑点：忽略系统提示的 Token 消耗：很多人的系统提示有 2000+ Token 但完全没意识到

本节小结

本节从成本构成分析入手，讲解了四个核心优化手段：Prompt 压缩减少无效 Token、缓存策略避免重复计算、模型路由让不同任务用不同价格的模型、流式输出提升用户感知性能。最后给出了一个带成本追踪和缓存的 Agent 包装器。

成本优化的原则是"先有数据，再有行动"。不要上来就各种优化，先用 CostTracker 或 LangSmith 跑一周，看看钱到底花在哪，再针对性优化。下一节我们将讨论部署最佳实践，把优化好的 Agent 安全稳定地跑在生产环境。

延伸阅读

• 本教程 5.1 节 LangSmith 可观测性与调试（Token 监控的数据基础）
• 本教程 4.1 节 Agent 基础与 create_agent
• 本教程 4.2 节 工具调用进阶与中间件
• LangChain 官方文档回调系统（Callbacks and Token Tracking）

关键词：LangChain 性能优化, LLM 成本控制, Token 优化, Agent 缓存, 模型路由, Prompt 压缩, 流式输出, LangChain 框架精通
难度：进阶
预计阅读：15 分钟