阅读时间：约 14 分钟
前置知识：D01 监控、D03 灰度发布

---

D01 让你知道延迟高了，D03 让你能安全上线新版本。现在面对一个直接的问题：用户说"你们的 AI 太慢了"，老板说"QPS 要到 1000"，账单上 Token 费比服务器费高 10 倍。

优化不是玄学。三个方向：减少等待（异步）、减少重复（缓存）、减少无效调用（批处理）。

---

前言：Agent 的性能瓶颈在哪？

搞清楚一件事：Agent 慢在哪？

环节	典型耗时	占比
LLM 调用	1-10s	60-80%
工具调用（搜索/数据库）	0.1-2s	10-20%
Prompt 组装	< 0.1s	< 5%
网络传输	0.05-0.5s	5-10%

大头在 LLM 调用。 优化应该从这里开始，不是去调 Python 的 GC。

📌 本章核心：异步（不阻塞主线程）→ 缓存（不重复调用）→ 批处理（不逐个调用）→ 连接池（不重复建连）。四层递进。

---

第一部分：异步处理

1.1 流式返回

import asyncio
from typing import AsyncIterator

class StreamingAgent:
    """
    流式 Agent：边生成边返回
    用户感知延迟大幅降低
    """
    
    async def process_stream(self, query: str) -> AsyncIterator[str]:
        """
        流式处理，逐 token 返回
        """
        # 模拟 LLM 流式输出
        response = f"这是对 '{query}' 的回答：RAG 是检索增强生成技术，它通过..."
        
        for i, char in enumerate(response):
            yield char
            await asyncio.sleep(0.01)  # 模拟 token 生成延迟
        
        # 可以边生成边做后处理
        yield "\n\n---\n*生成时间: 2.3s*"
    
    async def handle_request(self, query: str):
        """完整的流式请求处理"""
        result_parts = []
        
        async for chunk in self.process_stream(query):
            result_parts.append(chunk)
            # 实际项目中这里通过 WebSocket/SSE 推送给前端
        
        return "".join(result_parts)

1.2 并发工具调用

Agent 经常需要调多个工具，串行调用浪费时间。

class ConcurrentToolExecutor:
    """
    并发工具调用
    多个独立工具同时执行，不等前一个完成
    """
    
    async def execute_parallel(self, tools_to_call: list) -> list:
        """
        并发执行多个工具调用
        tools_to_call: [
            {"name": "search_web", "args": {...}},
            {"name": "search_db", "args": {...}},
            {"name": "get_weather", "args": {...}}
        ]
        """
        async def call_tool(tool_info):
            start = time.time()
            try:
                result = await self._execute_single(tool_info)
                return {
                    "tool": tool_info["name"],
                    "success": True,
                    "result": result,
                    "latency_ms": (time.time() - start) * 1000
                }
            except Exception as e:
                return {
                    "tool": tool_info["name"],
                    "success": False,
                    "error": str(e),
                    "latency_ms": (time.time() - start) * 1000
                }
        
        # 并发执行所有工具
        tasks = [call_tool(t) for t in tools_to_call]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        return list(results)
    
    async def _execute_single(self, tool_info: dict):
        """执行单个工具"""
        await asyncio.sleep(0.5)  # 模拟工具调用
        return f"[{tool_info['name']}] 的结果"


# 对比：串行 vs 并发
async def benchmark():
    executor = ConcurrentToolExecutor()
    tools = [
        {"name": "search_web", "args": {}},
        {"name": "search_db", "args": {}},
        {"name": "get_weather", "args": {}},
        {"name": "check_calendar", "args": {}}
    ]
    
    # 串行（4 个工具各 0.5s = 2s）
    start = time.time()
    for tool in tools:
        await executor._execute_single(tool)
    serial_time = time.time() - start
    
    # 并发（4 个工具同时跑 = 0.5s）
    start = time.time()
    await executor.execute_parallel(tools)
    parallel_time = time.time() - start
    
    print(f"串行: {serial_time:.1f}s, 并发: {parallel_time:.1f}s")
    print(f"提速: {serial_time/parallel_time:.1f}x")

📌 本章要点：两个异步优化点：流式返回（用户感知延迟降 50%+）和并发工具调用（提速 2-4 倍）。独立工具一定要并发，不用等。

---

第二部分：缓存

2.1 多级缓存

import hashlib
import time
from functools import lru_cache

class MultiLevelCache:
    """
    多级缓存：内存 → 本地磁盘 → 远程 Redis
    每层有不同的容量和速度
    """
    
    def __init__(self):
        # L1: 内存缓存（最快，最小）
        self.memory_cache = {}       # key -> (value, expire_time)
        self.memory_max_size = 1000
        
        # L2: 本地磁盘缓存（较慢，较大）
        self.disk_cache_dir = "./cache"
        
        # L3: 远程缓存（最慢，最大）
        self.redis_client = None  # 实际项目中连接 Redis
    
    def get(self, key: str) -> str:
        """多级缓存查询，从快到慢逐级查找"""
        
        # L1: 内存
        mem_key = self._hash(key)
        if mem_key in self.memory_cache:
            value, expire = self.memory_cache[mem_key]
            if time.time() < expire:
                return value
            else:
                del self.memory_cache[mem_key]
        
        # L2: 本地磁盘
        disk_value = self._disk_get(mem_key)
        if disk_value:
            # 回填到 L1
            self.memory_cache[mem_key] = (disk_value, time.time() + 3600)
            return disk_value
        
        # L3: 远程 Redis
        # redis_value = self.redis_client.get(mem_key)
        # if redis_value:
        #     self._disk_set(mem_key, redis_value)
        #     self.memory_cache[mem_key] = (redis_value, time.time() + 3600)
        #     return redis_value
        
        return None
    
    def set(self, key: str, value: str, ttl: int = 3600):
        """写入所有缓存层"""
        mem_key = self._hash(key)
        expire = time.time() + ttl
        
        # L1
        self.memory_cache[mem_key] = (value, expire)
        if len(self.memory_cache) > self.memory_max_size:
            # 淘汰最老的 10%
            self._evict_memory()
        
        # L2
        self._disk_set(mem_key, value)
        
        # L3
        # self.redis_client.setex(mem_key, ttl, value)
    
    def _hash(self, key: str) -> str:
        return hashlib.md5(key.encode()).hexdigest()
    
    def _disk_get(self, key: str) -> str:
        import os
        path = os.path.join(self.disk_cache_dir, f"{key}.cache")
        try:
            with open(path, "r", encoding="utf-8") as f:
                return f.read()
        except FileNotFoundError:
            return None
    
    def _disk_set(self, key: str, value: str):
        import os
        os.makedirs(self.disk_cache_dir, exist_ok=True)
        path = os.path.join(self.disk_cache_dir, f"{key}.cache")
        with open(path, "w", encoding="utf-8") as f:
            f.write(value)
    
    def _evict_memory(self):
        """淘汰最老的 10%"""
        sorted_keys = sorted(
            self.memory_cache.items(),
            key=lambda x: x[1][1]
        )
        evict_count = max(1, len(sorted_keys) // 10)
        for key, _ in sorted_keys[:evict_count]:
            del self.memory_cache[key]

2.2 回答缓存（结果级）

class ResponseCache:
    """
    回答缓存
    缓存完整 LLM 回答，避免重复调用
    """
    
    def __init__(self, ttl: int = 1800):
        self.ttl = ttl  # 默认 30 分钟
        self.cache = MultiLevelCache()
    
    def get_response(self, query: str) -> str:
        """获取缓存的回答"""
        return self.cache.get(f"response:{query}")
    
    def set_response(self, query: str, response: str):
        """缓存回答"""
        self.cache.set(f"response:{query}", response, self.ttl)
    
    def get_or_generate(self, query: str, generate_fn) -> str:
        """
        获取缓存或生成
        如果缓存命中就直接返回，否则调 generate_fn 生成并缓存
        """
        cached = self.get_response(query)
        if cached:
            return cached
        
        response = generate_fn(query)
        self.set_response(query, response)
        return response

📌 本章要点：多级缓存（内存 → 磁盘 → Redis），逐级回填。回答缓存直接把整个 LLM 调用省掉，命中率 10-30% 就能省 10-30% 的 Token 费。

---

第三部分：批处理

3.1 请求批处理

import asyncio
from collections import deque

class RequestBatcher:
    """
    请求批处理器
    把短时间内的多个请求合并成一次批量调用
    """
    
    def __init__(self, max_batch_size: int = 10, max_wait_ms: int = 100):
        self.max_batch_size = max_batch_size
        self.max_wait_ms = max_wait_ms
        self.queue = deque()
        self.waiting = None  # 等待中的 future
    
    async def submit(self, request) -> any:
        """
        提交请求，可能被批处理
        """
        future = asyncio.Future()
        self.queue.append((request, future))
        
        # 如果队列达到批处理大小，立即执行
        if len(self.queue) >= self.max_batch_size:
            await self._flush()
        elif self.waiting is None:
            # 启动等待计时器
            self.waiting = asyncio.create_task(self._wait_and_flush())
        
        return await future
    
    async def _wait_and_flush(self):
        """等待 max_wait_ms 后执行批处理"""
        try:
            await asyncio.sleep(self.max_wait_ms / 1000)
            await self._flush()
        finally:
            self.waiting = None
    
    async def _flush(self):
        """执行批处理"""
        if not self.queue:
            return
        
        # 取出当前队列中的所有请求
        batch = []
        while self.queue:
            batch.append(self.queue.popleft())
        
        requests = [r for r, _ in batch]
        futures = [f for _, f in batch]
        
        # 批量执行
        results = await self._batch_process(requests)
        
        # 分配结果
        for future, result in zip(futures, results):
            if not future.done():
                future.set_result(result)
    
    async def _batch_process(self, requests: list) -> list:
        """批量处理（实际项目中批量调 LLM API）"""
        await asyncio.sleep(0.3)  # 模拟批量调用
        return [f"批量处理结果: {r}" for r in requests]

3.2 嵌入批处理

向量检索时需要批量编码，一条一条编码很慢。

class BatchEmbeddingEncoder:
    """
    批量嵌入编码器
    把多个文本一次性编码，减少 API 调用次数
    """
    
    def __init__(self, batch_size: int = 32):
        self.batch_size = batch_size
        self.pending_texts = []
    
    async def encode(self, text: str) -> list:
        """编码单个文本"""
        self.pending_texts.append(text)
        
        if len(self.pending_texts) >= self.batch_size:
            return await self._batch_encode()
        
        # 返回单个编码（模拟）
        return [0.0] * 768
    
    async def _batch_encode(self):
        """批量编码"""
        texts = self.pending_texts.copy()
        self.pending_texts.clear()
        
        # 实际项目：一次性调 embedding API
        await asyncio.sleep(0.1)
        return [[0.0] * 768] * len(texts)

📌 本章要点：批处理把多个小请求合成一个大请求。LLM 批处理减少 API 调用次数，嵌入批处理减少编码延迟。max_wait_ms 控制最大等待时间，不让用户等太久。

---

第四部分：连接池与限流

4.1 LLM 连接池

class LLMConnectionPool:
    """
    LLM 连接池
    复用 HTTP 连接，避免每次请求都建连
    """
    
    def __init__(self, api_base: str, api_key: str, pool_size: int = 10):
        self.api_base = api_base
        self.api_key = api_key
        self.pool_size = pool_size
        self.semaphore = asyncio.Semaphore(pool_size)
        self.session = None
    
    async def __aenter__(self):
        import aiohttp
        self.session = aiohttp.ClientSession(
            base_url=self.api_base,
            headers={"Authorization": f"Bearer {self.api_key}"},
            timeout=aiohttp.ClientTimeout(total=30)
        )
        return self
    
    async def __aexit__(self, *args):
        if self.session:
            await self.session.close()
    
    async def call_llm(self, messages: list, model: str = "deepseek-v4-pro") -> dict:
        """
        通过连接池调用 LLM
        semaphore 控制并发数，防止打爆 API
        """
        async with self.semaphore:
            async with self.session.post(
                "/v1/chat/completions",
                json={"model": model, "messages": messages}
            ) as resp:
                return await resp.json()

4.2 自适应限流

class AdaptiveRateLimiter:
    """
    自适应限流
    根据 API 返回的剩余配额动态调整限流阈值
    """
    
    def __init__(self, initial_rate: int = 10):
        self.current_rate = initial_rate  # 当前允许的 QPS
        self.min_rate = 1
        self.max_rate = 100
        self.semaphore = asyncio.Semaphore(initial_rate)
        self.consecutive_successes = 0
        self.consecutive_failures = 0
    
    async def acquire(self):
        """获取许可"""
        await self.semaphore.acquire()
    
    def release(self):
        """释放许可"""
        self.semaphore.release()
    
    def report_success(self):
        """报告成功，可能提升限流"""
        self.consecutive_successes += 1
        self.consecutive_failures = 0
        
        # 连续 10 次成功后，提升 10%
        if self.consecutive_successes >= 10:
            new_rate = min(self.current_rate * 1.1, self.max_rate)
            if new_rate > self.current_rate:
                self._adjust_rate(int(new_rate))
            self.consecutive_successes = 0
    
    def report_failure(self, is_rate_limit: bool = False):
        """报告失败"""
        self.consecutive_failures += 1
        self.consecutive_successes = 0
        
        if is_rate_limit:
            # 遇到限流，立即降到一半
            self._adjust_rate(max(self.current_rate // 2, self.min_rate))
        elif self.consecutive_failures >= 3:
            # 连续 3 次失败，降低 20%
            new_rate = max(self.current_rate * 0.8, self.min_rate)
            self._adjust_rate(int(new_rate))
            self.consecutive_failures = 0
    
    def _adjust_rate(self, new_rate: int):
        """调整限流阈值"""
        old_rate = self.current_rate
        self.current_rate = new_rate
        
        # 重新创建 semaphore
        self.semaphore = asyncio.Semaphore(new_rate)
        
        print(f"限流调整: {old_rate} → {new_rate} QPS")

📌 本章要点：连接池复用 HTTP 连接，semaphore 控制并发数。自适应限流根据 API 返回动态调整 QPS，遇到限流自动降速，连续成功自动提速。

---

第五部分：Prompt 优化

5.1 减少 Prompt 长度

class PromptOptimizer:
    """
    Prompt 优化：减少 Token 消耗，降低延迟
    """
    
    @staticmethod
    def estimate_tokens(text: str) -> int:
        """估算 Token 数（中文约 1 字 = 1.5 token）"""
        chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
        other_chars = len(text) - chinese_chars
        return int(chinese_chars * 1.5 + other_chars * 0.3)
    
    @staticmethod
    def optimize_system_prompt(prompt: str) -> str:
        """
        压缩 system prompt
        """
        lines = prompt.split('\n')
        optimized = []
        
        for line in lines:
            stripped = line.strip()
            
            # 去掉空行
            if not stripped:
                continue
            
            # 去掉纯注释
            if stripped.startswith('#') and len(stripped) < 5:
                continue
            
            # 去掉冗余修饰词
            stripped = stripped.replace('请务必', '')
            stripped = stripped.replace('需要注意的是，', '')
            stripped = stripped.replace('特别提醒：', '')
            
            optimized.append(stripped)
        
        return '\n'.join(optimized)
    
    @staticmethod
    def measure_savings(original: str, optimized: str) -> dict:
        """衡量优化效果"""
        orig_tokens = PromptOptimizer.estimate_tokens(original)
        opt_tokens = PromptOptimizer.estimate_tokens(optimized)
        
        return {
            "original_tokens": orig_tokens,
            "optimized_tokens": opt_tokens,
            "saved": orig_tokens - opt_tokens,
            "saved_pct": round((orig_tokens - opt_tokens) / orig_tokens * 100, 1) if orig_tokens > 0 else 0
        }

5.2 动态上下文裁剪

class DynamicContextTrimmer:
    """
    动态上下文裁剪
    根据查询类型裁剪无关上下文
    """
    
    def __init__(self, max_tokens: int = 4000):
        self.max_tokens = max_tokens
    
    def trim_context(self, query: str, documents: list) -> list:
        """
        根据查询相关度裁剪文档列表
        """
        from sentence_transformers import SentenceTransformer
        import numpy as np
        
        model = SentenceTransformer("bge-large-zh")
        query_vec = model.encode(query)
        
        # 计算每个文档与查询的相关度
        scored_docs = []
        for doc in documents:
            doc_vec = model.encode(doc[:200])
            score = float(np.dot(query_vec, doc_vec))
            scored_docs.append((doc, score))
        
        # 按相关度排序
        scored_docs.sort(key=lambda x: x[1], reverse=True)
        
        # 在 Token 预算内选择文档
        selected = []
        total_tokens = 0
        
        for doc, score in scored_docs:
            tokens = PromptOptimizer.estimate_tokens(doc)
            if total_tokens + tokens > self.max_tokens:
                break
            selected.append(doc)
            total_tokens += tokens
        
        return selected

📌 本章要点：优化 Prompt 也能大幅降低延迟：system prompt 精简、冗余修饰词删除、上下文动态裁剪。省下来的 Token 就是省下来的时间。

---

第六部分：性能优化 checklist

优化项	预期效果	实施难度	优先级
流式返回	用户感知延迟降 50%	低	🔴 高
并发工具调用	延迟降 2-4 倍	中	🔴 高
回答缓存	Token 费降 10-30%	低	🔴 高
连接池	连接开销降 80%	低	🟡 中
批处理	API 调用次数降 5-10 倍	中	🟡 中
动态上下文裁剪	Prompt Token 降 20-40%	中	🟡 中
多级缓存	缓存命中率提升 20%	高	🟢 低
自适应限流	避免被限流	中	🟢 低

优化顺序建议：流式返回 → 并发工具调用 → 回答缓存 → 连接池 → 批处理。这个顺序投入小、收益大。

---

总结

1. 大头在 LLM。Agent 性能瓶颈 60-80% 在 LLM 调用，先优化这里。
2. 流式返回让用户不等。边生成边返回，感知延迟降 50%。
3. 独立工具并发调。多个工具同时执行，不等前一个。
4. 回答缓存直接省掉 LLM 调用。命中率 10-30%，省 10-30% Token 费。
5. 批处理减少 API 调用次数。max_wait_ms 控制不让用户等太久。
6. 连接池复用和自适应限流。避免建连开销，动态调整 QPS。

🤔 你的 Agent 最慢的环节是哪个？ 先量一下延迟分布，再决定优化方向。

---

思维导图

• 性能优化
  • 异步处理
    • 流式返回
    • 并发工具调用
  • 缓存
    • 多级缓存（内存/磁盘/Redis）
    • 回答缓存
  • 批处理
    • 请求批处理器
    • 嵌入批处理
  • 连接池与限流
    • LLM 连接池
    • 自适应限流
  • Prompt 优化
    • 精简 system prompt
    • 动态上下文裁剪
  • 优化 checklist
    • 按优先级排序
    • 流式返回 → 并发 → 缓存 → 连接池 → 批处理