性能调优实战

5.3 性能调优实战

本节摘要：OpenClaw 部署上线只是起点，用户反馈"响应太慢""并发一高就卡""Token 费用超标"才是真正的考验。本节从五类常见性能瓶颈（模型调用延迟、上下文膨胀、技能加载慢、并发不足、I/O 阻塞）入手，逐层给出模型选择、Token 优化、会话管理、缓存策略、异步 I/O、负载均衡的实战方案，附带三个真实优化案例——响应时间从 10 秒压到 1.2 秒、Token 消耗降低 50%、并发能力提升 10 倍。

本节导读

阅读完本节，你应当能够：

1. 识别五类常见性能瓶颈的症状和根因
2. 根据任务复杂度动态选择模型，平衡速度、成本和质量
3. 实施会话历史管理策略，防止上下文无限膨胀
4. 搭建多级缓存体系（内存 → Redis → 数据库），减少重复计算
5. 使用异步 I/O 和连接池提升系统吞吐能力

一、性能瓶颈分析：先找到卡在哪里

性能优化的第一条原则是"测量优先"。不知道瓶颈在哪就动手优化，等于蒙眼开车。

1.1 五类常见瓶颈

模型调用延迟是最直观的性能问题。简单问题也要等好几秒，不同模型的响应时间差异巨大。根因通常是选了过大的模型、模型 API 本身的延迟高、或者网络传输慢。

上下文过长是一个"温水煮青蛙"的问题。对话刚开始时响应很快，聊着聊着越来越慢，Token 消耗快速增长，最终达到上下文限制后直接报错。根因是会话历史无限增长、技能加载过多、重复信息没有去重。

技能加载慢影响首次体验。技能文件过大、结构不合理、没有使用渐进式加载，都会导致首次使用技能时等待时间过长，内存占用也随之飙升。

并发处理能力弱在多用户场景下暴露。单线程处理、无请求队列管理、资源未池化，导致多用户同时使用时出现卡顿、队列堆积、超时错误。

I/O 操作慢是底层瓶颈。同步 I/O 阻塞主线程、无缓存机制导致重复读取、数据库查询未优化，这些都会拖慢整个系统的响应速度。

二、模型选择与 Token 优化：花最少的钱办最多的事

模型调用是 OpenClaw 最耗时的环节，也是最花钱的环节。优化模型使用策略，一举两得。

2.1 智能模型路由

不是所有任务都需要最强的模型。日常闲聊用轻量模型就够了，只有复杂推理才需要大模型。根据输入长度动态选择模型，是一个简单但有效的策略：

def select_model(input_length):
    if input_length < 500:
        return "gpt-3.5-turbo"       # 便宜快速，适合简单问答
    elif input_length < 2000:
        return "claude-3-haiku"       # 平衡之选，中等复杂度
    else:
        return "claude-3-5-sonnet"    # 重量级选手，处理复杂任务
​

在配置层面，可以为不同场景预设不同的模型配置：

{
  agents: {
    models: {
      simple: {
        model: "gpt-3.5-turbo",
        maxTokens: 500,
        temperature: 0.7
      },
      medium: {
        model: "claude-3-haiku",
        maxTokens: 2000,
        temperature: 0.5
      },
      complex: {
        model: "claude-3-5-sonnet",
        maxTokens: 4000,
        temperature: 0.3
      }
    }
  }
}
​

2.2 Token 消耗控制

Token 消耗直接关联 API 费用。三个控制手段立竿见影：

控制 Max Tokens。根据响应类型设置合理的上限——简单问答 500 到 1000 tokens，代码生成 1000 到 2000 tokens，长文本生成 2000 到 4000 tokens。不要一刀切地设成 4000。

启用流式响应。流式输出能显著降低首字延迟，用户不用等整个响应生成完毕就能看到结果：

{
  agents: {
    defaults: {
      stream: true  // 流式响应，降低感知延迟
    }
  }
}
​

精简系统提示。系统提示是每次请求都要发送的，它的 Token 数会乘以请求次数。把系统提示从 500 字压到 100 字，长期下来节省的 Token 相当可观。

2.3 响应缓存

相同或相似的问题不需要每次都调用模型。响应缓存能大幅减少重复调用：

import hashlib
import json

def cache_key(model, messages):
    content = json.dumps(messages, sort_keys=True)
    return hashlib.md5(f"{model}:{content}".encode()).hexdigest()

def cached_response(model, messages):
    key = cache_key(model, messages)
    cached = redis.get(f"cache:{key}")
    
    if cached:
        return json.loads(cached)
    
    response = call_llm(model, messages)
    redis.setex(f"cache:{key}", 300, json.dumps(response))
    return response
​

对于语义相似但不完全相同的问题，可以使用向量检索来匹配缓存。把历史问答向量化存入向量数据库，新问题进来时先搜索相似问题，相似度超过阈值就直接返回缓存结果。

💡 经验之谈：缓存是性能优化的"低垂果实"。实现简单，效果立竿见影。但要设好过期时间——过期的缓存比没有缓存更危险，用户会拿到过时的信息。

三、会话管理优化：别让历史拖垮性能

3.1 历史长度控制

会话历史无限增长是性能退化的头号杀手。每多一条历史消息，发送给模型的 Token 就多一截，响应就更慢一点，费用就更高一点。

设置明确的历史长度上限——最多保留 50 条消息，或者最多 10000 tokens，两者取先到者。当历史超过阈值时，自动触发总结：

{
  sessions: {
    history: {
      maxMessages: 50,
      maxTokens: 10000,
      summarizeThreshold: 20  // 超过 20 条时触发总结
    }
  }
}
​

自动总结的思路是把旧消息压缩成一段摘要，只保留最近的对话原文。这样既保留了上下文连贯性，又控制了 Token 消耗：

def summarize_if_needed(messages):
    token_count = count_tokens(messages)
    
    if token_count > 10000:
        old_messages = messages[:-10]
        summary = generate_summary(old_messages)
        return [
            {"role": "system", "content": f"历史总结：{summary}"},
            *messages[-10:]
        ]
    return messages
​

3.2 分层存储

不同时间段的会话数据，访问频率差异很大。最近 1 小时的会话频繁读写，用 Redis 内存存储；更早的会话偶尔才需要回溯，归档到 PostgreSQL：

# Redis：最近 1 小时的会话（热数据）
redis.setex(f"session:{session_id}", 3600, json.dumps(messages))

# PostgreSQL：归档的会话（冷数据）
pg.execute(
    "INSERT INTO sessions_archive (session_id, messages) VALUES ($1, $2)",
    session_id, json.dumps(messages)
)
​

四、技能与 I/O 优化：把慢操作变快

4.1 技能渐进式加载

技能文件不应该一次性全部加载到内存。主文档保持简洁，只包含基础操作说明；高级功能和参考资料放在单独的文件中，按需加载：

# SKILL.md

## 快速开始
基础操作说明...

## 高级功能
参见 references/ADVANCED.md
​

常用技能可以预加载到内存中，避免首次使用时的磁盘读取延迟：

class SkillPreloader:
    def __init__(self):
        self.loaded_skills = {}
    
    def preload(self, skill_names):
        for name in skill_names:
            if name not in self.loaded_skills:
                skill = load_skill_from_disk(name)
                self.loaded_skills[name] = skill
​

4.2 异步 I/O

同步 I/O 是并发场景下的性能杀手。一个文件读取操作阻塞了主线程，其他所有请求都得排队等。改用异步 I/O，让等待时间重叠而非叠加：

import aiofiles
import asyncio

async def read_file_async(file_path):
    async with aiofiles.open(file_path, "r") as f:
        return await f.read()

async def process_files(file_paths):
    tasks = [read_file_async(fp) for fp in file_paths]
    return await asyncio.gather(*tasks)
​

数据库连接池是另一个关键优化。不要每次查询都新建连接，预先创建一组连接复用：

import asyncpg

class DatabasePool:
    async def init(self, dsn):
        self.pool = await asyncpg.create_pool(dsn, min_size=5, max_size=20)
    
    async def query(self, sql, *args):
        async with self.pool.acquire() as conn:
            return await conn.fetch(sql, *args)
​

4.3 多级缓存

单一缓存层不够用。内存缓存最快但容量有限，Redis 缓存容量大但多一次网络往返，数据库持久化但最慢。三级缓存配合使用，各取所长：

class MultiLevelCache:
    def __init__(self):
        self.l1 = {}              # L1：内存缓存（最快）
        self.l2 = redis_client    # L2：Redis 缓存（容量大）
        # L3：数据库（持久化）
    
    def get(self, key):
        if key in self.l1:
            return self.l1[key]
        
        value = self.l2.get(f"cache:{key}")
        if value:
            self.l1[key] = value
            return value
        
        value = self.fetch_from_db(key)
        if value:
            self.l1[key] = value
            self.l2.setex(f"cache:{key}", 300, value)
        return value
​

⚠️ 别过度缓存：缓存的一致性问题是分布式系统的经典难题。如果数据频繁变更，缓存过期策略要设短一些。宁可多查一次数据库，也不要让用户看到过期数据。

五、并发与负载均衡：让更多人同时用

5.1 请求队列

当并发请求超过处理能力时，队列是必需的缓冲层。工作线程从队列中取请求、处理、返回结果，避免请求堆积导致超时：

import queue
import threading

class RequestQueue:
    def __init__(self, max_size=1000):
        self.queue = queue.Queue(maxsize=max_size)
    
    def start_workers(self, num_workers=5):
        for i in range(num_workers):
            worker = threading.Thread(target=self._worker, daemon=True)
            worker.start()
    
    def _worker(self):
        while True:
            request = self.queue.get()
            try:
                response = self.process_request(request)
                self.send_response(request, response)
            finally:
                self.queue.task_done()
​

5.2 速率限制

速率限制保护系统不被单个客户端压垮。令牌桶算法是常用的限流方案——以固定速率往桶里放令牌，每次请求消耗一个令牌，桶空了就拒绝请求：

import time

class RateLimiter:
    def __init__(self, rate, per):
        self.rate = rate
        self.per = per
        self.allowance = rate
        self.last_check = time.time()
    
    def can_proceed(self):
        current = time.time()
        time_passed = current - self.last_check
        self.last_check = current
        self.allowance += time_passed * (self.rate / self.per)
        if self.allowance > self.rate:
            self.allowance = self.rate
        if self.allowance < 1:
            return False
        self.allowance -= 1
        return True
​

5.3 负载均衡

多实例部署时，负载均衡把请求分散到不同节点。轮询、随机、最少连接数是三种常用策略：

class LoadBalancer:
    def __init__(self, servers):
        self.servers = servers
        self.current = 0
    
    def next_server(self):
        server = self.servers[self.current]
        self.current = (self.current + 1) % len(self.servers)
        return server
    
    def least_connections_server(self, connections):
        return min(self.servers, key=lambda s: connections.get(s, 0))
​

六、实战案例：三个真实优化故事

案例一：响应时间从 10 秒压到 1.2 秒

问题：用户反馈平均响应时间 10 秒，体验极差。

诊断：使用了最大号模型处理所有请求；保留完整对话历史，每次请求发送 5000 以上 tokens；所有请求同步处理。

优化动作：根据任务复杂度动态选择模型（简单问题用轻量模型）；总结历史对话，只保留最近 10 条消息；实现异步处理。

结果：平均响应时间从 10 秒降到 1.2 秒，Token 消耗降低 60%，并发能力提升 300%。

案例二：Token 消耗降低 50%

问题：API 费用月月超标。

诊断：Max Tokens 一刀切设为 4000；系统提示长达 1000 tokens；相同问题反复调用模型。

优化动作：根据响应类型动态设置 Max Tokens；精简系统提示到 300 tokens；实现响应缓存。

结果：Token 消耗降低 50%，响应速度提升 30%，用户体验没有下降。

案例三：并发能力提升 10 倍

问题：5 个用户同时使用就开始卡顿。

诊断：单线程处理所有请求；没有请求队列；数据库连接每次新建。

优化动作：实现异步处理；添加请求队列；使用数据库连接池。

结果：并发能力从 5 提升到 50，P95 响应时间从 8 秒降到 3 秒，超时率从 20% 降到 1% 以下。

七、性能优化检查清单

模型层

• 根据任务复杂度选择合适的模型
• 优化 Max Tokens 设置，避免一刀切
• 精简系统提示，减少固定 Token 开销
• 实现响应缓存，减少重复调用

会话层

• 限制历史长度，防止上下文膨胀
• 实现自动总结机制
• 使用分层存储（热数据内存、冷数据数据库）

技能层

• 使用渐进式加载，避免一次性加载全部技能
• 脚本化重复逻辑，减少模型生成代码的开销
• 预加载常用技能到内存

I/O 层

• 使用异步 I/O 替代同步操作
• 实现批量操作，减少网络往返
• 配置多级缓存（内存 → Redis → 数据库）
• 使用数据库连接池

并发层

• 实现请求队列缓冲
• 配置速率限制保护
• 使用负载均衡分散压力

本节要点

1. 测量优先——不知道瓶颈在哪就优化，等于蒙眼开车。先采集指标，再针对性优化
2. 模型路由是性价比最高的优化——简单任务用小模型，复杂任务用大模型，Token 消耗立刻降下来
3. 会话历史管理防止"温水煮青蛙"式的性能退化——设上限、做总结、分层存储
4. 缓存是低垂果实——多级缓存体系（内存 → Redis → 数据库）能挡住大部分重复请求
5. 异步 I/O + 连接池是并发场景的必备——同步阻塞是吞吐量最大的敌人
6. 优化是持续过程——定期回顾性能指标，根据实际使用模式调整策略