别等上线被骂才想起优化，这三个维度决定你Agent的生死

一、先建一张「性能地图」：搞清楚钱和时间花在哪里

在动刀之前，得先知道哪里是瓶颈。一个典型的LangGraph Agent调用链路，耗时分布大概长这样：

阶段	典型耗时	占比
节点调度	~5ms	<1%
工具描述注入	50-300ms	~3%
LLM推理	500ms-3s	~60%
工具执行	200ms-2s	~30%
State读写	~5ms	<1%

LLM推理 + 工具执行占了全部耗时的90%以上。所以优化的本质只有两件事：减少Token数和减少等待时间。

同样，Agent的钱花在哪？一次Agent调用的Token成本拆解如下：

• 系统提示：固定，每次调用都付
• 工具定义（Schema） ：固定，注册多少工具就付多少
• 对话历史：随轮次线性增长
• 检索内容：动态
• 输出Token：模型思考 + 工具调用参数 + 最终回复

理解了钱和时间的去向，下面我们分三个维度逐个击破。

二、Token消耗优化：让每一分钱都花在刀刃上

2.1 系统提示瘦身——最容易被忽视的固定成本

系统提示在每次调用时都会发送给模型，是最容易被忽视的固定成本。

对比两版系统提示：

# 精简版 → 6 tokens
MINIMAL_PROMPT = "You are a helpful assistant."

# 冗余版 → 107 tokens
VERBOSE_PROMPT = """You are an extremely helpful, knowledgeable, and professional 
AI assistant for our enterprise software platform. You specialize in providing 
accurate information... Always be thorough, comprehensive, and leave no important 
detail unexplained."""

每次调用多付101 tokens。按GPT-4o输入$2.50/1M tokens计算：

• 每天1万次调用 → 每天多花$2.5
• 每天100万次调用 → 每月多花$750

实战建议：系统提示控制在2K tokens以内，能精简的精简，能压缩的压缩。

2.2 Prompt Caching——90%的成本折扣

Claude和OpenAI API都支持显式的提示缓存：

import openai

# 标记系统提示为可缓存
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": LARGE_SYSTEM_PROMPT},
        {"role": "user", "content": user_query}
    ],
    # OpenAI支持自动缓存：相同前缀的prompt自动命中缓存
    # 首次调用：正常计费，写入缓存
    # 后续同前缀调用：约90%成本折扣
)

# 查看缓存命中情况
print(response.usage.prompt_tokens)          # 总输入token
print(response.usage.prompt_tokens_details.cached_tokens)  # 缓存命中token

实测效果：Token消耗降低约45%，缓存成本降低约83%，响应速度提升40%。

2.3 上下文窗口的Token预算管理

长对话场景下，上下文会迅速膨胀。按轮数切分会导致Token数波动极大，正确的做法是按Token数切。

import tiktoken

class TokenBudgetManager:
    def __init__(self, model_name: str, max_tokens: int = 8000):
        self.encoder = tiktoken.encoding_for_model(model_name)
        self.max_tokens = max_tokens
        self.system_prompt_tokens = 0
        self.reserved_tokens = 2000  # 为输出预留
    
    def count_tokens(self, text: str) -> int:
        return len(self.encoder.encode(text))
    
    def build_context(self, system_prompt: str, history: list, 
                      user_query: str, retrieved_docs: list) -> str:
        """在Token预算内动态组装上下文"""
        budget = self.max_tokens - self.reserved_tokens
        
        # 1. 系统提示（固定，优先保留）
        system_tokens = self.count_tokens(system_prompt)
        budget -= system_tokens
        
        # 2. 用户当前问题（必须保留）
        query_tokens = self.count_tokens(user_query)
        budget -= query_tokens
        
        # 3. 检索文档（按相关性截断）
        docs_text = "\n\n".join([d.page_content for d in retrieved_docs])
        docs_tokens = self.count_tokens(docs_text)
        if docs_tokens > budget * 0.5:  # 检索内容最多占50%预算
            docs_text = self._truncate_by_tokens(docs_text, int(budget * 0.5))
            budget -= int(budget * 0.5)
        else:
            budget -= docs_tokens
        
        # 4. 对话历史（滑动窗口 + 摘要压缩）
        history_text = self._build_history(history, budget)
        
        return f"{system_prompt}\n\n{history_text}\n\n用户: {user_query}"
    
    def _truncate_by_tokens(self, text: str, max_tokens: int) -> str:
        tokens = self.encoder.encode(text)
        if len(tokens) <= max_tokens:
            return text
        return self.encoder.decode(tokens[:max_tokens]) + "…(已截断)"
    
    def _build_history(self, history: list, budget: int) -> str:
        """滑动窗口 + 摘要压缩"""
        # 优先保留最近的对话
        recent = []
        tokens_used = 0
        for msg in reversed(history):
            msg_tokens = self.count_tokens(msg)
            if tokens_used + msg_tokens > budget:
                break
            recent.insert(0, msg)
            tokens_used += msg_tokens
        
        if len(recent) < len(history):
            # 有被截断的历史 → 生成摘要
            return f"[历史摘要: 共{len(history)-len(recent)}轮对话已压缩]\n\n" + "\n".join(recent)
        return "\n".join(recent)

核心原则：对窗口外的内容做摘要，不是全部压缩，而是只压缩那些“有结论”的轮次。

2.4 动态工具路由——绕过大模型

不是所有请求都需要大模型。遇到简单的查询，直接调用API返回结构化数据，彻底绕过大模型的长文本生成：

from enum import Enum
from typing import Union

class TaskComplexity(Enum):
    SIMPLE = "simple"      # 直接API调用
    MEDIUM = "medium"      # 小模型处理
    COMPLEX = "complex"    # 完整Agent链路

class DynamicRouter:
    def __init__(self):
        self.simple_patterns = [
            "天气", "时间", "日期", "汇率", "计算"
        ]
    
    def route(self, query: str) -> TaskComplexity:
        """根据意图复杂度路由"""
        # 简单查询 → 直接调用工具，不走LLM
        if any(p in query for p in self.simple_patterns):
            return TaskComplexity.SIMPLE
        
        # 中等复杂度 → 用mini模型（便宜15倍，快2-3倍）
        if len(query) < 50 and "分析" not in query:
            return TaskComplexity.MEDIUM
        
        # 复杂推理 → 完整Agent
        return TaskComplexity.COMPLEX
    
    async def execute(self, query: str, tools: dict):
        complexity = self.route(query)
        
        if complexity == TaskComplexity.SIMPLE:
            # 直接调用工具API，零Token消耗
            return await tools["weather_api"](query)
        
        elif complexity == TaskComplexity.MEDIUM:
            # 使用gpt-4o-mini（便宜15倍）
            return await self._call_mini_model(query)
        
        else:
            # 完整Agent链路
            return await self._run_agent(query)

三、延迟优化：让用户感觉「快了一截」

延迟可以从两个角度切：降低真实延迟和优化感知延迟——两者都重要。

3.1 流式输出——200ms内看到第一个字

最直接的「感知延迟」优化是流式输出，不改任何推理逻辑，只把响应从「等全部生成完再发」改成「边生成边发」。

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import AsyncGenerator, Dict, Any

class StreamingAgent:
    def __init__(self, model: str = "gpt-4o"):
        self.llm = ChatOpenAI(
            model=model,
            streaming=True,  # 开启流式
            temperature=0.7
        )
        self._build_graph()
    
    def _build_graph(self):
        """构建LangGraph工作流"""
        builder = StateGraph(dict)
        builder.add_node("llm", self._llm_node)
        builder.add_node("tools", self._tools_node)
        builder.set_entry_point("llm")
        builder.add_conditional_edges("llm", self._should_continue)
        builder.add_edge("tools", "llm")
        self.graph = builder.compile()
    
    async def _llm_node(self, state: dict) -> dict:
        """LLM节点"""
        messages = state.get("messages", [])
        response = await self.llm.ainvoke(messages)
        return {"messages": messages + [response]}
    
    async def stream(self, query: str, thread_id: str) -> AsyncGenerator[str, None]:
        """流式输出Token"""
        config = {"configurable": {"thread_id": thread_id}}
        
        # 使用astream逐块输出
        async for chunk in self.graph.astream(
            {"messages": [{"role": "user", "content": query}]},
            config=config,
            stream_mode="values"
        ):
            if "messages" in chunk and chunk["messages"]:
                last_msg = chunk["messages"][-1]
                if hasattr(last_msg, "content") and last_msg.content:
                    yield last_msg.content  # 用户200ms内看到第一个字

关键指标：首字延迟（TTFT）从800ms上升到2.5s时，对话完成率下降34%。流式输出是性价比最高的优化手段。

3.2 语义缓存——5ms vs 1500ms

相同问题命中缓存直接返回：5ms vs 1500ms。

from langchain.cache import RedisCache
from langchain.globals import set_llm_cache
import redis
import hashlib
import numpy as np
from sentence_transformers import SentenceTransformer

class SemanticCache:
    """基于向量相似度的语义缓存"""
    
    def __init__(self, redis_url: str = "redis://localhost:6379", 
                 similarity_threshold: float = 0.92):
        self.redis = redis.from_url(redis_url)
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')
        self.threshold = similarity_threshold
        self.ttl = 3600  # 1小时
    
    def _get_embedding(self, text: str) -> np.ndarray:
        return self.encoder.encode(text)
    
    def _hash_key(self, query: str, model: str) -> str:
        """Cache Key = prompt + model + temperature的哈希"""
        content = f"{query}|{model}"
        return hashlib.md5(content.encode()).hexdigest()
    
    async def get(self, query: str, model: str) -> str | None:
        """语义缓存查询"""
        # 1. 精确匹配
        exact_key = self._hash_key(query, model)
        cached = self.redis.get(exact_key)
        if cached:
            return cached.decode()
        
        # 2. 语义相似匹配
        query_vec = self._get_embedding(query)
        # 从Redis中检索相似向量（使用Redis的向量搜索能力）
        similar = self._search_similar(query_vec)
        if similar and similar["score"] >= self.threshold:
            return similar["response"]
        
        return None
    
    async def set(self, query: str, model: str, response: str):
        """写入缓存"""
        key = self._hash_key(query, model)
        self.redis.setex(key, self.ttl, response)
        # 同时存入向量索引供语义检索
        self._index_vector(query, response)

注意：如果prompt里有时间戳或随机ID，缓存命中率会接近0。解法是把动态内容从prompt剥离，Prompt模板本身保持稳定。

3.3 并行工具调用——不再串行等待

如果Agent需要调用3个外部API，最慢的那个决定了整体延迟。串行改并行：

import asyncio
from typing import List, Dict, Any

class ParallelToolExecutor:
    """并行执行多个工具调用"""
    
    def __init__(self, tools: Dict[str, callable]):
        self.tools = tools
    
    async def _execute_one(self, tool_name: str, params: Dict) -> Any:
        """执行单个工具"""
        if tool_name not in self.tools:
            return {"error": f"Tool {tool_name} not found"}
        try:
            result = await self.tools[tool_name](**params)
            return {"tool": tool_name, "result": result, "error": None}
        except Exception as e:
            return {"tool": tool_name, "result": None, "error": str(e)}
    
    async def execute_parallel(self, tool_calls: List[Dict]) -> List[Dict]:
        """并行执行多个工具调用"""
        tasks = [
            self._execute_one(call["name"], call.get("params", {}))
            for call in tool_calls
        ]
        # 并发执行所有工具调用
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results
    
    async def execute_with_timeout(self, tool_calls: List[Dict], 
                                    timeout: float = 5.0) -> List[Dict]:
        """带超时的并行执行"""
        try:
            return await asyncio.wait_for(
                self.execute_parallel(tool_calls),
                timeout=timeout
            )
        except asyncio.TimeoutError:
            # 超时降级：返回已完成的结果，未完成的标记为超时
            return [{"tool": c["name"], "error": "timeout"} for c in tool_calls]

收益：3个工具调用从串行的_{3s降到并行的}1s，延迟降低66% 。

3.4 模型路由——简单任务用mini模型

from langchain_openai import ChatOpenAI

class ModelRouter:
    """按任务复杂度路由到不同模型"""
    
    def __init__(self):
        self.models = {
            "simple": ChatOpenAI(model="gpt-4o-mini", temperature=0.3),
            "medium": ChatOpenAI(model="gpt-4o", temperature=0.5),
            "complex": ChatOpenAI(model="o1-preview", temperature=0.7),
        }
    
    def route(self, query: str, context: dict = None) -> str:
        """根据复杂度选模型"""
        # 80%的简单请求用mini模型
        if len(query) < 30:
            return "simple"
        
        # 需要推理/分析 → 全量模型
        if any(kw in query for kw in ["推理", "设计", "分析", "代码"]):
            return "complex" if len(query) > 500 else "medium"
        
        return "simple"
    
    async def invoke(self, query: str, context: dict = None):
        tier = self.route(query, context)
        model = self.models[tier]
        return await model.ainvoke(query)

收益：gpt-4o-mini便宜15倍，速度快2-3倍。通过路由，80%的请求可以用mini模型处理。

四、吞吐量优化：从QPS=10到QPS=100+

4.1 异步I/O——提升吞吐量的关键

采用异步I/O模型是提升吞吐量的关键：

import asyncio
from typing import List, Dict, Any
from langgraph.graph import StateGraph

class AsyncAgentPool:
    """异步Agent池，支持高并发"""
    
    def __init__(self, agent, max_concurrent: int = 10):
        self.agent = agent
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    async def process(self, query: str, user_id: str) -> Dict[str, Any]:
        """带并发限制的请求处理"""
        async with self.semaphore:
            config = {"configurable": {"thread_id": user_id}}
            result = await self.agent.graph.ainvoke(
                {"messages": [{"role": "user", "content": query}]},
                config=config
            )
            return {"user_id": user_id, "result": result}
    
    async def process_batch(self, queries: List[tuple]) -> List[Dict]:
        """批量并发处理"""
        tasks = [self.process(q, uid) for q, uid in queries]
        return await asyncio.gather(*tasks)

4.2 LangGraph的「状态税」与规避策略

在同一benchmark下，LangGraph平均延迟10,155ms vs LangChain 6,046ms（慢68% ），吞吐2.70 rps vs 4.26 rps（低37% ）。差的这4.1秒不是框架低效，是 “状态税” 。

规避策略：

from langgraph.graph import StateGraph, END
from langgraph.checkpoint import MemorySaver

class OptimizedGraph:
    def __init__(self):
        # 1. 使用内存checkpoint而非持久化（测试环境）
        # 2. 精简State结构，只存必要字段
        self.checkpointer = MemorySaver()
        self._build_graph()
    
    def _build_graph(self):
        builder = StateGraph(dict)  # 使用简单dict而非复杂Pydantic模型
        
        # 3. 减少节点数量，合并简单操作
        builder.add_node("process", self._process_node)
        # 而非拆成3个独立节点
        
        builder.set_entry_point("process")
        builder.add_edge("process", END)
        self.graph = builder.compile(checkpointer=self.checkpointer)

4.3 向量检索优化——让RAG不再拖后腿

RAG的首字延迟主要卡在embedding和向量检索：

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

class OptimizedRetriever:
    def __init__(self):
        # 1. 降低向量维度：512 vs 1536，速度提升40%
        self.embeddings = OpenAIEmbeddings(
            model="text-embedding-3-small",
            dimensions=512  # 默认1536，降到512
        )
        
        # 2. 使用HNSW索引加速检索
        self.vectorstore = Chroma(
            embedding_function=self.embeddings,
            collection_metadata={"hnsw:space": "cosine"}
        )
    
    async def retrieve(self, query: str, top_k: int = 5) -> List:
        """异步检索"""
        # 3. 限制召回数量，减少后续处理
        docs = await self.vectorstore.asimilarity_search(query, k=top_k)
        return docs

优化收益：Recall@5仅下降1.2%，检索速度提升40%。

4.4 全链路异步架构

from fastapi import FastAPI
import uvicorn

app = FastAPI()

class AgentService:
    def __init__(self):
        self.agent = OptimizedAgent()
        self.cache = SemanticCache()
        self.router = ModelRouter()
        self.pool = AsyncAgentPool(self.agent, max_concurrent=20)
    
    async def handle(self, query: str, user_id: str) -> dict:
        # 1. 查缓存
        cached = await self.cache.get(query, "gpt-4o")
        if cached:
            return {"source": "cache", "response": cached}
        
        # 2. 路由决策
        tier = self.router.route(query)
        
        # 3. 执行
        result = await self.pool.process(query, user_id)
        
        # 4. 写缓存
        await self.cache.set(query, "gpt-4o", result["result"])
        
        return {"source": "agent", "response": result}

@app.post("/agent")
async def agent_endpoint(query: str, user_id: str):
    service = AgentService()
    return await service.handle(query, user_id)

# 启动：uvicorn main:app --workers 4

多worker部署可进一步提升吞吐量。

五、量化效果总结

优化维度	优化手段	量化收益
Token	系统提示瘦身	每次节省~100 tokens
Token	Prompt Caching	成本降低~90%
Token	动态工具路由	绕过LLM，Token归零
延迟	流式输出	TTFT从秒级降到200ms内
延迟	语义缓存	5ms vs 1500ms
延迟	并行工具调用	3工具从3s降到~1s
吞吐	异步I/O + 连接池	QPS提升3-5倍
吞吐	降低向量维度	检索速度提升40%

核心原则：减少LLM被调用的次数、降低每次调用的输入Token、让工具执行不再串行等待。优化不是一次性工作，需要定期审查检索日志与Token账单，将优化动作常态化。