> For the complete documentation index, see [llms.txt](https://yeasy.gitbook.io/harness_engineering_guide/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://yeasy.gitbook.io/harness_engineering_guide/di-san-bu-fen-xi-tong-ji-cheng-yu-gong-cheng-shi-jian/10_production/10.3_performance.md). # 10.3 性能优化与成本控制 本节介绍 AI Agent 系统在生产环境中的性能和成本优化策略，包括 Token 效率优化（Schema 缓存、提示词压缩）、延迟优化（流式响应、并发执行、多级缓存）、成本控制（多模型路由、模型选择、预算监控）以及性能基准测试框架。通过这些实践，可以实现显著的成本降低和性能提升。 ## 10.3.1 概述 在生产环境中，AI Agent 系统的性能（延迟、吞吐量）和成本（API 调用费用、计算资源）是关键指标。本节介绍 Token 效率优化、延迟优化和成本控制的工程实践。 ## 10.3.2 Token 效率优化 **Schema 缓存：** Schema 是工具参数的 JSON Schema 定义。在每次工具调用前，模型需要理解 Schema 来生成正确的参数。 **问题：** 相同的 Schema 在多个请求中重复发送，造成 Token 浪费。 **解决方案：** Schema 缓存机制 ```mermaid flowchart LR R1["Request 1"] -->|"Tool A Schema
Tool B Schema"| Cache["Schema Cache
Hash(Schema) → 占位符"] R2["Request 2"] -->|"Tool A Schema
Tool B Schema"| Cache Cache --> Send["发送占位符
而非完整 Schema"] style Cache fill:#fff3e0,stroke:#ffb74d style Send fill:#e8f5e9,stroke:#388e3c ``` 图 10-4：Schema 缓存流程 ```python # core/schema_cache.py import hashlib import json from typing import Dict, Any from functools import lru_cache class SchemaCache: """Schema 哈希与缓存管理""" def __init__(self, max_schemas: int = 1000): self.cache: Dict[str, str] = {} self.max_schemas = max_schemas @staticmethod def schema_hash(schema: dict) -> str: """计算 Schema 哈希值""" schema_json = json.dumps(schema, sort_keys=True) return hashlib.sha256(schema_json.encode()).hexdigest()[:16] def get_schema_ref(self, schema: dict) -> str: """ 获取 Schema 引用(哈希值或占位符) Returns: 对于新 Schema,返回完整定义 对于缓存的 Schema,返回占位符 """ schema_hash = self.schema_hash(schema) if schema_hash in self.cache: # 返回占位符,模型理解 return f"" # 新 Schema,存入缓存并返回定义 self.cache[schema_hash] = json.dumps(schema) return json.dumps({ "hash": schema_hash, "schema": schema }) def get_full_schema(self, schema_ref: str) -> dict: """恢复完整的 Schema 定义""" if schema_ref.startswith("") return json.loads(self.cache[schema_hash]) else: return json.loads(schema_ref) def get_cache_stats(self) -> dict: """获取缓存统计""" return { 'cached_schemas': len(self.cache), 'cache_size_kb': sum( len(v.encode()) for v in self.cache.values() ) / 1024 } class TokenEfficientPromptBuilder: """Token高效的提示词构建器""" def __init__(self, schema_cache: SchemaCache): self.schema_cache = schema_cache def build_tools_section(self, tools: list[dict]) -> str: """构建工具描述,使用 Schema 缓存""" tools_desc = [] for tool in tools: schema_ref = self.schema_cache.get_schema_ref(tool['schema']) tool_desc = f""" Tool: {tool['name']} Description: {tool['description']} Parameters: {schema_ref} """ tools_desc.append(tool_desc) return "\n".join(tools_desc) def estimate_token_savings( self, tools: list[dict], num_requests: int ) -> dict: """估算 Schema 缓存的节省""" # 使用近似估算:1 个英文 token ≈ 4 字符,1 个中文字符 ≈ 1.3 token # 生产环境推荐使用 Anthropic 的 messages.count_tokens() API full_section = self.build_tools_section(tools) full_tokens = int(len(full_section) * 0.4) # 使用缓存后的Token数(估算) # 假设占位符比完整 Schema 小 50-70%,节省 30-50% cached_tokens = full_tokens * 0.5 total_savings = (full_tokens - cached_tokens) * num_requests return { 'tokens_per_request_without_cache': full_tokens, 'tokens_per_request_with_cache': cached_tokens, 'estimated_total_savings': int(total_savings), 'estimated_cost_savings_usd': total_savings * 0.003 / 1000 # $3/MTok } ``` **提示词前缀缓存：** 除了应用层的 Schema 缓存，LLM 推理层面还有一种更底层的优化——提示词前缀缓存(Prompt Prefix Caching)。当多次请求共享相同的提示词前缀时，LLM 可以复用已计算的 KV 缓存，跳过重复的前向传播。 | 缓存层次 | 作用域 | 命中条件 | 典型收益 | | --------- | ------- | ---------- | ---------------------- | | Schema 缓存 | 应用层 | 工具定义哈希匹配 | 减少重复 Token 发送 | | 提示词前缀缓存 | LLM 推理层 | 提示词前缀按字节一致 | 命中读取按缓存部分基础输入价的 10% 计费 | | 结果缓存 | 应用层 | 请求内容完全匹配 | 避免重复 API 调用 | 前缀缓存要求提示词的前缀部分（通常是系统提示词中的静态模块）在多轮对话中保持完全稳定。详细的缓存边界设计和实现方法参见 10.1.3 节。 **提示词压缩的实现方式：** ````python class PromptCompressor: """提示词压缩技术""" def __init__(self): self.compression_techniques = { 'abbreviate_examples': self.abbreviate_examples, 'remove_redundant_context': self.remove_redundant_context, 'use_bullet_points': self.use_bullet_points, 'inline_related_concepts': self.inline_related_concepts } def compress(self, prompt: str, target_reduction: float = 0.3) -> str: """ 压缩提示词,目标降低指定百分比的Token数 Args: prompt: 原始提示词 target_reduction: 目标压缩比例(0.3 = 压缩 30%) """ # 近似估算 token 数(生产环境建议用 Anthropic count_tokens API) original_tokens = int(len(prompt) * 0.4) target_tokens = int(original_tokens * (1 - target_reduction)) compressed = prompt for technique in self.compression_techniques.values(): compressed = technique(compressed) current_tokens = int(len(compressed) * 0.4) if current_tokens <= target_tokens: break return compressed def abbreviate_examples(self, text: str) -> str: """缩减示例代码""" # 将多行示例替换为简短说明 return text.replace( "Example:\n```python\n...\n```", "Example: [see docs]" ) def remove_redundant_context(self, text: str) -> str: """移除冗余上下文""" lines = text.split('\n') # 移除连续空行 return '\n'.join( line for i, line in enumerate(lines) if not (i > 0 and lines[i-1].strip() == '' and line.strip() == '') ) def use_bullet_points(self, text: str) -> str: """转换为要点格式""" return text.replace("The system supports:", "Supports:\n• ") def inline_related_concepts(self, text: str) -> str: """内联相关概念""" return text.replace( "See the Advanced Concepts section", "[advanced:...]" ) ```` ## 10.3.3 延迟优化 **流式响应：** 流式响应使得响应可以逐块传输给用户，减少首字节延迟。实现方式如下： ```python # core/streaming.py import asyncio from typing import AsyncIterator, Any class StreamingResponseHandler: """流式响应处理""" async def stream_completion( self, client, prompt: str, system: str = None, model: str = "claude-sonnet-4-6" ) -> AsyncIterator[str]: """ 流式获取完成结果,降低首字节延迟 """ with client.messages.stream( model=model, max_tokens=2048, system=system, messages=[{"role": "user", "content": prompt}] ) as stream: for text in stream.text_stream: yield text # 允许其他任务在 I/O 等待时执行 await asyncio.sleep(0) async def process_stream_with_backpressure( self, stream: AsyncIterator[str], processor: callable, buffer_size: int = 10 ) -> AsyncIterator[str]: """ 带背压的流处理,避免内存溢出 """ buffer = [] async for chunk in stream: buffer.append(chunk) if len(buffer) >= buffer_size: # 处理缓冲区 result = await processor("".join(buffer)) yield result buffer.clear() # 处理剩余数据 if buffer: result = await processor("".join(buffer)) yield result ``` ### 并发工具调用 并发工具调用的实现方式如下： ```python class ConcurrentToolExecutor: """并发执行多个工具调用""" async def execute_parallel( self, tools: list[dict], timeout: int = 30 ) -> list[Any]: """ 并发执行工具,利用 Python 异步特性 Example: tools = [ {'id': 'search_web', 'params': {'query': '...'}}, {'id': 'fetch_docs', 'params': {'url': '...'}}, {'id': 'analyze_sentiment', 'params': {'text': '...'}} ] results = await executor.execute_parallel(tools, timeout=10) """ tasks = [] for tool in tools: task = asyncio.create_task( self.execute_with_timeout( tool['id'], tool['params'], timeout ) ) tasks.append(task) results = await asyncio.gather(*tasks, return_exceptions=True) # 分离成功和失败结果 return [ {'status': 'success', 'result': r} if not isinstance(r, Exception) else {'status': 'error', 'error': str(r)} for r in results ] async def execute_with_timeout( self, tool_id: str, params: dict, timeout: int ) -> Any: """执行单个工具,带超时""" try: return await asyncio.wait_for( self.call_tool(tool_id, params), timeout=timeout ) except asyncio.TimeoutError: raise TimeoutError(f"Tool {tool_id} exceeded {timeout}s timeout") async def call_tool(self, tool_id: str, params: dict) -> Any: """实际的工具调用""" # 调用对应的工具实现 pass ``` ### 缓存层 多级缓存的实现方式如下： ```python # core/cache.py import hashlib import json from datetime import datetime, timedelta from typing import Any, Optional class MultiLevelCache: """多级缓存:内存 → Redis → 数据库""" def __init__(self, redis_client=None): self.memory_cache = {} # L1:内存缓存 self.redis_client = redis_client # L2:Redis 缓存 self.ttl_map = {} # 缓存过期时间 def get_cache_key(self, tool_id: str, params: dict) -> str: """生成缓存 Key""" params_json = json.dumps(params, sort_keys=True) params_hash = hashlib.md5(params_json.encode()).hexdigest() return f"tool_result:{tool_id}:{params_hash}" async def get(self, tool_id: str, params: dict) -> Optional[Any]: """获取缓存结果""" key = self.get_cache_key(tool_id, params) # L1:检查内存缓存 if key in self.memory_cache: cached_item = self.memory_cache[key] if self.is_valid(key): return cached_item['value'] else: del self.memory_cache[key] # L2:检查 Redis(如果可用) if self.redis_client: try: cached_value = self.redis_client.get(key) if cached_value: result = json.loads(cached_value) # 回写到内存缓存 self.memory_cache[key] = { 'value': result, 'timestamp': datetime.now() } return result except Exception: pass # Redis 不可用,继续 return None async def set( self, tool_id: str, params: dict, result: Any, ttl_seconds: int = 3600 ): """缓存结果""" key = self.get_cache_key(tool_id, params) # L1:写入内存缓存 self.memory_cache[key] = { 'value': result, 'timestamp': datetime.now() } self.ttl_map[key] = datetime.now() + timedelta(seconds=ttl_seconds) # L2:写入 Redis if self.redis_client: try: self.redis_client.setex( key, ttl_seconds, json.dumps(result) ) except Exception: pass # Redis 写入失败,继续 def is_valid(self, key: str) -> bool: """检查缓存是否仍有效""" if key not in self.ttl_map: return False return datetime.now() < self.ttl_map[key] def get_stats(self) -> dict: """获取缓存统计""" return { 'memory_items': len(self.memory_cache), 'valid_items': sum(1 for k in self.memory_cache if self.is_valid(k)), 'memory_size_mb': sum( len(json.dumps(v['value']).encode()) for v in self.memory_cache.values() ) / 1024 / 1024 } ``` 缓存策略有效地提升了系统的响应速度，但系统运维的另一个重要维度是成本控制。在大规模部署中，模型调用成本往往是系统运营的主要开支。本节介绍了如何通过智能路由和优化策略来降低系统成本。 ## 10.3.4 成本控制 ### 多模型路由 2026 年的成本优化实践已发展出更成熟的模式。核心思路是：**并非所有调用都需要最强（也最贵）的模型**。 多模型路由(Multi-Model Routing)按任务复杂度将请求路由到不同模型： ```python class MultiModelRouter: """按任务复杂度路由到不同模型""" def __init__(self): self.models = { "simple": {"model": "haiku", "cost_per_mtok": 1.0}, # Haiku 4.5: $1/$5 "moderate": {"model": "sonnet", "cost_per_mtok": 3.0}, # Sonnet 4.6: $3/$15 "complex": {"model": "opus", "cost_per_mtok": 5.0}, # Opus 4.7: $5/$25 } def route(self, task: str, context_length: int) -> str: """根据任务特征选择模型""" if self._is_simple_task(task): return "simple" # 简单格式化、分类、提取 elif context_length > 100_000 or self._needs_deep_reasoning(task): return "complex" # 复杂推理、长上下文分析 else: return "moderate" # 大多数常规任务 ``` 结合提示词前缀缓存（参见 10.1.3 节）和三级预算管控（per-request → per-task → per-day/month，在 50%/80% 阈值触发告警），综合优化可达 **60-80% 的成本降低**。 ### 模型选择与降级 模型选择与降级的实现方式如下： ```python from datetime import datetime class ModelSelector: """根据任务复杂度选择成本最优的模型""" # 模型成本映射(USD per 1M tokens) MODEL_COSTS = { 'claude-haiku-4-5': {'input': 1.00, 'output': 5.00}, 'claude-sonnet-4-6': {'input': 3.00, 'output': 15.00}, 'claude-opus-4-7': {'input': 5.00, 'output': 25.00}, } # 模型能力评级 MODEL_CAPABILITIES = { 'claude-haiku-4-5': 2, # 基础任务 'claude-sonnet-4-6': 3, # 中等复杂 'claude-opus-4-7': 4, # 复杂推理 } def select_model( self, task_complexity: int, budget_per_request: float = 0.10 ) -> str: """ 选择合适的模型 Args: task_complexity: 1-4,任务复杂度 budget_per_request: 每个请求的成本预算(USD) """ candidates = [] for model, capabilities in self.MODEL_CAPABILITIES.items(): # 能力充分 if capabilities >= task_complexity: cost = self.estimate_cost(model) # 按默认的 1000 入/500 出估算 if cost <= budget_per_request: candidates.append({ 'model': model, 'cost': cost, 'capability_margin': capabilities - task_complexity }) if not candidates: # 没有符合预算的模型 return None # 选择成本最低但能力充分的模型 return min(candidates, key=lambda x: x['cost'])['model'] def estimate_cost( self, model: str, input_tokens: int = 1000, output_tokens: int = 500 ) -> float: """估算请求成本""" costs = self.MODEL_COSTS[model] return ( input_tokens * costs['input'] + output_tokens * costs['output'] ) / 1_000_000 class CostMonitor: """成本监控与告警""" def __init__(self, daily_budget_usd: float): self.daily_budget = daily_budget_usd self.daily_cost = 0.0 self.request_costs = [] def log_request(self, model: str, input_tokens: int, output_tokens: int): """记录请求成本""" from core.model_selector import ModelSelector selector = ModelSelector() cost = selector.estimate_cost(model, input_tokens, output_tokens) self.daily_cost += cost self.request_costs.append({ 'model': model, 'cost': cost, 'input_tokens': input_tokens, 'output_tokens': output_tokens, 'timestamp': datetime.now() }) def check_budget(self) -> dict: """检查预算状态""" remaining = self.daily_budget - self.daily_cost usage_pct = (self.daily_cost / self.daily_budget) * 100 return { 'budget': self.daily_budget, 'spent': self.daily_cost, 'remaining': remaining, 'usage_percent': usage_pct, 'status': 'ok' if usage_pct < 80 else 'warning' if usage_pct < 95 else 'critical' } def get_cost_by_model(self) -> dict: """按模型统计成本""" costs_by_model = {} for request in self.request_costs: model = request['model'] if model not in costs_by_model: costs_by_model[model] = {'count': 0, 'cost': 0.0} costs_by_model[model]['count'] += 1 costs_by_model[model]['cost'] += request['cost'] return costs_by_model ``` ### Multi-Model Routing 与 Prompt 缓存 **多模型路由** 是 2026 年业界标准的成本优化实践。核心原则是：**并非所有请求都需要最强模型**。按任务复杂度智能路由可节省 40-60% 成本。 ```python class RoutingStrategy: """任务复杂度路由策略""" def __init__(self): self.complexity_thresholds = { 'simple': ['分类', '格式化', '提取', '总结'], # Haiku 'moderate': ['编辑', '改写', '分析'], # Sonnet 'complex': ['推理', '规划', '多步骤', '创意'] # Opus } def classify_task(self, task_description: str) -> str: """分类任务复杂度""" task_lower = task_description.lower() for level, keywords in self.complexity_thresholds.items(): if any(kw in task_lower for kw in keywords): return level return 'moderate' def route(self, task: str) -> str: """根据复杂度路由到模型""" level = self.classify_task(task) models = { 'simple': 'claude-haiku-4-5', # $1/$5 per MTok 'moderate': 'claude-sonnet-4-6', # $3/$15 per MTok 'complex': 'claude-opus-4-7' # $5/$25 per MTok } return models[level] ``` **提示词缓存** 的成本收益来自缓存命中读取：Anthropic 对 cache read tokens 按基础输入 token 价格的 10% 计费（Fable 5.1 与 Mythos 5.1 为 2.5%）；首次写入缓存不是免费，5 分钟 TTL 写入按 1.25x 基础输入价、1 小时 TTL 写入按 2x 计费。长系统提示（Agent 系统常见）是否能节省 20-30% 账单，取决于静态前缀大小、命中率和写入刷新频率。缓存预热策略：预先加载常见系统提示变体。 | 预算级别 | 范围 | 告警阈值 | 示例 | | ------- | ---- | -------------- | ------------------ | | Level 1 | 单个请求 | max\_tokens 限制 | per-request: $0.10 | | Level 2 | 单个任务 | 多个 LLM 调用聚合 | per-task: $0.50 | | Level 3 | 日/月 | 50% 和 80% 告警 | daily: $100 | 模型路由通常可贡献 **40-60%** 的成本降低；叠加提示词前缀缓存、上下文裁剪和预算管控后，组合优化可达到 **60-80%**。上下文管理占总支出 60-70%，关键在于仅包含相关上下文，避免冗余数据。 ### 缓存命中率优化 缓存命中率优化的实现方式如下： ```python import hashlib import json from datetime import datetime class CacheHitAnalyzer: """分析与优化缓存命中率""" def __init__(self): self.cache_accesses = [] def log_access( self, tool_id: str, params: dict, hit: bool, response_time_ms: float ): """记录缓存访问""" self.cache_accesses.append({ 'tool_id': tool_id, 'params_hash': hashlib.md5( json.dumps(params, sort_keys=True).encode() ).hexdigest(), 'hit': hit, 'response_time_ms': response_time_ms, 'timestamp': datetime.now() }) def get_hit_rate(self) -> float: """计算缓存命中率""" if not self.cache_accesses: return 0.0 hits = sum(1 for a in self.cache_accesses if a['hit']) return hits / len(self.cache_accesses) def analyze_by_tool(self) -> dict: """按工具统计命中率""" by_tool = {} for access in self.cache_accesses: tool_id = access['tool_id'] if tool_id not in by_tool: by_tool[tool_id] = {'hits': 0, 'misses': 0, 'avg_response_ms': 0} if access['hit']: by_tool[tool_id]['hits'] += 1 else: by_tool[tool_id]['misses'] += 1 # 计算平均响应时间 for tool_id in by_tool: tool_accesses = [ a for a in self.cache_accesses if a['tool_id'] == tool_id ] avg_response = sum( a['response_time_ms'] for a in tool_accesses ) / len(tool_accesses) by_tool[tool_id]['avg_response_ms'] = avg_response by_tool[tool_id]['hit_rate'] = ( by_tool[tool_id]['hits'] / (by_tool[tool_id]['hits'] + by_tool[tool_id]['misses']) ) return by_tool def recommend_cache_ttl(self, tool_id: str) -> int: """根据访问模式推荐缓存 TTL""" tool_accesses = [ a for a in self.cache_accesses if a['tool_id'] == tool_id ] if not tool_accesses: return 3600 # 默认 1 小时 # 简单启发式:高命中率工具使用更长的 TTL hit_rate = sum(1 for a in tool_accesses if a['hit']) / len(tool_accesses) if hit_rate > 0.7: return 86400 # 24 小时 elif hit_rate > 0.4: return 3600 # 1 小时 else: return 600 # 10 分钟 ``` ### 价值计费：为成功任务而非词元买单 多模型路由把请求引向更便宜的模型以压低单位 token 成本，但更低的单位成本常常是一种错觉。较弱的模型单步成功率也更低，一旦把失败、重试乃至人工兜底的代价折算进来，把任务交给弱模型反而可能更贵。真正应当优化的目标不是每 token 成本，而是**每成功任务成本**： C\_success ≈ C\_token × E\[尝试次数] / 成功率 公式里的 E\[尝试次数] 与成功率都随模型能力和 Harness 的验证-重试闭环而变，本书已有的两处机制正好落在它上面： * **压低分母侧的失败**：第 1.4.5 节的定量论证表明，结果验证加最多 3 次重试可把单步有效成功率从 p 提升到 p' = 1 − (1−p)^4，用更少的尝试摊薄每成功任务成本——省钱的杠杆常在 Harness，而非更贵的模型。 * **用正确的指标度量**：成本监控应以第 13.4 节 ProductionMetrics 的 cost\_per\_successful\_task 字段为北极星，而非孤立地看 total\_cost\_usd 或平均 token 消耗。 这一转向也与政策导向一致：《北京市加快智能体引领发展若干措施》（京发改〔2026〕1185 号）第六条在发展 Token（词元）经济时，明确“鼓励创新主体从 Token 消耗量计费转向价值计费”（见附录参考文献）。 ## 10.3.5 实战：性能基准测试 性能基准测试的实现代码如下： ```python # benchmarks/performance_benchmark.py import asyncio import time from typing import Callable class PerformanceBenchmark: """性能基准测试框架""" def __init__(self, name: str): self.name = name self.results = [] async def run_test( self, test_fn: Callable, iterations: int = 100, warmup: int = 10 ) -> dict: """运行性能测试""" # 预热 for _ in range(warmup): await test_fn() # 实际测试 latencies = [] start_time = time.time() for _ in range(iterations): iter_start = time.time() await test_fn() latencies.append((time.time() - iter_start) * 1000) # 毫秒 total_time = time.time() - start_time results = { 'test_name': self.name, 'iterations': iterations, 'total_time_s': total_time, 'throughput_per_sec': iterations / total_time, 'p50_latency_ms': sorted(latencies)[len(latencies) // 2], 'p99_latency_ms': sorted(latencies)[int(len(latencies) * 0.99)], 'avg_latency_ms': sum(latencies) / len(latencies), 'max_latency_ms': max(latencies), 'min_latency_ms': min(latencies) } self.results.append(results) return results def print_report(self): """打印测试报告""" print(f"\n{'Performance Benchmark Report':^60}") print("=" * 60) for result in self.results: print(f"\nTest: {result['test_name']}") print(f" Iterations: {result['iterations']}") print(f" Throughput: {result['throughput_per_sec']:.2f} req/s") print(f" P50 Latency: {result['p50_latency_ms']:.2f} ms") print(f" P99 Latency: {result['p99_latency_ms']:.2f} ms") print(f" Avg Latency: {result['avg_latency_ms']:.2f} ms") ``` ## 10.3.6 总结 性能优化与成本控制的关键策略： | 维度 | 优化方向 | 预期收益 | | -------- | --------------- | --------------------------------------------------------- | | Token 效率 | Schema 缓存、提示词压缩 | Schema token 30-50% 削减；总体按 30-40% 保守估算 | | 延迟 | 流式处理、并发执行、缓存 | 端到端 Agent 请求 P99 参见第 11 章目标（<10 秒）；非 LLM 内部路径可按 <500ms 设计 | | 成本 | 模型路由、缓存命中率、预算管控 | 路由单项 40-60%；组合优化 60-80% | 下一节将介绍配置管理与特性门控，实现灰度发布和 A/B 测试。