"凌晨2点37分，我看着监控面板上的响应时间曲线，终于理解了什么叫'时间就是金钱'——原来每一毫秒都是用户体验。"
Agent性能调优（Agent Performance Tuning）是指通过分析和优化Agent系统的各个组件，提高响应速度、降低资源消耗、提升整体效率的过程。
性能调优就像给Agent做体检。你不会等到生病了才去医院吧？你会定期检查，发现问题，及时治疗。
就像周星驰在《少林足球》里说的："做人如果没有梦想，那跟咸鱼有什么区别？"做Agent如果没有性能优化，那跟蜗牛有什么区别？
| 指标 | 说明 | 目标值 |
|---|---|---|
| TTFT | 首字响应时间 | < 2秒 |
| TPS | 每秒Token数 | > 50 tokens/s |
| 延迟 | 端到端响应时间 | < 10秒 |
| 吞吐量 | 每秒处理请求数 | 根据需求 |
| 资源利用率 | CPU/内存使用率 | < 80% |
# 性能分析示例
import time
from contextlib import contextmanager
@contextmanager
def measure_time(name):
start = time.time()
yield
elapsed = time.time() - start
print(f"{name}: {elapsed:.3f}s")
# 使用
with measure_time("模型推理"):
response = await llm.generate(prompt)
with measure_time("工具调用"):
result = await tool.execute(params)
with measure_time("上下文处理"):
context = prepare_context(messages)
同时执行多个独立任务：
import asyncio
# ❌ 顺序执行
result1 = await task1()
result2 = await task2()
result3 = await task3()
# ✅ 并行执行
results = await asyncio.gather(
task1(),
task2(),
task3()
)
缓存重复计算的结果：
from functools import lru_cache
@lru_cache(maxsize=1000)
def expensive_computation(input):
# 耗时计算
return result
# 或使用Redis缓存
async def cached_call(key, ttl=3600):
# 先查缓存
cached = await redis.get(key)
if cached:
return cached
# 缓存未命中，执行计算
result = await expensive_call()
await redis.setex(key, ttl, result)
return result
边生成边返回，提升用户体验：
# 流式响应
async def stream_response(prompt):
async for chunk in llm.stream(prompt):
yield chunk
# 立即发送给用户，不等待完整响应
根据任务复杂度选择合适的模型：
def select_model(task):
"""根据任务选择模型"""
if task.complexity == 'simple':
return 'gpt-3.5-turbo' # 快速便宜
elif task.complexity == 'medium':
return 'gpt-4-turbo' # 平衡
else:
return 'gpt-4' # 最强能力
OpenClaw内置了多种性能优化机制：
# openclaw.config.js - 性能配置
module.exports = {
performance: {
// 并发控制
concurrency: {
maxParallel: 10,
queueSize: 100
},
// 缓存
cache: {
enabled: true,
ttl: 3600,
maxSize: 10000
},
// 流式响应
streaming: {
enabled: true,
chunkSize: 50
},
// 超时设置
timeout: {
toolCall: 30000,
modelResponse: 60000
}
}
}
一个完整的性能监控和优化系统：
import time
import asyncio
from dataclasses import dataclass
from typing import Dict, List
from collections import defaultdict
@dataclass
class PerformanceMetrics:
ttft: float # 首字响应时间
tps: float # 每秒Token数
latency: float # 端到端延迟
tokens: int # Token使用量
class PerformanceMonitor:
def __init__(self):
self.metrics: Dict[str, List[float]] = defaultdict(list)
self.start_times: Dict[str, float] = {}
def start(self, operation: str):
"""开始计时"""
self.start_times[operation] = time.time()
def end(self, operation: str) -> float:
"""结束计时"""
elapsed = time.time() - self.start_times[operation]
self.metrics[operation].append(elapsed)
return elapsed
def get_stats(self, operation: str) -> Dict:
"""获取统计信息"""
times = self.metrics[operation]
if not times:
return {}
return {
'avg': sum(times) / len(times),
'min': min(times),
'max': max(times),
'p50': sorted(times)[len(times) // 2],
'p95': sorted(times)[int(len(times) * 0.95)],
'count': len(times)
}
class OptimizedAgent:
def __init__(self):
self.monitor = PerformanceMonitor()
self.cache = {}
self.semaphore = asyncio.Semaphore(10) # 并发控制
async def execute_with_optimization(self, task):
"""带优化的任务执行"""
# 1. 检查缓存
cache_key = self._get_cache_key(task)
if cache_key in self.cache:
return self.cache[cache_key]
# 2. 并发控制
async with self.semaphore:
# 3. 性能监控
self.monitor.start('task_execution')
try:
result = await self._execute(task)
# 4. 缓存结果
self.cache[cache_key] = result
return result
finally:
elapsed = self.monitor.end('task_execution')
print(f"任务执行时间: {elapsed:.3f}s")
async def _execute(self, task):
"""实际执行逻辑"""
# 并行执行独立子任务
subtasks = task.get_subtasks()
if subtasks:
results = await asyncio.gather(
*[self._execute_subtask(st) for st in subtasks]
)
return self._merge_results(results)
return await self._execute_single(task)
def get_performance_report(self) -> Dict:
"""获取性能报告"""
return {
'task_execution': self.monitor.get_stats('task_execution'),
'cache_hit_rate': self._calculate_cache_hit_rate(),
'concurrency': self.semaphore._value
}
# 使用示例
agent = OptimizedAgent()
result = await agent.execute_with_optimization(task)
report = agent.get_performance_report()
print(f"性能报告: {report}")