# 安装 vLLM
pip install vllm

# 用 vLLM 启动一个 OpenAI 兼容的 API 服务
# --model 指定模型，--host 和 --port 指定监听地址
# --tensor-parallel-size 用几张卡并行推理
python -m vllm.entrypoints.openai.api_server \
    --model Qwen/Qwen2.5-7B-Instruct \
    --host 0.0.0.0 \
    --port 8000 \
    --tensor-parallel-size 1 \
    --gpu-memory-utilization 0.9 \
    --max-model-len 8192

# 文件路径：test_vllm_client.py
from openai import OpenAI

# 连接到本地 vLLM 服务
client = OpenAI(
    base_url="http://localhost:8000/v1",
    api_key="runoob-demo-key"  # vLLM 默认不需要真实 API key
)

# 调用聊天补全
response = client.chat.completions.create(
    model="Qwen/Qwen2.5-7B-Instruct",
    messages=[
        {"role": "system", "content": "你是一个乐于助人的 AI 助手。"},
        {"role": "user", "content": "用一句话介绍 runoob 教程"}
    ],
    temperature=0.7,
    max_tokens=500,
    stream=True  # 流式输出
)

print("AI 回复：", end="", flush=True)
for chunk in response:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)
print()

# 文件路径：custom_vllm_server.py
from vllm import LLM, SamplingParams

# 初始化模型
llm = LLM(
    model="Qwen/Qwen2.5-7B-Instruct",
    gpu_memory_utilization=0.9,  # GPU 显存使用比例
    tensor_parallel_size=1,      # 张量并行度（几张卡）
    max_model_len=8192,          # 最大上下文长度
)

# 配置采样参数
sampling_params = SamplingParams(
    temperature=0.7,
    top_p=0.9,
    max_tokens=500,
    stop=["</s>"],
)

# 批量推理
prompts = [
    "介绍一下 runoob",
    "什么是 AI？",
    "如何学习编程？",
]

# 生成回复
outputs = llm.generate(prompts, sampling_params)

# 打印结果
for output in outputs:
    prompt = output.prompt
    generated_text = output.outputs[0].text
    print(f"Prompt: {prompt}")
    print(f"Generated: {generated_text}")
    print("-" * 50)

# 用 Docker 运行 TGI（最简单的方式）
# 注意：需要先安装 Docker 并配置好 GPU 支持
docker run -d \
    --gpus all \
    -p 8080:80 \
    -v $PWD/data:/data \
    --name tgi-server \
    ghcr.io/huggingface/text-generation-inference:latest \
    --model-id Qwen/Qwen2.5-7B-Instruct \
    --max-input-length 4096 \
    --max-total-tokens 8192 \
    --max-batch-prefill-tokens 4096

# 或者直接用 Python 安装（开发环境）
# pip install text-generation

# 文件路径：test_tgi_client.py
from text_generation import Client

# 连接到 TGI 服务
client = Client("http://localhost:8080", timeout=60)

# 非流式调用
response = client.generate(
    "介绍一下 runoob 教程",
    max_new_tokens=500,
    temperature=0.7,
    top_p=0.9,
)
print(response.generated_text)

print("-" * 50)

# 流式调用
print("流式输出：")
for chunk in client.generate_stream(
    "用 Python 写一个 Hello World",
    max_new_tokens=200,
):
    if not chunk.token.special:
        print(chunk.token.text, end="", flush=True)
print()

# 安装 Ollama（Linux 服务器）
curl -fsSL https://ollama.com/install.sh | sh

# 配置 Ollama 监听所有网络接口（生产环境注意防火墙）
# 编辑 systemd 服务文件
sudo systemctl edit ollama.service

# 添加以下内容：
# [Service]
# Environment="OLLAMA_HOST=0.0.0.0:11434"
# Environment="OLLAMA_MODELS=/var/lib/ollama/models"

# 重启服务
sudo systemctl restart ollama

# 拉取并运行模型
ollama pull qwen2.5:7b

# 文件路径：test_ollama_client.py
import requests
import json

# Ollama API 地址
base_url = "http://localhost:11434/api"

# 聊天接口（流式）
def chat_stream(model, messages):
    url = f"{base_url}/chat"
    data = {
        "model": model,
        "messages": messages,
        "stream": True,
        "options": {
            "temperature": 0.7,
            "num_ctx": 8192,
        }
    }

    response = requests.post(url, json=data, stream=True)
    full_response = ""

    for line in response.iter_lines():
        if line:
            chunk = json.loads(line)
            if "message" in chunk:
                content = chunk["message"].get("content", "")
                full_response += content
                print(content, end="", flush=True)
            if chunk.get("done", False):
                break

    print()
    return full_response

# 调用示例
messages = [
    {"role": "system", "content": "你是一个乐于助人的 AI 助手。"},
    {"role": "user", "content": "介绍一下 runoob 教程"}
]

print("AI 回复：", end="")
result = chat_stream("qwen2.5:7b", messages)

# vLLM 启用投机采样
# --speculative-model 指定小模型（草稿模型）
# --num-speculative-tokens 指定每次猜几个字
python -m vllm.entrypoints.openai.api_server \
    --model Qwen/Qwen2.5-7B-Instruct \
    --speculative-model Qwen/Qwen2.5-0.5B-Instruct \
    --num-speculative-tokens 5 \
    --host 0.0.0.0 \
    --port 8000

# 直接用 vLLM 运行 AWQ 量化模型
# 很多量化模型可以直接从 Hugging Face 下载
python -m vllm.entrypoints.openai.api_server \
    --model TheBloke/Qwen2.5-7B-Instruct-AWQ \
    --quantization awq \
    --host 0.0.0.0 \
    --port 8000

# 或者用 Ollama 运行 GGUF 量化模型
ollama run qwen2.5:7b-instruct-q4_0

# 文件路径：awq_inference.py
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

# 加载 AWQ 量化模型
model_name = "TheBloke/Qwen2.5-7B-Instruct-AWQ"

# 注意：实际运行需要安装 autoawq 库
# pip install autoawq

tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype=torch.float16,
    device_map="auto",
    trust_remote_code=True,
)

# 推理
messages = [
    {"role": "system", "content": "你是一个乐于助人的 AI 助手。"},
    {"role": "user", "content": "介绍 runoob"}
]

text = tokenizer.apply_chat_template(
    messages,
    tokenize=False,
    add_generation_prompt=True
)

inputs = tokenizer([text], return_tensors="pt").to(model.device)

with torch.no_grad():
    outputs = model.generate(
        **inputs,
        max_new_tokens=500,
        temperature=0.7,
    )

response = tokenizer.decode(
    outputs[0][inputs.input_ids.shape[1]:],
    skip_special_tokens=True
)
print(response)

# 文件路径：llm_api_server.py
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from typing import List, Optional, AsyncGenerator
from enum import Enum
import time
import asyncio
import uuid
from datetime import datetime, timedelta
from collections import defaultdict

# ============================================
# 1. 初始化应用和配置
# ============================================

app = FastAPI(title="RUNOOB LLM API", version="1.0.0")

# 简单的 API Key 认证（生产环境建议用更完善的方案）
VALID_API_KEYS = {
    "sk-runoob-123456": {"user": "demo_user", "quota": 1000},
    "sk-runoob-789012": {"user": "pro_user", "quota": 10000},
}

# 简单的限流器（内存实现，生产环境建议用 Redis）
class RateLimiter:
    def __init__(self):
        self.requests = defaultdict(list)  # key -> [timestamp, ...]

    async def is_allowed(self, key: str, max_requests: int, window_seconds: int) -> bool:
        """检查是否超过限流"""
        now = time.time()
        # 清理过期记录
        self.requests[key] = [t for t in self.requests[key] if now - t < window_seconds]
        # 检查是否超限
        if len(self.requests[key]) >= max_requests:
            return False
        # 记录这次请求
        self.requests[key].append(now)
        return True

rate_limiter = RateLimiter()

# ============================================
# 2. 数据模型
# ============================================

class Role(str, Enum):
    SYSTEM = "system"
    USER = "user"
    ASSISTANT = "assistant"

class Message(BaseModel):
    role: Role = Field(..., description="消息角色")
    content: str = Field(..., description="消息内容")

class ChatCompletionRequest(BaseModel):
    model: str = Field(..., description="模型名称")
    messages: List[Message] = Field(..., description="对话消息列表")
    temperature: Optional[float] = Field(default=0.7, ge=0, le=2, description="采样温度")
    top_p: Optional[float] = Field(default=0.9, ge=0, le=1, description="Top-P 采样")
    max_tokens: Optional[int] = Field(default=500, ge=1, le=4096, description="最大生成 token 数")
    stream: Optional[bool] = Field(default=False, description="是否流式输出")

class ChatCompletionResponse(BaseModel):
    id: str
    object: str = "chat.completion"
    created: int
    model: str
    choices: List[dict]
    usage: dict

# ============================================
# 3. 认证依赖
# ============================================

security = HTTPBearer()

async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    """验证 API Key 并返回用户信息"""
    api_key = credentials.credentials
    if api_key not in VALID_API_KEYS:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid API Key"
        )
    return VALID_API_KEYS[api_key]

# ============================================
# 4. 模拟 LLM 生成（实际项目中替换成真实模型调用）
# ============================================

async def mock_llm_generate(messages: List[Message], max_tokens: int) -> str:
    """模拟 LLM 生成过程"""
    # 这里只是模拟，实际项目中应该调用真实模型
    response_text = (
        "你好！我是 RUNOOB 的 AI 助手。\n\n"
        "RUNOOB 提供了丰富的编程教程，包括 Python、Java、C++、前端开发等。\n"
        "你可以访问 https://www.runoob.com 学习更多内容！\n\n"
        "有什么我可以帮助你的吗？"
    )
    # 模拟生成延迟
    await asyncio.sleep(0.5)
    return response_text

async def mock_llm_generate_stream(messages: List[Message], max_tokens: int) -> AsyncGenerator[str, None]:
    """模拟 LLM 流式生成"""
    response_text = await mock_llm_generate(messages, max_tokens)
    # 逐字输出，模拟流式效果
    for char in response_text:
        yield char
        await asyncio.sleep(0.03)  # 每个字延迟 30ms

# ============================================
# 5. API 端点
# ============================================

@app.post("/v1/chat/completions")
async def chat_completions(
    request: ChatCompletionRequest,
    user: dict = Depends(get_current_user)
):
    """聊天补全接口，兼容 OpenAI 格式"""

    # 限流检查：每个用户每分钟最多 60 次请求
    user_key = user["user"]
    if not await rate_limiter.is_allowed(user_key, max_requests=60, window_seconds=60):
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="Rate limit exceeded"
        )

    # 生成请求 ID
    request_id = f"chatcmpl-{uuid.uuid4().hex[:24]}"
    created = int(datetime.now().timestamp())

    if request.stream:
        # 流式响应
        async def stream_generator():
            async for token in mock_llm_generate_stream(request.messages, request.max_tokens):
                # 构造 SSE 格式的 chunk
                chunk = {
                    "id": request_id,
                    "object": "chat.completion.chunk",
                    "created": created,
                    "model": request.model,
                    "choices": [{
                        "index": 0,
                        "delta": {"content": token},
                        "finish_reason": None
                    }]
                }
                yield f"data: {__import__('json').dumps(chunk)}\n\n"
            # 发送结束标记
            yield "data: [DONE]\n\n"

        return StreamingResponse(
            stream_generator(),
            media_type="text/event-stream",
            headers={"Cache-Control": "no-cache"}
        )

    else:
        # 非流式响应
        response_text = await mock_llm_generate(request.messages, request.max_tokens)

        return ChatCompletionResponse(
            id=request_id,
            created=created,
            model=request.model,
            choices=[{
                "index": 0,
                "message": {
                    "role": "assistant",
                    "content": response_text
                },
                "finish_reason": "stop"
            }],
            usage={
                "prompt_tokens": sum(len(m.content) for m in request.messages),
                "completion_tokens": len(response_text),
                "total_tokens": sum(len(m.content) for m in request.messages) + len(response_text)
            }
        )

@app.get("/health")
async def health_check():
    """健康检查端点"""
    return {"status": "ok", "timestamp": datetime.now().isoformat()}

@app.get("/v1/models")
async def list_models(user: dict = Depends(get_current_user)):
    """列出可用模型"""
    return {
        "object": "list",
        "data": [
            {
                "id": "qwen2.5-7b-instruct",
                "object": "model",
                "created": 1699000000,
                "owned_by": "runoob"
            },
            {
                "id": "llama-3.1-8b-instruct",
                "object": "model",
                "created": 1699000000,
                "owned_by": "runoob"
            }
        ]
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# 安装依赖
pip install fastapi uvicorn python-multipart pydantic

# 启动服务
python llm_api_server.py

# 在另一个终端测试
curl -X POST http://localhost:8000/v1/chat/completions \
    -H "Content-Type: application/json" \
    -H "Authorization: Bearer sk-runoob-123456" \
    -d '{
        "model": "qwen2.5-7b-instruct",
        "messages": [
            {"role": "user", "content": "你好"}
        ],
        "stream": false
    }'

# 文件路径：client_example.py
from openai import OpenAI

# 连接到我们自己的 API 服务
client = OpenAI(
    base_url="http://localhost:8000/v1",
    api_key="sk-runoob-123456"
)

# 测试非流式
print("非流式调用：")
response = client.chat.completions.create(
    model="qwen2.5-7b-instruct",
    messages=[{"role": "user", "content": "介绍一下 runoob"}],
    stream=False
)
print(response.choices[0].message.content)

print("\n" + "="*50 + "\n")

# 测试流式
print("流式调用：")
stream = client.chat.completions.create(
    model="qwen2.5-7b-instruct",
    messages=[{"role": "user", "content": "介绍一下 runoob"}],
    stream=True
)
for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)
print()

# 文件路径：redis_rate_limiter.py
import redis.asyncio as redis
import time

class RedisRateLimiter:
    """基于 Redis 的分布式限流器"""

    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)

    async def is_allowed(
        self,
        key: str,
        max_requests: int,
        window_seconds: int
    ) -> bool:
        """
        滑动窗口限流算法
        key: 限流键（比如 user_id 或 api_key）
        max_requests: 窗口内最大请求数
        window_seconds: 窗口大小（秒）
        """
        now = time.time()
        window_start = now - window_seconds

        # Redis 键
        redis_key = f"rate_limit:{key}"

        # 使用 pipeline 原子操作
        async with self.redis.pipeline() as pipe:
            # 1. 移除窗口外的记录
            pipe.zremrangebyscore(redis_key, 0, window_start)
            # 2. 统计当前窗口内的请求数
            pipe.zcard(redis_key)
            # 3. 添加当前请求
            pipe.zadd(redis_key, {str(now): now})
            # 4. 设置过期时间
            pipe.expire(redis_key, window_seconds)
            # 执行
            _, current_count, _, _ = await pipe.execute()

        return current_count < max_requests

    async def close(self):
        await self.redis.close()


# 使用示例
async def main():
    limiter = RedisRateLimiter("redis://localhost:6379")

    # 测试：同一用户每分钟最多 10 次请求
    user_id = "user_123"
    for i in range(15):
        allowed = await limiter.is_allowed(user_id, max_requests=10, window_seconds=60)
        print(f"请求 {i+1}: {'允许' if allowed else '拒绝'}")

    await limiter.close()

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

# 文件路径：nginx.conf
# Nginx 配置：多台 LLM 服务的负载均衡

upstream llm_backends {
    # 最少连接数算法：把请求发给当前连接最少的服务器
    least_conn;

    # 后端服务器列表
    server 192.168.1.101:8000 weight=1 max_fails=3 fail_timeout=30s;
    server 192.168.1.102:8000 weight=1 max_fails=3 fail_timeout=30s;
    server 192.168.1.103:8000 weight=1 max_fails=3 fail_timeout=30s;

    # 如果有多型号 GPU，可以按 weight 分配不同比例的流量
    # weight=2 表示这台服务器分到的流量是其他的 2 倍
}

server {
    listen 80;
    server_name api.runoob.com;

    # API 请求转发
    location /v1/ {
        # 转发给后端
        proxy_pass http://llm_backends;

        # 超时设置（AI 生成可能比较慢）
        proxy_connect_timeout 10s;
        proxy_send_timeout 600s;
        proxy_read_timeout 600s;

        # 传递客户端真实 IP
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

        # 流式响应需要的设置
        proxy_buffering off;
        proxy_cache off;
        proxy_set_header Connection '';
        proxy_http_version 1.1;
    }

    # 健康检查端点
    location /health {
        proxy_pass http://llm_backends/health;
        proxy_connect_timeout 5s;
        proxy_read_timeout 5s;
    }

    # 限流：每个 IP 每秒最多 10 个请求
    limit_req_zone $binary_remote_addr zone=api_limit:10m rate=10r/s;
    limit_req zone=api_limit burst=20 nodelay;
}

# 文件路径：llm-deployment.yaml
# LLM 服务的 Kubernetes 部署配置

# 1. Deployment：部署 LLM Pod
apiVersion: apps/v1
kind: Deployment
metadata:
  name: llm-server
  namespace: runoob
spec:
  replicas: 3  # 初始副本数
  selector:
    matchLabels:
      app: llm-server
  template:
    metadata:
      labels:
        app: llm-server
    spec:
      containers:
      - name: llm-server
        image: runoob/llm-server:latest
        ports:
        - containerPort: 8000
        resources:
          requests:
            cpu: "4"
            memory: "16Gi"
            nvidia.com/gpu: 1  # 请求 1 块 GPU
          limits:
            cpu: "8"
            memory: "32Gi"
            nvidia.com/gpu: 1  # 限制 1 块 GPU
        env:
        - name: MODEL_NAME
          value: "Qwen/Qwen2.5-7B-Instruct"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 300  # 模型加载需要时间
          periodSeconds: 30
        readinessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 300
          periodSeconds: 10
      tolerations:
      - key: "nvidia.com/gpu"
        operator: "Exists"
        effect: "NoSchedule"

---
# 2. Service：负载均衡 Service
apiVersion: v1
kind: Service
metadata:
  name: llm-service
  namespace: runoob
spec:
  selector:
    app: llm-server
  ports:
  - port: 80
    targetPort: 8000
  type: LoadBalancer

---
# 3. HPA：自动扩缩容
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: llm-hpa
  namespace: runoob
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: llm-server
  minReplicas: 3   # 最少 3 个 Pod
  maxReplicas: 20  # 最多 20 个 Pod
  metrics:
 # 根据 CPU 使用率扩容
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  # 根据 GPU 使用率扩容（需要自定义指标）
  - type: Pods
    pods:
      metric:
        name: gpu_utilization
      target:
        type: AverageValue
        averageValue: 80
  # 根据队列长度扩容（自定义指标）
  - type: Pods
    pods:
      metric:
        name: pending_requests
      target:
        type: AverageValue
        averageValue: 10
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 60    # 扩容前观察 60 秒
      policies:
      - type: Percent
        value: 50
        periodSeconds: 60
    scaleDown:
      stabilizationWindowSeconds: 300   # 缩容前观察 5 分钟
      policies:
      - type: Percent
        value: 20
        periodSeconds: 120

# 文件路径：ab_test_router.py
import hashlib
import random
from typing import Dict, List, Tuple
from dataclasses import dataclass

@dataclass
class Variant:
    """A/B 测试的一个变体"""
    name: str           # 变体名称
    model_name: str     # 模型名称
    weight: int         # 流量权重
    config: dict        # 其他配置


class ABTestRouter:
    """A/B 测试流量路由器"""

    def __init__(self, test_name: str, variants: List[Variant]):
        self.test_name = test_name
        self.variants = variants
        # 计算总权重
        total_weight = sum(v.weight for v in variants)
        # 构建分段范围
        self.ranges: List[Tuple[int, int, Variant]] = []
        current = 0
        for v in variants:
            self.ranges.append((current, current + v.weight, v))
            current += v.weight
        self.total_weight = total_weight

    def get_variant_by_user(self, user_id: str) -> Variant:
        """
        基于用户 ID 的稳定分流
        同一个用户总是被分到同一个组，避免体验不一致
        """
        # 用哈希计算用户的分数
        hash_input = f"{self.test_name}-{user_id}".encode()
        hash_value = hashlib.md5(hash_input).hexdigest()
        # 转换成 0-999 的数字
        score = int(hash_value[:4], 16) % 1000
        # 查找对应的变体
        for start, end, variant in self.ranges:
            if start <= score < end:
                return variant
        return self.variants[-1]

    def get_variant_random(self) -> Variant:
        """随机分流（适合不需要粘性的场景）"""
        score = random.randint(0, self.total_weight - 1)
        for start, end, variant in self.ranges:
            if start <= score < end:
                return variant
        return self.variants[-1]


# 使用示例
def main():
    # 定义三个变体
    variants = [
        Variant(
            name="control",
            model_name="Qwen2.5-7B-Instruct",
            weight=50,  # 50% 流量
            config={"temperature": 0.7}
        ),
        Variant(
            name="treatment-v1",
            model_name="Qwen2.5-14B-Instruct",
            weight=30,  # 30% 流量
            config={"temperature": 0.7}
        ),
        Variant(
            name="treatment-v2",
            model_name="Llama-3.1-8B-Instruct",
            weight=20,  # 20% 流量
            config={"temperature": 0.8}
        )
    ]

    router = ABTestRouter("model-comparison-v1", variants)

    # 测试用户分流
    test_users = [f"user_{i}" for i in range(10)]
    for user_id in test_users:
        variant = router.get_variant_by_user(user_id)
        print(f"用户 {user_id} → 组 {variant.name} → 模型 {variant.model_name}")

    # 统计分布
    count: Dict[str, int] = {}
    for i in range(10000):
        variant = router.get_variant_random()
        count[variant.name] = count.get(variant.name, 0) + 1
    print("\n随机分流统计（10000 次）：")
    for name, cnt in count.items():
        print(f"  {name}: {cnt} 次 ({cnt/100}%)")


if __name__ == "__main__":
    main()

# 文件路径：metrics_collector.py
import time
import json
from dataclasses import dataclass, asdict
from typing import Optional
from datetime import datetime
import uuid

@dataclass
class ChatMetrics:
    """聊天请求的完整指标"""
    # 基础信息
    request_id: str
    user_id: str
    variant_name: str       # A/B 测试组名
    model_name: str

    # 时间指标
    timestamp: str
    ttft: float             # Time to First Token
    total_time: float       # 总耗时

    # 生成指标
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int

    # 业务指标
    user_rating: Optional[int] = None  # 用户评分（1-5）
    is_error: bool = False
    error_message: Optional[str] = None


class MetricsCollector:
    """指标收集器"""

    def __init__(self):
        # 实际项目中，这里应该写入 Kafka 或数据库
        self.buffer = []

    def collect(self, metrics: ChatMetrics):
        """收集一条指标"""
        self.buffer.append(asdict(metrics))
        # 简单输出，实际项目中写入存储
        print(f"[Metrics] {json.dumps(asdict(metrics), ensure_ascii=False)}")

        # 缓冲区满了就批量写入
        if len(self.buffer) >= 100:
            self.flush()

    def flush(self):
        """批量写入存储"""
        if self.buffer:
            # 实际项目中：写入 ClickHouse、BigQuery 等
            print(f"Flushed {len(self.buffer)} metrics")
            self.buffer = []


# 使用示例
collector = MetricsCollector()

def process_chat_request(user_id: str, prompt: str, router):
    """处理一次聊天请求，记录指标"""
    request_id = str(uuid.uuid4())

    # 分流
    variant = router.get_variant_by_user(user_id)

    # 记录开始时间
    start_time = time.time()
    ttft = 0.0
    prompt_tokens = len(prompt)
    completion_tokens = 0

    try:
        # 这里替换成真实的模型调用
        # response = call_model(variant.model_name, prompt)
        response = "这是 AI 的回复"
        completion_tokens = len(response)

        # 模拟 TTFT（假设在 0.2 秒后收到第一个 token）
        ttft = 0.2

        # 记录成功指标
        metrics = ChatMetrics(
            request_id=request_id,
            user_id=user_id,
            variant_name=variant.name,
            model_name=variant.model_name,
            timestamp=datetime.now().isoformat(),
            ttft=ttft,
            total_time=time.time() - start_time,
            prompt_tokens=prompt_tokens,
            completion_tokens=completion_tokens,
            total_tokens=prompt_tokens + completion_tokens,
            is_error=False
        )
        collector.collect(metrics)
        return response, request_id

    except Exception as e:
        # 记录错误指标
        metrics = ChatMetrics(
            request_id=request_id,
            user_id=user_id,
            variant_name=variant.name,
            model_name=variant.model_name,
            timestamp=datetime.now().isoformat(),
            ttft=ttft,
            total_time=time.time() - start_time,
            prompt_tokens=prompt_tokens,
            completion_tokens=0,
            total_tokens=prompt_tokens,
            is_error=True,
            error_message=str(e)
        )
        collector.collect(metrics)
        raise

# 用户反馈
def submit_user_feedback(request_id: str, rating: int):
    """用户提交评分反馈"""
    # 实际项目中：更新数据库中的对应记录
    print(f"Request {request_id} got rating {rating}")

# 文件路径：ab_test_analysis.py
import math
from typing import Tuple

def calculate_z_test(
    control_conv: float,
    treatment_conv: float,
    control_total: int,
    treatment_total: int
) -> Tuple[float, float]:
    """
    计算 Z 检验的 p-value
    control_conv: 对照组转化率
    treatment_conv: 实验组转化率
    control_total: 对照组样本量
    treatment_total: 实验组样本量
    """
    # 合并转化率
    p_pool = (control_conv * control_total + treatment_conv * treatment_total) / (control_total + treatment_total)
    # 标准误差
    se_pool = math.sqrt(p_pool * (1 - p_pool) * (1 / control_total + 1 / treatment_total))
    # Z 统计量
    z_score = (treatment_conv - control_conv) / se_pool
    # 计算 p-value（双尾检验）
    p_value = 2 * (1 - normal_cdf(abs(z_score)))
    return z_score, p_value

def normal_cdf(x: float) -> float:
    """标准正态分布的累积分布函数（近似）"""
    return (1.0 + math.erf(x / math.sqrt(2.0))) / 2.0

def analyze_ab_test(
    control_conversions: int,
    control_total: int,
    treatment_conversions: int,
    treatment_total: int,
    alpha: float = 0.05
):
    """分析 A/B 测试结果"""
    control_conv = control_conversions / control_total
    treatment_conv = treatment_conversions / treatment_total

    z_score, p_value = calculate_z_test(
        control_conv, treatment_conv,
        control_total, treatment_total
    )

    relative_lift = (treatment_conv - control_conv) / control_conv

    print("A/B 测试结果分析")
    print("=" * 50)
    print(f"对照组: {control_conversions}/{control_total} ({control_conv:.2%})")
    print(f"实验组: {treatment_conversions}/{treatment_total} ({treatment_conv:.2%})")
    print(f"相对提升: {relative_lift:+.2%}")
    print("-" * 50)
    print(f"Z 统计量: {z_score:.4f}")
    print(f"p-value: {p_value:.4f}")
    print("-" * 50)

    if p_value < alpha:
        print(f"&#x2705; 结果统计显著（p < {alpha}）")
        if relative_lift > 0:
            print("   实验组表现更好！")
        else:
            print("   对照组表现更好！")
    else:
        print(f"&#x274c; 结果不显著（p >= {alpha}）")
        print("   无法确定哪个版本更好，需要更多数据")


# 示例：假设 A/B 测试结果
if __name__ == "__main__":
    # 场景 1：有显著差异
    print("场景 1：有显著差异")
    analyze_ab_test(
        control_conversions=450,   # 对照组 450 人满意
        control_total=1000,        # 对照组共 1000 人
        treatment_conversions=520, # 实验组 520 人满意
        treatment_total=1000       # 实验组共 1000 人
    )

    print("\n" + "="*50 + "\n")

    # 场景 2：无显著差异
    print("场景 2：无显著差异")
    analyze_ab_test(
        control_conversions=48,
        control_total=100,
        treatment_conversions=52,
        treatment_total=100
    )

# 文件路径：mlflow_experiment.py
import mlflow
import mlflow.pyfunc
import mlflow.sklearn
from mlflow.tracking import MlflowClient
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, f1_score
import numpy as np
import time

# 设置 MLflow 跟踪地址（本地或远程服务器）
mlflow.set_tracking_uri("http://localhost:5000")

# 创建或设置实验
experiment_name = "runoob-llm-finetuning"
try:
    experiment_id = mlflow.create_experiment(experiment_name)
except mlflow.exceptions.MlflowException:
    experiment_id = mlflow.get_experiment_by_name(experiment_name).experiment_id

mlflow.set_experiment(experiment_name)

# 模拟一个 LLM 微调实验
def run_finetuning_experiment(
    model_name: str,
    learning_rate: float,
    batch_size: int,
    num_epochs: int,
    lora_rank: int,
    dataset_name: str = "runoob-demo"
):
    """运行一次微调实验并记录到 MLflow"""

    with mlflow.start_run(run_name=f"{model_name}-lr{learning_rate}-bs{batch_size}") as run:
        run_id = run.info.run_id

        # 记录参数
        mlflow.log_param("model_name", model_name)
        mlflow.log_param("learning_rate", learning_rate)
        mlflow.log_param("batch_size", batch_size)
        mlflow.log_param("num_epochs", num_epochs)
        mlflow.log_param("lora_rank", lora_rank)
        mlflow.log_param("dataset_name", dataset_name)

        # 模拟训练过程，记录指标
        print(f"开始训练实验 {run_id}...")

        for epoch in range(num_epochs):
            # 模拟训练损失（随 epoch 下降）
            train_loss = 2.0 - 0.3 * epoch + np.random.normal(0, 0.1)
            val_loss = 2.2 - 0.25 * epoch + np.random.normal(0, 0.15)

            # 记录指标
            mlflow.log_metric("train_loss", train_loss, step=epoch)
            mlflow.log_metric("val_loss", val_loss, step=epoch)
            mlflow.log_metric("epoch", epoch, step=epoch)

            print(f"Epoch {epoch}: train_loss={train_loss:.4f}, val_loss={val_loss:.4f}")
            time.sleep(0.5)

        # 最终评估指标
        final_val_loss = 1.2 + np.random.normal(0, 0.1)
        final_accuracy = 0.85 + np.random.normal(0, 0.03)
        final_f1 = 0.83 + np.random.normal(0, 0.04)

        mlflow.log_metric("final_val_loss", final_val_loss)
        mlflow.log_metric("final_accuracy", final_accuracy)
        mlflow.log_metric("final_f1", final_f1)

        # 记录文件（比如训练日志、示例输出）
        with open("training_log.txt", "w") as f:
            f.write(f"Experiment run_id: {run_id}\n")
            f.write(f"Model: {model_name}\n")
            f.write(f"Final accuracy: {final_accuracy:.4f}\n")

        mlflow.log_artifact("training_log.txt")

        # 记录模型（这里用 sklearn 模拟，实际项目记录真实 LLM）
        # 实际项目中可以用 mlflow.transformers.log_model
        X, y = load_iris(return_X_y=True)
        X_train, X_test, y_train, y_test = train_test_split(X, y, random_state=42)
        model = RandomForestClassifier(n_estimators=100)
        model.fit(X_train, y_train)

        mlflow.sklearn.log_model(
            model,
            "model",
            registered_model_name=f"{model_name.replace('/', '-')}-demo"
        )

        print(f"实验完成！run_id: {run_id}")
        print(f"最终准确率: {final_accuracy:.4f}")

        return run_id, final_accuracy


# 运行几次实验
if __name__ == "__main__":
    print("启动 MLflow UI（在另一个终端运行）：")
    print("  mlflow ui --port 5000")
    print("\n开始实验...\n")

    # 实验 1：基础配置
    run_id1, acc1 = run_finetuning_experiment(
        model_name="Qwen2.5-7B-Instruct",
        learning_rate=2e-5,
        batch_size=16,
        num_epochs=3,
        lora_rank=8
    )

    print()

    # 实验 2：更大的 LoRA rank
    run_id2, acc2 = run_finetuning_experiment(
        model_name="Qwen2.5-7B-Instruct",
        learning_rate=2e-5,
        batch_size=16,
        num_epochs=3,
        lora_rank=16
    )

    print()

    # 实验 3：更高的学习率
    run_id3, acc3 = run_finetuning_experiment(
        model_name="Qwen2.5-7B-Instruct",
        learning_rate=5e-5,
        batch_size=16,
        num_epochs=3,
        lora_rank=8
    )

    print()
    print("所有实验完成！")
    print(f"实验 1 (run_id={run_id1}): {acc1:.4f}")
    print(f"实验 2 (run_id={run_id2}): {acc2:.4f}")
    print(f"实验 3 (run_id={run_id3}): {acc3:.4f}")
    print("\n查看结果: http://localhost:5000")

# 文件路径：mlflow_model_registry.py
import mlflow
from mlflow.tracking import MlflowClient

mlflow.set_tracking_uri("http://localhost:5000")

client = MlflowClient()

def promote_model_to_production(
    model_name: str,
    version: int,
    description: str = ""
):
    """把模型升级到 Production 阶段"""
    # 更新版本描述
    client.update_model_version(
        name=model_name,
        version=version,
        description=description
    )

    # 把这个版本设为 Production
    client.transition_model_version_stage(
        name=model_name,
        version=version,
        stage="Production",
        archive_existing_versions=True  # 归档旧版本
    )
    print(f"模型 {model_name} 版本 {version} 已升级到 Production")


def get_production_model(model_name: str):
    """获取当前 Production 模型"""
    versions = client.get_latest_versions(model_name, stages=["Production"])
    if versions:
        return versions[0]
    return None


def list_all_models():
    """列出所有已注册的模型"""
    models = client.search_registered_models()
    print("已注册的模型：")
    for model in models:
        print(f"  - {model.name}")
        # 列出该模型的所有版本
        versions = client.search_model_versions(f"name='{model.name}'")
        for v in versions:
            print(f"    Version {v.version}: {v.current_stage}")


# 使用示例
if __name__ == "__main__":
    model_name = "Qwen2.5-7B-Instruct-demo"

    print("列出所有模型：")
    list_all_models()

    print("\n" + "="*50)
    # 假设我们要把版本 1 推送到生产环境
    # promote_model_to_production(
    #     model_name=model_name,
    #     version=1,
    #     description="生产环境模型：2024年优化版"
    # )

    print("\n获取当前生产模型：")
    prod_model = get_production_model(model_name)
    if prod_model:
        print(f"  模型名: {prod_model.name}")
        print(f"  版本: {prod_model.version}")
        print(f"  阶段: {prod_model.current_stage}")
        print(f"  Run ID: {prod_model.run_id}")
    else:
        print("  暂无生产模型")

# 文件路径：metrics_prometheus.py
from prometheus_client import (
    Counter, Histogram, Gauge,
    start_http_server, generate_latest
)
import time
import random

# ============================================
# 定义指标
# ============================================

# Counter：只增不减的计数器
REQUEST_COUNT = Counter(
    "llm_requests_total",
    "Total number of LLM requests",
    ["model", "variant", "status"]  # 标签
)

# Histogram：统计分布（比如延迟）
REQUEST_LATENCY = Histogram(
    "llm_request_latency_seconds",
    "LLM request latency",
    ["model", "variant"],
    buckets=[0.1, 0.5, 1, 2, 5, 10, 30, 60]  # 分桶
)

TTFT_HISTOGRAM = Histogram(
    "llm_ttft_seconds",
    "Time to first token",
    ["model", "variant"],
    buckets=[0.05, 0.1, 0.2, 0.5, 1, 2, 5]
)

# Gauge：可增可减的仪表盘
ACTIVE_CONNECTIONS = Gauge(
    "llm_active_connections",
    "Number of active connections",
    ["model"]
)

GPU_UTILIZATION = Gauge(
    "llm_gpu_utilization_percent",
    "GPU utilization",
    ["gpu_id", "model"]
)

GPU_MEMORY_USED = Gauge(
    "llm_gpu_memory_used_bytes",
    "GPU memory used",
    ["gpu_id", "model"]
)

TOKENS_PER_SECOND = Gauge(
    "llm_tokens_per_second",
    "Tokens generated per second",
    ["model"]
)

# ============================================
# 模拟服务
# ============================================

def process_request(model: str, variant: str):
    """模拟处理一次请求，记录指标"""
    start_time = time.time()

    # 活跃连接数 +1
    ACTIVE_CONNECTIONS.labels(model=model).inc()

    try:
        # 模拟 TTFT（第一个 token 的时间）
        ttft = random.uniform(0.05, 0.3)
        TTFT_HISTOGRAM.labels(model=model, variant=variant).observe(ttft)

        # 模拟处理时间
        processing_time = random.uniform(0.5, 3.0)
        time.sleep(processing_time)

        # 模拟成功
        REQUEST_COUNT.labels(
            model=model, variant=variant, status="success"
        ).inc()

        # 记录延迟
        latency = time.time() - start_time
        REQUEST_LATENCY.labels(model=model, variant=variant).observe(latency)

        # 模拟 TPS
        tps = random.uniform(50, 200)
        TOKENS_PER_SECOND.labels(model=model).set(tps)

    except Exception:
        # 模拟错误
        REQUEST_COUNT.labels(
            model=model, variant=variant, status="error"
        ).inc()
        raise
    finally:
        # 活跃连接数 -1
        ACTIVE_CONNECTIONS.labels(model=model).dec()


def update_gpu_metrics():
    """模拟更新 GPU 指标"""
    # 模拟 GPU 0 的指标
    gpu0_util = random.uniform(40, 90)
    gpu0_mem = random.uniform(10, 20) * 1024**3  # 10-20 GB
    GPU_UTILIZATION.labels(gpu_id="0", model="Qwen2.5-7B").set(gpu0_util)
    GPU_MEMORY_USED.labels(gpu_id="0", model="Qwen2.5-7B").set(gpu0_mem)


# ============================================
# 启动服务
# ============================================

if __name__ == "__main__":
    # 启动 Prometheus metrics 端点
    start_http_server(8081)
    print("Prometheus metrics server started on port 8081")
    print("Metrics available at http://localhost:8081/metrics")

    # 模拟流量
    models = ["Qwen2.5-7B-Instruct", "Llama-3.1-8B-Instruct"]
    variants = ["control", "treatment-v1"]

    while True:
        model = random.choice(models)
        variant = random.choice(variants)
        process_request(model, variant)
        update_gpu_metrics()
        time.sleep(random.uniform(0.1, 0.5))

# 文件路径：prometheus.yml
global:
  scrape_interval: 15s

scrape_configs:
 # 抓取 LLM 服务的指标
  - job_name: "llm-service"
    static_configs:
      - targets: ["localhost:8081"]
        labels:
          service: "llm-api"
          environment: "production"

  # 抓取节点指标（GPU、CPU、内存）
  - job_name: "node"
    static_configs:
      - targets: ["localhost:9100"]

alerting:
  alertmanagers:
    - static_configs:
        - targets: ["localhost:9093"]

# 告警规则
rule_files:
 - "alerts.yml"

# 文件路径：alerts.yml
groups:
  - name: llm_alerts
    interval: 30s
    rules:
     # 高错误率告警
      - alert: HighErrorRate
        expr: |
         sum(rate(llm_requests_total{status="error"}[5m]))
          /
          sum(rate(llm_requests_total[5m])) > 0.05
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "高错误率告警"
          description: "错误率超过 5%，当前值: {{ $value | humanizePercentage }}"

      # TTFT 过高告警
      - alert: SlowTTFT
        expr: |
         histogram_quantile(0.95, sum(rate(llm_ttft_seconds_bucket[5m])) by (le, model)) > 2
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "TTFT 过高"
          description: "模型 {{ $labels.model }} 的 95分位 TTFT 超过 2 秒"

      # GPU 使用率过低（资源浪费）
      - alert: LowGPUUtilization
        expr: llm_gpu_utilization_percent < 30
        for: 30m
        labels:
          severity: info
        annotations:
          summary: "GPU 使用率过低"
          description: "GPU {{ $labels.gpu_id }} 使用率 {{ $value }}%，可能资源浪费"

      # GPU 使用率过高（瓶颈风险）
      - alert: HighGPUUtilization
        expr: llm_gpu_utilization_percent > 95
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "GPU 使用率过高"
          description: "GPU {{ $labels.gpu_id }} 使用率 {{ $value }}%，可能成为瓶颈"

# 文件路径：drift_detection.py
import numpy as np
from scipy import stats
from dataclasses import dataclass
from typing import List, Tuple
import json

@dataclass
class DistributionStats:
    """分布统计信息"""
    mean: float
    std: float
    min: float
    max: float
    percentiles: dict


class DataDriftDetector:
    """数据漂移检测器"""

    def __init__(self, baseline_data: np.ndarray):
        """用基线数据初始化"""
        self.baseline_stats = self._compute_stats(baseline_data)
        self.baseline_data = baseline_data

    def _compute_stats(self, data: np.ndarray) -> DistributionStats:
        """计算数据的统计信息"""
        return DistributionStats(
            mean=float(np.mean(data)),
            std=float(np.std(data)),
            min=float(np.min(data)),
            max=float(np.max(data)),
            percentiles={
                "p10": float(np.percentile(data, 10)),
                "p25": float(np.percentile(data, 25)),
                "p50": float(np.percentile(data, 50)),
                "p75": float(np.percentile(data, 75)),
                "p90": float(np.percentile(data, 90))
            }
        )

    def detect(self, new_data: np.ndarray, threshold: float = 0.05) -> dict:
        """
        检测漂移
        返回：是否有漂移、p值、统计信息
        """
        # KS 检验：比较两个分布是否相同
        ks_statistic, ks_pvalue = stats.ks_2samp(self.baseline_data, new_data)

        # 计算新数据的统计
        new_stats = self._compute_stats(new_data)

        # 计算均值差异（百分比）
        mean_diff_pct = abs((new_stats.mean - self.baseline_stats.mean) / self.baseline_stats.mean * 100)

        # 判断是否漂移
        is_drift = ks_pvalue < threshold

        return {
            "is_drift": is_drift,
            "ks_statistic": float(ks_statistic),
            "ks_pvalue": float(ks_pvalue),
            "mean_diff_percent": float(mean_diff_pct),
            "baseline_stats": self.baseline_stats,
            "new_stats": new_stats,
            "threshold": threshold
        }


# 模拟：用户输入长度的漂移检测
def simulate_input_length_drift():
    # 基线数据：用户输入平均长度 50
    np.random.seed(42)
    baseline = np.random.normal(loc=50, scale=15, size=1000)
    baseline = np.clip(baseline, 5, 200)

    detector = DataDriftDetector(baseline)

    print("数据漂移检测示例")
    print("="*50)
    print(f"基线: 均值={detector.baseline_stats.mean:.2f}, 标准差={detector.baseline_stats.std:.2f}")
    print()

    # 场景 1：正常数据（无漂移）
    print("场景 1：正常数据")
    normal_data = np.random.normal(loc=50, scale=15, size=500)
    normal_data = np.clip(normal_data, 5, 200)
    result = detector.detect(normal_data)
    print(f"  新数据均值: {result['new_stats'].mean:.2f}")
    print(f"  KS p-value: {result['ks_pvalue']:.4f}")
    print(f"  漂移? {'是' if result['is_drift'] else '否'}")
    print()

    # 场景 2：有漂移（输入变长了）
    print("场景 2：输入变长了")
    drift_data = np.random.normal(loc=100, scale=20, size=500)
    drift_data = np.clip(drift_data, 5, 300)
    result = detector.detect(drift_data)
    print(f"  新数据均值: {result['new_stats'].mean:.2f}")
    print(f"  KS p-value: {result['ks_pvalue']:.4f}")
    print(f"  漂移? {'是' if result['is_drift'] else '否'}")
    print(f"  均值差异: {result['mean_diff_percent']:.1f}%")


if __name__ == "__main__":
    simulate_input_length_drift()

# 文件路径：semantic_cache.py
import numpy as np
from typing import Optional, Tuple
import time

class SemanticCache:
    """简单的语义缓存实现"""

    def __init__(self, threshold: float = 0.9):
        self.threshold = threshold
        self.queries = []  # 存储问题的向量
        self.answers = []  # 存储对应的答案
        self.timestamps = []  # 存储时间

    def _simulate_embedding(self, text: str) -> np.ndarray:
        """
        模拟 embedding，实际项目中用 OpenAI Embeddings、Sentence-Transformers 等
        """
        # 简单的哈希模拟，实际项目中用真实 embedding 模型
        np.random.seed(hash(text) % (2**32))
        return np.random.randn(128)  # 128 维向量

    def _cosine_similarity(self, a: np.ndarray, b: np.ndarray) -> float:
        """计算余弦相似度"""
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

    def get(self, query: str) -> Optional[str]:
        """查询缓存，找到相似的问题就返回答案"""
        query_vec = self._simulate_embedding(query)

        for i, (q_vec, answer, ts) in enumerate(zip(self.queries, self.answers, self.timestamps)):
            # 过期清理（超过 1 小时）
            if time.time() - ts > 3600:
                continue

            similarity = self._cosine_similarity(query_vec, q_vec)
            if similarity >= self.threshold:
                return answer

        return None

    def put(self, query: str, answer: str):
        """存入缓存"""
        query_vec = self._simulate_embedding(query)
        self.queries.append(query_vec)
        self.answers.append(answer)
        self.timestamps.append(time.time())

        # 简单的容量控制：最多存 1000 条
        if len(self.queries) > 1000:
            self.queries = self.queries[-1000:]
            self.answers = self.answers[-1000:]
            self.timestamps = self.timestamps[-1000:]


# 使用示例
def main():
    cache = SemanticCache(threshold=0.9)

    # 第一次请求
    query1 = "如何用 Python 读取 CSV 文件？"
    print(f"查询 1: {query1}")
    answer1 = cache.get(query1)
    if answer1:
        print(f"缓存命中: {answer1}")
    else:
        print("缓存未命中，调用模型...")
        answer1 = "可以用 pandas.read_csv('file.csv') 来读取 CSV 文件。"
        cache.put(query1, answer1)
        print(f"存入缓存: {answer1}")

    print()

    # 第二次：相似的问题
    query2 = "Python 怎么读 CSV？"
    print(f"查询 2: {query2}")
    answer2 = cache.get(query2)
    if answer2:
        print(f"缓存命中: {answer2}")
    else:
        print("缓存未命中，调用模型...")

    print()

    # 第三次：完全不同的问题
    query3 = "什么是机器学习？"
    print(f"查询 3: {query3}")
    answer3 = cache.get(query3)
    if answer3:
        print(f"缓存命中: {answer3}")
    else:
        print("缓存未命中，调用模型...")


if __name__ == "__main__":
    main()

# 文件路径：model_router.py
from typing import List, Dict, Any
from dataclasses import dataclass

@dataclass
class ModelConfig:
    name: str
    cost_per_1k_tokens: float
    max_tokens: int
    capabilities: List[str]  # 能力列表


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

    def __init__(self):
        self.models = {
            "small": ModelConfig(
                name="Qwen2.5-0.5B-Instruct",
                cost_per_1k_tokens=0.001,
                max_tokens=4096,
                capabilities=["chat", "simple_qa", "summarization"]
            ),
            "medium": ModelConfig(
                name="Qwen2.5-7B-Instruct",
                cost_per_1k_tokens=0.01,
                max_tokens=8192,
                capabilities=["chat", "qa", "coding", "reasoning"]
            ),
            "large": ModelConfig(
                name="Qwen2.5-72B-Instruct",
                cost_per_1k_tokens=0.1,
                max_tokens=32768,
                capabilities=["complex_qa", "advanced_coding", "deep_reasoning"]
            )
        }

    def _classify_query(self, query: str) -> str:
        """
        简单的问题分类（实际项目中可以用分类器）
        返回：small/medium/large
        """
        query_lower = query.lower()

        # 简单问题的关键词
        simple_keywords = [
            "你好", "hello", "谢谢", "hi", "再见",
            "今天天气", "几点", "日期",
            "帮忙翻译", "translate",
            "简单介绍", "什么是"
        ]

        # 复杂问题的关键词
        complex_keywords = [
            "代码", "写一个", "实现", "code", "python",
            "分析", "解释为什么", "推导", "证明",
            "比较", "对比", "区别",
            "详细", "深入", "复杂"
        ]

        # 检查是否是简单问题
        for keyword in simple_keywords:
            if keyword in query_lower:
                return "small"

        # 检查是否是复杂问题
        for keyword in complex_keywords:
            if keyword in query_lower:
                return "large"

        # 默认用中等模型
        return "medium"

    def route(self, query: str, user_tier: str = "free") -> str:
        """
        路由决策
        """
        # 先分类问题
        category = self._classify_query(query)

        # 付费用户可以用更好的模型
        if user_tier == "pro":
            upgrade_map = {
                "small": "small",
                "medium": "medium",
                "large": "large"
            }
            category = upgrade_map[category]
        elif user_tier == "enterprise":
            upgrade_map = {
                "small": "medium",
                "medium": "large",
                "large": "large"
            }
            category = upgrade_map.get(category, category)

        # 返回模型名
        return self.models[category].name

    def get_cost_estimate(self, model_name: str, input_tokens: int, output_tokens: int) -> float:
        """估算成本"""
        for model in self.models.values():
            if model.name == model_name:
                return (input_tokens + output_tokens) * model.cost_per_1k_tokens / 1000
        return 0.0


# 使用示例
def main():
    router = SmartModelRouter()

    test_queries = [
        "你好，我想了解一下",
        "什么是 Python？",
        "帮我写一个 Python 快速排序",
        "详细分析一下 Transformer 架构"
    ]

    print("智能路由示例")
    print("="*50)

    for query in test_queries:
        model = router.route(query, user_tier="free")
        cost = router.get_cost_estimate(model, 100, 200)
        print(f"问题: {query}")
        print(f"  路由到: {model}")
        print(f"  估算成本: ${cost:.4f}")
        print()

    # 不同用户等级
    print("不同用户等级对比")
    print("-"*50)
    query = "帮我写一个 Python 快速排序"
    for tier in ["free", "pro", "enterprise"]:
        model = router.route(query, user_tier=tier)
        print(f"{tier} 用户: {model}")


if __name__ == "__main__":
    main()

# 文件路径：fallback_handler.py
import time
from typing import Optional
from dataclasses import dataclass
from enum import Enum

class ServiceStatus(Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    FAILED = "failed"

@dataclass
class FallbackResponse:
    content: str
    is_fallback: bool
    fallback_reason: Optional[str] = None


class FallbackHandler:
    """降级处理器"""

    def __init__(self):
        self.error_count = 0
        self.last_error_time = 0
        self.circuit_open = False
        self.circuit_open_time = 0

        # 熔断器配置
        self.error_threshold = 5  # 连续 5 次错误就熔断
        self.circuit_timeout = 30  # 熔断 30 秒后尝试恢复

    def _call_primary_service(self, query: str) -> str:
        """模拟调用主服务"""
        # 这里替换成真实的模型调用
        return "这是主服务的回复"

    def _call_fallback_service(self, query: str) -> str:
        """模拟调用降级服务（更快但质量稍低）"""
        # 可以是更小的模型、缓存的回复、甚至是预设的模板
        return "抱歉，服务暂时繁忙，这是一个简化的回复。"

    def _call_static_fallback(self, query: str) -> str:
        """静态降级回复（最后一道防线）"""
        return (
            "非常抱歉，我们的服务目前遇到了一些问题。\n"
            "请稍后再试。有紧急问题请联系 support@runoob.com。"
        )

    def process_request(self, query: str) -> FallbackResponse:
        """处理请求，带完整的降级逻辑"""

        # 1. 检查熔断器是否打开
        if self.circuit_open:
            if time.time() - self.circuit_open_time > self.circuit_timeout:
                # 超时了，尝试半开状态，允许少量请求通过
                print("熔断器半开，尝试恢复...")
                self.circuit_open = False
            else:
                # 直接降级
                print("熔断器打开，直接降级")
                return FallbackResponse(
                    content=self._call_fallback_service(query),
                    is_fallback=True,
                    fallback_reason="circuit_open"
                )

        # 2. 尝试主服务
        try:
            result = self._call_primary_service(query)
            # 成功：重置错误计数
            self.error_count = 0
            return FallbackResponse(
                content=result,
                is_fallback=False
            )

        except Exception as e:
            print(f"主服务调用失败: {e}")
            self.error_count += 1
            self.last_error_time = time.time()

            # 3. 检查是否需要熔断
            if self.error_count >= self.error_threshold:
                print("错误次数过多，打开熔断器")
                self.circuit_open = True
                self.circuit_open_time = time.time()

            # 4. 尝试降级服务
            try:
                print("尝试调用降级服务")
                result = self._call_fallback_service(query)
                return FallbackResponse(
                    content=result,
                    is_fallback=True,
                    fallback_reason="primary_failed"
                )
            except Exception as e2:
                print(f"降级服务也失败: {e2}")

                # 5. 最后的静态回复
                return FallbackResponse(
                    content=self._call_static_fallback(query),
                    is_fallback=True,
                    fallback_reason="all_failed"
                )


# 使用示例
def main():
    handler = FallbackHandler()

    print("降级处理示例")
    print("="*50)

    # 正常请求
    print("正常请求:")
    resp = handler.process_request("你好")
    print(f"  回复: {resp.content}")
    print(f"  是降级? {resp.is_fallback}")
    print()

    # 模拟错误（实际项目中会真的抛出异常）
    # 这里我们简单演示逻辑
    print("降级策略总结：")
    print("  1. 优先用主服务")
    print("  2. 主服务失败 → 用小模型/缓存")
    print("  3. 连续失败 → 熔断，避免雪崩")
    print("  4. 熔断期间 → 快速返回降级回复")
    print("  5. 最后一道防线 → 静态回复")