多 Agent 系统 token 浪费的根因与优化方案

每个工程团队在部署 AI Agent 时，最终都会发现一个令人不安的真相：模型本身并非最大的开销。真正的隐藏成本来自周边的一切：重复的检索、重复的 prompt、不必要的工具调用、过大 context window、多个 Agent 对同一信息进行推理。这些架构决策单独看似乎无害，但在生产规模下，它们会成为延迟、基础设施和云开销的沉重负担。

一个概念验证 Agent 每天回答 50 个问题，尚可容忍低效。但一个每分钟协调数千请求的企业平台，绝对不行。

本文探讨在不牺牲输出质量的前提下，构建 token 高效 AI 系统的实用技术。我们不会仅聚焦于 prompt 压缩，而是从路由、检索到缓存和模型选择的整个工作流进行优化。

为什么 token 优化是一个系统问题

大多数关于 token 优化的讨论始于 prompt 工程，也止于 prompt 工程。在实践中，架构决定了 token 消耗。

考虑一个典型的多 Agent 工作流：

User 
  ↓
Intent Agent
  ↓
Retriever
  ↓
Research Agent
  ↓
Planning Agent
  ↓
Writer Agent
  ↓
Reviewer Agent
  ↓
Final Response

在每个阶段，系统可能检索相同的文档、重复相同的指令、调用相同的模型，并重新发送整个对话历史。当响应到达用户时，架构已经处理了数以万计的不必要 token。

"提升效率需要重新设计工作流，而非仅仅缩短 prompt。"

架构概述

一个生产就绪的、token 高效的架构，在每次昂贵的模型调用之前都引入优化层。

User Request
       │ 
       ▼
Intent Router 
       │ 
       ▼
Semantic Cache ───────► Cached Response
       │ 
       ▼
Context Budget Manager
       │ 
       ▼
Adaptive Retriever
       │ 
       ▼
Model Router
       │ 
       ▼
LLM
       │ 
       ▼
Validated Response

"大语言模型不再是第一个组件。它是最后的、也是最昂贵的操作。"

注意这个关键转变：大语言模型不再是第一个组件。它是最后的、也是最昂贵的操作。

步骤 1：安装现代依赖

使用最新的包结构，避免已废弃的导入，并与当前 LangChain 生态对齐。

pip install \
   langchain \
   langchain-core \
   langchain-openai \
   langchain-community \
   fastapi \
   faiss-cpu \
   tiktoken \
   rank-bm25 \
   pydantic \
   python-dotenv

步骤 2：配置模型

生产系统必须通过环境配置重试、超时和凭证。

import os
from langchain_openai import ChatOpenAI 

api_key = os.getenv("OPENAI_API_KEY") 
if not api_key: 
    raise ValueError("OPENAI_API_KEY must be configured.")

llm = ChatOpenAI( 
    model="gpt-4o-mini", 
    temperature=0, 
    api_key=api_key, 
    timeout=30.0, 
    max_retries=2, 
)

设置低 temperature 提升一致性，而显式的超时和重试限制帮助系统从临时 API 故障中优雅恢复。

步骤 3：生成前先路由

并非每个请求都需要大语言 model。确定性逻辑通常可以回答简单问题。在生产系统中，将廉价请求从 LLM 路由开，是最显著的成本降低。

def classify_request(question: str) -> str:
    q = question.lower()

    if "status" in q:
        return "metrics"

    if "runbook" in q:
        return "retrieval"
   
    return "generation"

步骤 4：添加语义缓存

最简单也最有效的优化之一是精确匹配缓存，当相同问题针对相同检索文档再次被询问时，返回先前生成的响应，避免不必要的模型调用。

import hashlib

# Using an exact-match (lexical) cache
exact_match_cache = {}

def cache_key(question: str, sources: list[str]) -> str:
    """
    Generate a deterministic cache key from the user question
    and the retrieved document identifiers.
    """
    fingerprint = question + "|" + "|".join(sorted(sources))
    return hashlib.sha256(fingerprint.encode()).hexdigest()

# Example usage in the pipeline:
# key = cache_key(question, source_ids)
# if key in semantic_cache:
#     return semantic_cache[key]

步骤 5：为 context 做预算

大多数检索管道返回的文本远远超过模型实际所需。不要用每个检索到的文档填满 context window，而是建立严格的 context 预算。

import tiktoken

encoder = tiktoken.encoding_for_model("gpt-4o-mini")
MAX_CONTEXT_TOKENS = 2500

def build_context(chunks):
    context = []
    used = 0

    for chunk in chunks:
        tokens = len(encoder.encode(chunk.page_content, disallowed_special=()))

        if used + tokens > MAX_CONTEXT_TOKENS:
            break 

        context.append(chunk.page_content)
        used += tokens

    return "\n\n".join(context)

步骤 6：只检索一次

重复检索是多 Agent 系统中一个出人意料的常见缺陷。规则很简单：检索一次，到处复用。

from langchain_core.documents import Document
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

documents = [
    Document(
        page_content="Database latency often follows connection pool exhaustion.",
        metadata={"source": "db_runbook"},
    ),
    Document(
        page_content="Node pressure can increase API response times.",
        metadata={"source": "cluster_runbook"},
    ),
]

embeddings = OpenAIEmbeddings(api_key=api_key)
index = FAISS.from_documents(documents, embeddings)

retrieved_docs = index.similarity_search(question, k=4)
shared_context = build_context(retrieved_docs)

现在，每个下游 Agent 消耗相同的优化 context，而非启动各自冗余的检索管道。

步骤 7：智能路由模型

大模型应该解决复杂问题。其他一切属于更小更快的模型。

from langchain_openai import ChatOpenAI

small_model = ChatOpenAI(model="gpt-4o-mini", temperature=0, api_key=api_key)
large_model = ChatOpenAI(model="gpt-4.1", temperature=0, api_key=api_key)

def choose_model(question: str):
    """Route requests to the most appropriate model based on complexity."""
    if len(question) < 200:
        return small_model
    return large_model

此策略在不明显影响响应质量的情况下大幅降低运营成本。

步骤 8：发送前估算 token

没有 token 遥测，优化只是猜测。监控使用量使效率变得可衡量，并帮助工程师检测成本回归。

import tiktoken

encoder = tiktoken.encoding_for_model("gpt-4o-mini")

def estimate_tokens(messages):
    """
    Estimate input tokens for an OpenAI-style chat payload.
    Note: This is an estimate, not an exact billing calculation.
    """
    tokens_per_message = 3
    tokens_per_name = 1
    total = 0

    for message in messages:
        total += tokens_per_message
        for key, value in message.items():
            if isinstance(value, str):
                total += len(encoder.encode(value))
            if key == "name":
                total += tokens_per_name

    # Every reply is primed with additional assistant tokens.
    total += 3
    return total

步骤 9：验证响应

生产系统必须返回结构化输出，以确保下游系统接收可预测、格式良好的数据。

from pydantic import BaseModel

class AgentResponse(BaseModel):
    answer: str
    sources: list[str]

def validate_response(answer: str, sources: list[str]):
    """Validate and serialize the agent response using a structured schema."""
    response = AgentResponse(
        answer=answer,
        sources=sources,
    )
    return response.model_dump()

步骤 10：构建优化管道

最后，将架构组件组装成单个工作流。注意故障是如何优雅降级而非让服务崩溃。

import logging

from langchain_core.prompts import ChatPromptTemplate

logger = logging.getLogger(__name__)

def run_pipeline(question: str):
    """Execute the token-efficient AI workflow with graceful degradation."""
    try:
        route = classify_request(question)

        # Route deterministic requests away from the LLM.
        if route == "metrics":
            return {
                "answer": "Retrieve metrics directly from the monitoring system.",
                "sources": [],
            }

        # Retrieve context once.
        docs = index.similarity_search(question, k=4)

        context = build_context(docs)

        source_ids = [
            doc.metadata.get("source")
            for doc in docs
            if doc.metadata.get("source")
        ]

        # Check exact-match cache.
        key = cache_key(question, source_ids)

        if key in exact_match_cache:
            return exact_match_cache[key]

        # Select the most appropriate model.
        model = choose_model(question)

        # Keep trusted instructions separate from untrusted user input.
        prompt_template = ChatPromptTemplate.from_messages(
            [
                (
                    "system",
                    (
                        "Answer the user's question using ONLY the provided context. "
                        "If the answer cannot be determined from the context, say so."
                        "\n\nContext:\n{context}"
                    ),
                ),
                ("user", "{question}"),
            ]
        )

        chain = prompt_template | model

        result = chain.invoke(
            {
                "context": context,
                "question": question,
            }
        )

        payload = validate_response(
            answer=result.content,
            sources=source_ids,
        )

        # Cache validated response.
        exact_match_cache[key] = payload

        return payload

    except Exception:
        logger.exception("Token-efficient pipeline failed.")

        # Gracefully degrade instead of crashing.
        return {
            "answer": (
                "The AI pipeline encountered an error. "
                "Please continue using the standard operational workflow."
            ),
            "sources": [],
        }

什么真正降低了 token 使用量？

当团队为此类架构配备工具时，最大的节省很少来自编辑 prompt。它们来自消除不必要的工作。

最大的改进通常源于：

• 文档只检索一次，而非多次
• 缓存语义相同的请求
• 将简单请求从 LLM 路由开
• 用明确的 token 预算限制 context
• 选择最小适用模型

这些架构转变在降低成本和延迟的同时，也让系统行为更容易推理。

核心原则

在优化 AI 系统投入生产时，几个核心原则始终浮现：

像对待基础设施一样对待 token： Token 是有限的资源，就像 CPU 周期或内存一样。监控它们、为它们做预算、优化它们。

检索通常是最大的浪费来源： 重复检索通常比冗长 prompt 贡献更多不必要的 token。尽可能共享 context。

更大的模型并不总是更好： 更小更快的模型可以有效处理许多操作任务。把更大的模型留给真正复杂的推理。

缓存是一项工程特性： 语义缓存不仅仅是一种性能优化——它是减少成本、延迟和提供商依赖的核心架构组件。

优化前先测量： 仪表化必须伴随每个生产部署。

结论

随着 AI 系统日趋成熟，成功将越来越多地取决于工程效率，而非原始模型规模。AI Agent 的隐藏税很少是单个昂贵的 prompt；而是跨分布式工作流的重复检索、过大的 context、不必要的模型调用和重复推理的积累。

"最高效的生产 AI 系统不是生成最多 token 的系统，而是只生成真正所需的 token 的系统。"

通过将 token 消耗视为系统工程问题，组织可以构建更快、更便宜、高度可扩展的 AI 平台。智能路由请求、预算 context、共享检索结果、验证结构化输出和引入语义缓存，都是保证效率而不影响质量的实用技术。

最高效的生产 AI 系统不是生成最多 token 的系统，而是只生成真正所需的 token 的系统。