上下文膨胀定律：为什么你的 Agent 比想象中贵 10 倍

class SemanticCache:
    def __init__(self, threshold=0.92):
        self.threshold = threshold
        self.vector_store = VectorStore()
        self.cache = {}
    
    def get(self, query: str) -> Optional[str]:
        """语义检索缓存"""
        query_embedding = embed(query)
        results = self.vector_store.search(query_embedding, top_k=1)
        
        if results and results[0].score >= self.threshold:
            return results[0].cached_response
        return None
    
    def set(self, query: str, response: str):
        """写入缓存"""
        embedding = embed(query)
        self.vector_store.add(embedding)
        self.cache[hash(query)] = response

class ContextCompressor:
    def __init__(self, max_tokens: int):
        self.max_tokens = max_tokens
    
    def compress(self, context: dict) -> dict:
        """智能压缩上下文"""
        compressed = {}
        
        # 知识库：只保留 top-10 相关片段
        if 'knowledge' in context:
            docs = context['knowledge']
            scored = [(d, relevance(d.query)) for d in docs]
            top_docs = sorted(scored, key=lambda x: x[1], reverse=True)[:10]
            compressed['knowledge'] = [d for d, _ in top_docs]
        
        # 对话历史：只保留关键轮次
        if 'history' in context:
            compressed['history'] = self.extract_key_turns(context['history'])
        
        return compressed
    
    def extract_key_turns(self, history: list) -> list:
        """提取关键对话轮次"""
        key_turns = []
        for turn in history:
            if self.is_key_turn(turn):
                key_turns.append(turn)
        return self.truncate_to_token_limit(key_turns)

def route_request(query: str, user_segment: str) -> str:
    """智能路由"""
    complexity = estimate_complexity(query)
    
    # 高价值用户：始终用最强模型
    if user_segment == 'enterprise':
        return 'claude-3-5-sonnet'
    
    # 低复杂度：便宜模型
    if complexity < 0.3:
        return 'gpt-3.5-turbo'
    
    # 高复杂度：强模型
    elif complexity > 0.7:
        return 'claude-3-opus'
    
    # 中等复杂度：平衡选择
    else:
        return 'claude-3-sonnet'

class BatchProcessor:
    """低价值请求批量处理"""
    
    def __init__(self, batch_size=10, max_wait_seconds=30):
        self.batch_size = batch_size
        self.queue = []
        self.max_wait = max_wait_seconds
    
    def add(self, request: dict) -> Future:
        """加入批处理队列"""
        future = Future()
        self.queue.append((request, future))
        
        if len(self.queue) >= self.batch_size:
            self.process_batch()
        
        return future
    
    def process_batch(self):
        """批量执行"""
        batch = self.queue[:self.batch_size]
        results = self_llm.batch_generate([r for r, _ in batch])
        
        for (req, future), result in zip(batch, results):
            future.set_result(result)
        
        self.queue = self.queue[self.batch_size:]