┌──────────────────────────────────────────────────────────┐
│                    LLM API 工程化能力矩阵                   │
├──────────┬──────────┬──────────┬─────────────────────────┤
│  限流保护  │  语义缓存  │  故障降级  │  成本控制              │
│ 令牌桶算法 │ 精确匹配  │ 熔断机制   │  Token计数             │
│ 滑动窗口   │ 向量相似  │ 降级模型   │  路由优化              │
│ 用户级限流 │ LRU淘汰  │ 本地回退   │  缓存命中              │
├──────────┼──────────┼──────────┼─────────────────────────┤
│  重试机制  │ 可观测性  │  安全控制  │  模型路由              │
│ 指数退避  │ 请求日志  │  鉴权认证  │  智能路由              │
│ 幂等性    │ 指标监控  │  内容过滤  │  负载均衡              │
│ 超时控制  │ 链路追踪  │  数据脱敏  │  故障转移              │
└──────────┴──────────┴──────────┴─────────────────────────┘

import time
from threading import Lock
from collections import defaultdict
from typing import Dict

class TokenBucket:
    """令牌桶限流器"""

    def __init__(self, rate: float, capacity: int):
        """
        rate: 每秒生成令牌数
        capacity: 桶容量（最大突发请求数）
        """
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last_time = time.time()
        self.lock = Lock()

    def acquire(self, tokens: int = 1) -> bool:
        """获取令牌，返回是否成功"""
        with self.lock:
            now = time.time()
            # 计算新增令牌
            elapsed = now - self.last_time
            self.tokens = min(
                self.capacity,
                self.tokens + elapsed * self.rate
            )
            self.last_time = now

            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

    def get_wait_time(self, tokens: int = 1) -> float:
        """获取需要等待的时间（秒）"""
        with self.lock:
            if self.tokens >= tokens:
                return 0
            deficit = tokens - self.tokens
            return deficit / self.rate

class MultiLevelRateLimiter:
    """多级限流器"""

    def __init__(self):
        # 全局限流：1000请求/分钟
        self.global_limiter = TokenBucket(rate=1000/60, capacity=200)
        # 用户级限流
        self.user_limiters: Dict[str, TokenBucket] = {}
        self.user_default_rate = 100/60  # 100次/分钟
        self.user_default_capacity = 30
        self.lock = Lock()

    def _get_user_limiter(self, user_id: str) -> TokenBucket:
        if user_id not in self.user_limiters:
            with self.lock:
                if user_id not in self.user_limiters:
                    self.user_limiters[user_id] = TokenBucket(
                        rate=self.user_default_rate,
                        capacity=self.user_default_capacity
                    )
        return self.user_limiters[user_id]

    def check(self, user_id: str, tokens: int = 1) -> dict:
        """
        检查是否允许请求
        返回: {allowed: bool, reason: str, wait_time: float}
        """
        # 全局检查
        if not self.global_limiter.acquire(tokens):
            return {
                'allowed': False,
                'reason': 'global_rate_limit',
                'wait_time': self.global_limiter.get_wait_time(tokens)
            }

        # 用户级检查
        user_limiter = self._get_user_limiter(user_id)
        if not user_limiter.acquire(tokens):
            return {
                'allowed': False,
                'reason': 'user_rate_limit',
                'wait_time': user_limiter.get_wait_time(tokens)
            }

        return {'allowed': True, 'reason': '', 'wait_time': 0}

# 使用示例
if __name__ == "__main__":
    limiter = MultiLevelRateLimiter()

    # 模拟用户请求
    for i in range(5):
        result = limiter.check("user_001")
        status = "✅ 允许" if result['allowed'] else "❌ 限流"
        print(f"请求 {i+1}: {status} - {result['reason']}")

import hashlib
import json
import time
from typing import Optional, Tuple, List
from dataclasses import dataclass

@dataclass
class CacheEntry:
    key: str
    query: str
    response: str
    model: str
    created_at: float
    access_count: int
    embedding: List[float] = None

class LRUCache:
    """LRU缓存实现"""

    def __init__(self, max_size: int = 1000):
        self.max_size = max_size
        self.cache = {}  # key -> CacheEntry
        self.access_order = []  # 按访问顺序，最新在末尾
        self.lock = None  # 生产环境加锁

    def get(self, key: str) -> Optional[CacheEntry]:
        if key in self.cache:
            # 移到末尾（最新访问）
            self.access_order.remove(key)
            self.access_order.append(key)
            entry = self.cache[key]
            entry.access_count += 1
            return entry
        return None

    def put(self, key: str, entry: CacheEntry):
        if key in self.cache:
            self.access_order.remove(key)
        elif len(self.cache) >= self.max_size:
            # 淘汰最旧的
            oldest_key = self.access_order.pop(0)
            del self.cache[oldest_key]

        self.cache[key] = entry
        self.access_order.append(key)

    def size(self) -> int:
        return len(self.cache)

class SemanticCache:
    """语义缓存 - 结合精确匹配和向量相似匹配"""

    def __init__(self, exact_cache_size: int = 5000, 
                 semantic_threshold: float = 0.95):
        self.exact_cache = LRUCache(max_size=exact_cache_size)
        self.semantic_threshold = semantic_threshold
        # 生产环境应使用向量数据库（FAISS/Milvus/Qdrant）
        self.vector_entries = []  # 简化：内存向量列表

    def _exact_key(self, query: str, model: str) -> str:
        """生成精确匹配的key"""
        content = f"{model}:{query.strip().lower()}"
        return hashlib.md5(content.encode()).hexdigest()

    def _simple_embedding(self, text: str) -> List[float]:
        """
        简化版文本向量（生产环境应使用Embedding API）
        实际使用：openai.embeddings.create 或 text-embedding-3-small
        """
        # 这里用字符级统计模拟，实际项目请用真实embedding模型
        text = text.lower()
        vec = [0.0] * 128
        for i, char in enumerate(text):
            vec[ord(char) % 128] += 1.0
        # 归一化
        norm = sum(x*x for x in vec) ** 0.5
        if norm > 0:
            vec = [x/norm for x in vec]
        return vec

    def _cosine_similarity(self, v1: List[float], v2: List[float]) -> float:
        """余弦相似度"""
        return sum(a*b for a, b in zip(v1, v2))

    def lookup(self, query: str, model: str) -> Optional[str]:
        """
        查找缓存
        返回: 缓存的回复，或None
        """
        # 1. 先尝试精确匹配
        key = self._exact_key(query, model)
        entry = self.exact_cache.get(key)
        if entry:
            return entry.response

        # 2. 语义匹配（简化版，生产环境用向量数据库）
        query_emb = self._simple_embedding(query)
        best_score = 0
        best_response = None

        for entry in self.vector_entries:
            if entry.model != model:
                continue
            score = self._cosine_similarity(query_emb, entry.embedding)
            if score > best_score and score >= self.semantic_threshold:
                best_score = score
                best_response = entry.response

        return best_response

    def store(self, query: str, model: str, response: str):
        """存入缓存"""
        key = self._exact_key(query, model)
        embedding = self._simple_embedding(query)

        entry = CacheEntry(
            key=key,
            query=query,
            response=response,
            model=model,
            created_at=time.time(),
            access_count=1,
            embedding=embedding
        )

        self.exact_cache.put(key, entry)
        self.vector_entries.append(entry)

        # 限制向量条目数量
        if len(self.vector_entries) > 2000:
            self.vector_entries = self.vector_entries[-1500:]

# 使用示例
if __name__ == "__main__":
    cache = SemanticCache()

    # 存入缓存
    cache.store(
        query="Python如何实现快速排序？",
        model="gpt-4o",
        response="快速排序的Python实现如下：\n\ndef quick_sort(arr):\n    if len(arr) <= 1:\n        return arr\n    pivot = arr[len(arr)//2]\n    left = [x for x in arr if x < pivot]\n    middle = [x for x in arr if x == pivot]\n    right = [x for x in arr if x > pivot]\n    return quick_sort(left) + middle + quick_sort(right)"
    )

    # 精确匹配查询
    result = cache.lookup("Python如何实现快速排序？", "gpt-4o")
    print(f"精确匹配: {'命中' if result else '未命中'}")

    # 相似查询
    result2 = cache.lookup("python快速排序怎么写？", "gpt-4o")
    print(f"语义匹配: {'命中' if result2 else '未命中'}")

import time
import random
import asyncio
from typing import Callable, Any, Optional
from functools import wraps

def retry_with_backoff(
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 30.0,
    backoff_factor: float = 2.0,
    jitter: bool = True,
    retry_on_exceptions: tuple = (Exception,)
):
    """
    指数退避重试装饰器

    Args:
        max_retries: 最大重试次数
        base_delay: 初始延迟（秒）
        max_delay: 最大延迟（秒）
        backoff_factor: 退避因子
        jitter: 是否添加抖动
        retry_on_exceptions: 需要重试的异常类型
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None

            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except retry_on_exceptions as e:
                    last_exception = e
                    if attempt == max_retries:
                        raise

                    # 计算延迟
                    delay = min(base_delay * (backoff_factor ** attempt), max_delay)
                    if jitter:
                        delay = delay * (0.5 + random.random() * 0.5)

                    time.sleep(delay)

            raise last_exception

        return wrapper
    return decorator

# 异步版本
def async_retry_with_backoff(
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 30.0,
    backoff_factor: float = 2.0,
    jitter: bool = True,
    retry_on_exceptions: tuple = (Exception,)
):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            last_exception = None

            for attempt in range(max_retries + 1):
                try:
                    return await func(*args, **kwargs)
                except retry_on_exceptions as e:
                    last_exception = e
                    if attempt == max_retries:
                        raise

                    delay = min(base_delay * (backoff_factor ** attempt), max_delay)
                    if jitter:
                        delay = delay * (0.5 + random.random() * 0.5)

                    await asyncio.sleep(delay)

            raise last_exception

        return wrapper
    return decorator

import time
from enum import Enum
from threading import Lock

class CircuitState(Enum):
    CLOSED = "closed"       # 正常状态
    OPEN = "open"           # 熔断状态
    HALF_OPEN = "half_open" # 半开状态（探测恢复）

class CircuitBreaker:
    """熔断器实现"""

    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 30.0,
        half_open_max_calls: int = 3,
        failure_window: float = 60.0
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max_calls = half_open_max_calls
        self.failure_window = failure_window

        self.state = CircuitState.CLOSED
        self.failure_times = []
        self.half_open_calls = 0
        self.last_failure_time = 0
        self.lock = Lock()

    def _is_window_expired(self) -> bool:
        """检查熔断窗口是否过期"""
        return time.time() - self.last_failure_time > self.recovery_timeout

    def _record_failure(self):
        """记录失败"""
        now = time.time()
        self.failure_times.append(now)
        self.last_failure_time = now

        # 清理窗口外的失败记录
        self.failure_times = [
            t for t in self.failure_times 
            if now - t < self.failure_window
        ]

    def can_execute(self) -> bool:
        """判断是否可以执行请求"""
        with self.lock:
            if self.state == CircuitState.CLOSED:
                return True

            if self.state == CircuitState.OPEN:
                if self._is_window_expired():
                    # 进入半开状态
                    self.state = CircuitState.HALF_OPEN
                    self.half_open_calls = 0
                    return True
                return False

            if self.state == CircuitState.HALF_OPEN:
                return self.half_open_calls < self.half_open_max_calls

            return False

    def record_success(self):
        """记录成功"""
        with self.lock:
            if self.state == CircuitState.HALF_OPEN:
                # 半开状态下成功，恢复正常
                self.state = CircuitState.CLOSED
                self.failure_times.clear()
                self.half_open_calls = 0
            elif self.state == CircuitState.CLOSED:
                # 正常状态下成功，可以适当清理旧的失败记录
                pass

    def record_failure(self):
        """记录失败"""
        with self.lock:
            self._record_failure()

            if self.state == CircuitState.HALF_OPEN:
                # 半开状态下失败，重新熔断
                self.state = CircuitState.OPEN
                self.half_open_calls = 0
            elif self.state == CircuitState.CLOSED:
                # 检查是否达到熔断阈值
                if len(self.failure_times) >= self.failure_threshold:
                    self.state = CircuitState.OPEN

    def get_state(self) -> str:
        return self.state.value

# 使用示例
if __name__ == "__main__":
    breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=5.0)

    print("初始状态:", breaker.get_state())

    # 模拟连续失败
    for i in range(4):
        if breaker.can_execute():
            print(f"请求 {i+1}: 执行 -> 失败")
            breaker.record_failure()
        else:
            print(f"请求 {i+1}: 被熔断")

    print("当前状态:", breaker.get_state())
    print(f"等待 {breaker.recovery_timeout} 秒后恢复...")

import tiktoken  # pip install tiktoken
from typing import Dict, List, Optional
from dataclasses import dataclass
from datetime import datetime

@dataclass
class CostRecord:
    request_id: str
    user_id: str
    model: str
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    cost: float
    timestamp: float

class CostManager:
    """LLM成本管理器"""

    # 2026年7月主流模型定价（美元/1K tokens）
    MODEL_PRICING = {
        'gpt-4o': {'input': 0.005, 'output': 0.015},
        'gpt-4o-mini': {'input': 0.00015, 'output': 0.0006},
        'claude-3-5-sonnet': {'input': 0.003, 'output': 0.015},
        'claude-3-opus': {'input': 0.015, 'output': 0.075},
        'deepseek-v3': {'input': 0.00014, 'output': 0.00028},
        'qwen2-72b': {'input': 0.0008, 'output': 0.002},
    }

    def __init__(self):
        self.encoders = {}  # 缓存encoder
        self.records: List[CostRecord] = []
        self.user_daily_cost: Dict[str, float] = {}
        self.total_cost = 0.0

    def _get_encoder(self, model: str):
        """获取token编码器"""
        if model not in self.encoders:
            try:
                self.encoders[model] = tiktoken.encoding_for_model(model)
            except:
                # 默认使用cl100k_base
                self.encoders[model] = tiktoken.get_encoding("cl100k_base")
        return self.encoders[model]

    def count_tokens(self, text: str, model: str = 'gpt-4o') -> int:
        """计算文本的token数量"""
        encoder = self._get_encoder(model)
        return len(encoder.encode(text))

    def count_messages_tokens(self, messages: List[Dict], 
                              model: str = 'gpt-4o') -> int:
        """计算对话消息的token数量"""
        encoder = _get_encoder(model) if False else None  # 简化
        # 使用tiktoken的计算方式
        try:
            enc = tiktoken.encoding_for_model(model)
        except:
            enc = tiktoken.get_encoding("cl100k_base")

        tokens_per_message = 4  # 每条消息的基础开销
        num_tokens = 0

        for message in messages:
            num_tokens += tokens_per_message
            for key, value in message.items():
                num_tokens += len(enc.encode(str(value)))

        num_tokens += 3  # 回复的基础开销
        return num_tokens

    def calculate_cost(self, prompt_tokens: int, 
                       completion_tokens: int, model: str) -> float:
        """计算请求成本"""
        pricing = self.MODEL_PRICING.get(model, 
                                        {'input': 0.01, 'output': 0.03})
        cost = (
            prompt_tokens / 1000 * pricing['input'] +
            completion_tokens / 1000 * pricing['output']
        )
        return round(cost, 6)

    def record_cost(self, request_id: str, user_id: str, model: str,
                    prompt_tokens: int, completion_tokens: int):
        """记录成本"""
        cost = self.calculate_cost(prompt_tokens, completion_tokens, model)
        record = CostRecord(
            request_id=request_id,
            user_id=user_id,
            model=model,
            prompt_tokens=prompt_tokens,
            completion_tokens=completion_tokens,
            total_tokens=prompt_tokens + completion_tokens,
            cost=cost,
            timestamp=datetime.now().timestamp()
        )
        self.records.append(record)
        self.total_cost += cost

        # 用户日成本统计
        today = datetime.now().strftime('%Y-%m-%d')
        key = f"{user_id}:{today}"
        self.user_daily_cost[key] = self.user_daily_cost.get(key, 0) + cost

        return cost

    def get_user_daily_cost(self, user_id: str) -> float:
        """获取用户今日成本"""
        today = datetime.now().strftime('%Y-%m-%d')
        key = f"{user_id}:{today}"
        return self.user_daily_cost.get(key, 0)

    def get_cost_summary(self) -> Dict:
        """获取成本统计摘要"""
        model_costs = {}
        user_costs = {}

        for record in self.records:
            model_costs[record.model] = model_costs.get(record.model, 0) + record.cost
            user_costs[record.user_id] = user_costs.get(record.user_id, 0) + record.cost

        return {
            'total_cost': round(self.total_cost, 4),
            'total_requests': len(self.records),
            'total_tokens': sum(r.total_tokens for r in self.records),
            'avg_cost_per_request': round(
                self.total_cost / len(self.records), 6
            ) if self.records else 0,
            'by_model': {k: round(v, 4) for k, v in model_costs.items()},
            'top_users': sorted(
                user_costs.items(), key=lambda x: x[1], reverse=True
            )[:10]
        }

from typing import Dict, Tuple, List
import re

class ModelRouter:
    """智能模型路由器"""

    def __init__(self):
        self.models = {
            'fast': {
                'name': 'gpt-4o-mini',
                'cost_per_1k_input': 0.00015,
                'speed': 'fast',
                'quality': 'good'
            },
            'balanced': {
                'name': 'gpt-4o',
                'cost_per_1k_input': 0.005,
                'speed': 'medium',
                'quality': 'excellent'
            },
            'premium': {
                'name': 'claude-3-opus',
                'cost_per_1k_input': 0.015,
                'speed': 'slow',
                'quality': 'best'
            }
        }

    def _assess_complexity(self, query: str, context: str = '') -> str:
        """
        评估任务复杂度
        返回: simple, medium, complex
        """
        text = query + ' ' + context
        score = 0

        # 长度指标
        if len(text) > 2000:
            score += 2
        elif len(text) > 500:
            score += 1

        # 关键词指标
        complex_keywords = [
            '分析', '优化', '设计', '架构', '策略', '方案',
            '对比', '评估', '预测', '推理', '代码审查',
            'debug', '优化', '重构', '复杂', '系统'
        ]
        for kw in complex_keywords:
            if kw in text:
                score += 0.5

        # 代码相关
        if re.search(r'```|def |function |class ', text):
            score += 1

        # 数学/逻辑
        if re.search(r'数学|证明|推导|公式|算法', text):
            score += 1

        if score >= 3:
            return 'complex'
        elif score >= 1:
            return 'medium'
        else:
            return 'simple'

    def route(self, query: str, context: str = '',
              preference: str = 'balanced') -> Tuple[str, Dict]:
        """
        选择合适的模型
        返回: (模型名, 路由信息)
        """
        complexity = self._assess_complexity(query, context)

        # 基于复杂度选择
        model_tier = {
            'simple': 'fast',
            'medium': 'balanced', 
            'complex': 'premium'
        }[complexity]

        # 用户偏好调整
        if preference == 'cost':
            if complexity == 'complex':
                model_tier = 'balanced'  # 复杂任务也用平衡模型
            else:
                model_tier = 'fast'
        elif preference == 'quality':
            model_tier = 'premium'

        model = self.models[model_tier]

        return model['name'], {
            'complexity': complexity,
            'model_tier': model_tier,
            'preference': preference,
            'estimated_cost': model['cost_per_1k_input']
        }

# 使用示例
if __name__ == "__main__":
    router = ModelRouter()

    test_queries = [
        ("今天天气怎么样？", "simple"),
        ("帮我写一个Python爬虫", "medium"),
        ("设计一个高并发的微服务架构，需要考虑缓存、消息队列、数据库分片等", "complex"),
    ]

    for query, expected in test_queries:
        model, info = router.route(query)
        print(f"[{expected}] {query[:30]}... -> {model} (复杂度: {info['complexity']})")

import time
import uuid
from typing import Dict, Any, Optional
from contextlib import contextmanager
from dataclasses import dataclass, field

@dataclass
class LLMRequestMetrics:
    request_id: str
    model: str
    user_id: str
    prompt_tokens: int = 0
    completion_tokens: int = 0
    latency_ms: float = 0
    status: str = "success"  # success, error, rate_limited, cached
    error_type: str = ""
    cache_hit: bool = False
    cost: float = 0.0
    metadata: Dict = field(default_factory=dict)

class LLMObserver:
    """LLM可观测性管理器"""

    def __init__(self):
        self.metrics_history = []
        self.error_count = 0
        self.total_requests = 0
        self.cached_requests = 0
        self.total_latency = 0.0

    @contextmanager
    def trace(self, model: str, user_id: str, 
              metadata: Dict = None):
        """
        请求追踪上下文管理器
        使用方式:
        with observer.trace(model, user_id) as metrics:
            response = call_llm(...)
            metrics.prompt_tokens = ...
        """
        request_id = str(uuid.uuid4())
        metrics = LLMRequestMetrics(
            request_id=request_id,
            model=model,
            user_id=user_id,
            metadata=metadata or {}
        )
        start_time = time.time()

        try:
            yield metrics
        except Exception as e:
            metrics.status = "error"
            metrics.error_type = type(e).__name__
            self.error_count += 1
            raise
        finally:
            metrics.latency_ms = (time.time() - start_time) * 1000
            self.total_requests += 1
            self.total_latency += metrics.latency_ms
            if metrics.cache_hit:
                self.cached_requests += 1
            self.metrics_history.append(metrics)

            # 生产环境：上报到监控系统（Prometheus/Datadog等）
            self._report_metrics(metrics)

    def _report_metrics(self, metrics: LLMRequestMetrics):
        """上报指标（生产环境实现）"""
        # 生产环境中上报到监控系统
        pass

    def get_dashboard_stats(self) -> Dict:
        """获取统计面板数据"""
        if self.total_requests == 0:
            return {'total_requests': 0}

        avg_latency = self.total_latency / self.total_requests
        cache_hit_rate = self.cached_requests / self.total_requests * 100
        error_rate = self.error_count / self.total_requests * 100

        return {
            'total_requests': self.total_requests,
            'avg_latency_ms': round(avg_latency, 2),
            'cache_hit_rate': f"{cache_hit_rate:.1f}%",
            'error_rate': f"{error_rate:.2f}%",
            'cached_requests': self.cached_requests,
            'errors': self.error_count,
            'total_cost': round(sum(m.cost for m in self.metrics_history), 4)
        }

# 使用示例
if __name__ == "__main__":
    observer = LLMObserver()

    # 模拟请求
    for i in range(3):
        try:
            with observer.trace("gpt-4o", "user_001") as metrics:
                # 模拟API调用
                time.sleep(0.1)
                metrics.prompt_tokens = 150
                metrics.completion_tokens = 250
                metrics.cost = 0.0045
                if i == 1:
                    metrics.cache_hit = True
        except Exception as e:
            print(f"请求失败: {e}")

    stats = observer.get_dashboard_stats()
    print("=== 监控面板 ===")
    for k, v in stats.items():
        print(f"  {k}: {v}")

from typing import Dict, List, Optional
import time
import uuid

class LLMGateway:
    """
    生产级LLM API网关
    集成：限流 + 缓存 + 重试 + 熔断 + 成本管理 + 可观测性 + 模型路由
    """

    def __init__(self):
        self.rate_limiter = MultiLevelRateLimiter()
        self.cache = SemanticCache()
        self.circuit_breaker = CircuitBreaker(
            failure_threshold=10,
            recovery_timeout=60.0
        )
        self.cost_manager = CostManager()
        self.observer = LLMObserver()
        self.router = ModelRouter()

    def chat(self, user_id: str, messages: List[Dict],
             model: Optional[str] = None,
             preference: str = 'balanced',
             use_cache: bool = True,
             max_tokens: int = 1000) -> Dict:
        """统一聊天接口"""
        request_id = str(uuid.uuid4())

        with self.observer.trace(model or "auto", user_id, 
                                 {"request_id": request_id}) as metrics:

            # 1. 限流检查
            rate_result = self.rate_limiter.check(user_id)
            if not rate_result['allowed']:
                metrics.status = "rate_limited"
                return {
                    'success': False,
                    'error': 'rate_limit_exceeded',
                    'message': f"请求过于频繁，请等待 {rate_result['wait_time']:.1f} 秒",
                    'request_id': request_id
                }

            # 2. 模型路由
            query = messages[-1].get('content', '') if messages else ''
            context = ' '.join([m.get('content', '') for m in messages[:-1]])

            if model is None:
                model, route_info = self.router.route(query, context, preference)
                metrics.metadata['route_info'] = route_info

            metrics.model = model

            # 3. 缓存查询
            if use_cache:
                cached_response = self.cache.lookup(query, model)
                if cached_response:
                    metrics.cache_hit = True
                    metrics.prompt_tokens = 0
                    metrics.completion_tokens = 0
                    return {
                        'success': True,
                        'content': cached_response,
                        'model': model,
                        'from_cache': True,
                        'request_id': request_id,
                        'latency_ms': metrics.latency_ms
                    }

            # 4. 熔断检查
            if not self.circuit_breaker.can_execute():
                metrics.status = "circuit_open"
                return {
                    'success': False,
                    'error': 'circuit_breaker_open',
                    'message': '服务暂时不可用，请稍后再试',
                    'request_id': request_id
                }

            # 5. 调用LLM API（带重试）
            try:
                response = self._call_llm_with_retry(model, messages, max_tokens)

                # 记录成功
                self.circuit_breaker.record_success()

                # 成本记录
                prompt_tokens = response.get('usage', {}).get('prompt_tokens', 0)
                completion_tokens = response.get('usage', {}).get('completion_tokens', 0)
                cost = self.cost_manager.record_cost(
                    request_id, user_id, model,
                    prompt_tokens, completion_tokens
                )

                metrics.prompt_tokens = prompt_tokens
                metrics.completion_tokens = completion_tokens
                metrics.cost = cost

                # 存入缓存
                if use_cache:
                    self.cache.store(query, model, response['content'])

                return {
                    'success': True,
                    'content': response['content'],
                    'model': model,
                    'usage': response.get('usage', {}),
                    'cost': cost,
                    'from_cache': False,
                    'request_id': request_id,
                    'latency_ms': metrics.latency_ms
                }

            except Exception as e:
                self.circuit_breaker.record_failure()
                metrics.status = "error"
                metrics.error_type = type(e).__name__
                raise

    @retry_with_backoff(
        max_retries=3,
        base_delay=1.0,
        retry_on_exceptions=(TimeoutError, ConnectionError)
    )
    def _call_llm_with_retry(self, model: str, messages: List[Dict],
                             max_tokens: int) -> Dict:
        """调用LLM API（带重试）"""
        # 实际项目中调用OpenAI/Anthropic等API
        # 这里是模拟
        return {
            'content': f"这是 {model} 的回复（模拟）",
            'usage': {
                'prompt_tokens': 150,
                'completion_tokens': 200,
                'total_tokens': 350
            }
        }