告别Demo！聊聊AI技术从实验环境走向业务生产环境全过程

```python
# 典型的Demo代码，问题一堆
model = load_model("gpt-model")  # 模型路径硬编码
api_key = "sk-xxxxx"  # 密钥直接写代码里，安全大坑

def answer(question):
    prompt = f"请回答问题: {question}"
    result = model.generate(prompt)  # 没有超时控制
    return result  # 没有任何异常处理
```

```python
import os
import time
import logging
import hashlib
from dataclasses import dataclass, field
from typing import Optional
from functools import lru_cache

logger = logging.getLogger(__name__)


@dataclass
class LLMServiceConfig:
    """LLM服务的全部配置，从环境变量或配置中心读取"""
    model_name: str = os.getenv("LLM_MODEL_NAME", "default-model")
    api_key: str = os.getenv("LLM_API_KEY", "")  # 密钥不落代码
    max_retries: int = int(os.getenv("LLM_MAX_RETRIES", "3"))
    timeout_seconds: int = int(os.getenv("LLM_TIMEOUT", "30"))
    fallback_enabled: bool = os.getenv("LLM_FALLBACK", "true").lower() == "true"
    max_concurrent: int = int(os.getenv("LLM_MAX_CONCURRENT", "10"))


class LLMService:
    """生产级LLM调用封装: 重试/超时/降级/缓存/监控埋点"""

    def __init__(self, config: LLMServiceConfig):
        self.config = config
        self.client = self._init_client()
        self.metrics = {"total_calls": 0, "success": 0, "failures": 0, "timeouts": 0}

    def _init_client(self):
        """延迟初始化客户端，失败不阻塞服务启动"""
        try:
            # 实际场景这里会初始化 vLLM/OpenAI 兼容客户端
            return {"api_key": self.config.api_key, "model": self.config.model_name}
        except Exception as e:
            logger.error(f"LLM客户端初始化失败: {e}")
            raise RuntimeError("LLM服务不可用") from e

    def answer(self, question: str, use_cache: bool = True) -> dict:
        """带完整防护的问答接口"""
        self.metrics["total_calls"] += 1
        start_time = time.time()

        # 1. 入参校验，别让垃圾数据进 pipeline
        if not question or not question.strip():
            return self._error_response(400, "问题不能为空")

        # 2. 缓存命中检测，省钱省时间
        cache_key = hashlib.md5(question.encode()).hexdigest()
        if use_cache:
            cached = self._get_cache(cache_key)
            if cached:
                logger.info(f"缓存命中: {cache_key[:8]}")
                return cached

        # 3. 带重试的模型调用
        result = self._call_with_retry(question)

        # 4. 失败降级
        if result is None:
            if self.config.fallback_enabled:
                logger.warning("模型调用失败，走降级规则")
                return self._fallback_response(question)
            self.metrics["failures"] += 1
            return self._error_response(503, "AI服务暂时不可用")

        # 5. 成功路径: 记缓存、记指标
        response = {
            "answer": result,
            "model": self.config.model_name,
            "latency_ms": round((time.time() - start_time) * 1000, 2),
        }
        self.metrics["success"] += 1
        if use_cache:
            self._set_cache(cache_key, response)
        return response

    def _call_with_retry(self, question: str) -> Optional[str]:
        """指数退避重试"""
        for attempt in range(self.config.max_retries):
            try:
                # 实际场景: self.client.chat.completions.create(...)
                return self._do_generate(question)
            except TimeoutError:
                self.metrics["timeouts"] += 1
                logger.warning(f"第{attempt + 1}次调用超时")
            except Exception as e:
                logger.error(f"第{attempt + 1}次调用异常: {e}")

            # 指数退避: 1s, 2s, 4s
            time.sleep(2 ** attempt)
        return None

    def _do_generate(self, question: str) -> str:
        """实际的模型调用，带超时控制"""
        # 模拟实现，实际用 openai/vLLM 客户端
        time.sleep(0.1)
        return f"关于「{question}」的回答内容"

    def _get_cache(self, key: str) -> Optional[dict]:
        """实际项目用 Redis，这里简化"""
        return None

    def _set_cache(self, key: str, value: dict):
        """实际项目用 Redis + TTL"""
        pass

    def _fallback_response(self, question: str) -> dict:
        """降级: 返回规则引擎结果或标准话术"""
        return {
            "answer": "当前AI服务繁忙，已为您转接人工客服。",
            "degraded": True,
            "model": "fallback_rules",
        }

    def _error_response(self, code: int, message: str) -> dict:
        return {"error": {"code": code, "message": message}}


# 依赖注入式初始化，方便测试时替换
def create_llm_service() -> LLMService:
    return LLMService(LLMServiceConfig())
```

```python
import json
import statistics
from datetime import datetime
from typing import Callable


class Evaluator:
    """AI应用效果评估器"""

    def __init__(self, llm_client, golden_dataset: list):
        """
        golden_dataset 格式:
        [
            {"question": "退货政策是什么", 
             "reference_answer": "7天无理由退货...",
             "category": "售后"},
            ...
        ]
        """
        self.llm = llm_client
        self.golden_dataset = golden_dataset

    def run_evaluation(self, answer_fn: Callable, categories: list = None) -> dict:
        """
        answer_fn: 被评估的问答函数
        categories: 只评估指定分类，None则全量评估
        """
        dataset = self.golden_dataset
        if categories:
            dataset = [d for d in dataset if d["category"] in categories]

        results = []
        for item in dataset:
            start = datetime.now()
            actual_answer = answer_fn(item["question"])
            latency_ms = (datetime.now() - start).total_seconds() * 1000

            score = self._score_answer(
                question=item["question"],
                reference=item["reference_answer"],
                actual=actual_answer,
            )
            results.append({
                "question": item["question"],
                "category": item["category"],
                "score": score,
                "latency_ms": round(latency_ms, 1),
            })

        return self._summarize(results)

    def _score_answer(self, question: str, reference: str, actual: str) -> float:
        """LLM-as-Judge 打分: 0到1"""
        prompt = (
            "你是评估专家。请对比标准答案和待评答案，"
            "从准确性、完整性、相关性三个维度打分。\n\n"
            f"问题: {question}\n"
            f"标准答案: {reference}\n"
            f"待评答案: {actual}\n\n"
            "请只输出一个0到1之间的小数分数，"
            "1表示完全正确，0表示完全错误。"
        )
        try:
            response = self.llm.chat(prompt)
            score = float(response.strip())
            return max(0.0, min(1.0, score))
        except (ValueError, TypeError):
            return 0.0  # 解析失败按0分处理

    def _summarize(self, results: list) -> dict:
        """汇总评估结果"""
        scores = [r["score"] for r in results]
        latencies = [r["latency_ms"] for r in results]

        # 按分类统计
        by_category = {}
        for r in results:
            cat = r["category"]
            if cat not in by_category:
                by_category[cat] = []
            by_category[cat].append(r["score"])

        # 找出低分case，方便针对性优化
        bad_cases = [r for r in results if r["score"] < 0.6]

        return {
            "evaluated_at": datetime.now().isoformat(),
            "total_cases": len(results),
            "avg_score": round(statistics.mean(scores), 3) if scores else 0,
            "pass_rate": round(
                sum(1 for s in scores if s >= 0.8) / len(scores), 3
            ) if scores else 0,
            "p50_latency_ms": round(statistics.median(latencies), 1) if latencies else 0,
            "category_scores": {
                cat: round(statistics.mean(sc), 3)
                for cat, sc in by_category.items()
            },
            "bad_case_count": len(bad_cases),
            "bad_cases_preview": bad_cases[:5],
        }
```

```python
import time
import threading


class InferenceHealthChecker:
    """推理服务深度健康检查"""

    PROBE_PROMPT = "回复OK两个字母即可"

    def __init__(self, infer_client, check_interval=30):
        self.client = infer_client
        self.check_interval = check_interval
        self.consecutive_failures = 0
        self.max_failures = 3

    def deep_check(self) -> dict:
        """深度健康检查: 发送真实推理探针"""
        start = time.time()
        try:
            response = self.client.generate(self.PROBE_PROMPT, timeout=10)
            latency = round((time.time() - start) * 1000, 1)

            # 探针验证: 能回且延迟正常才算健康
            is_healthy = (
                response is not None
                and len(response) > 0
                and latency < 3000  # 探针延迟超3s说明过载
            )

            if is_healthy:
                self.consecutive_failures = 0
            else:
                self.consecutive_failures += 1

            return {
                "healthy": is_healthy,
                "latency_ms": latency,
                "consecutive_failures": self.consecutive_failures,
            }

        except Exception as e:
            self.consecutive_failures += 1
            return {
                "healthy": False,
                "error": str(e)[:100],
                "consecutive_failures": self.consecutive_failures,
            }

    def start_monitoring(self):
        """后台监控线程: 连续失败触发自动重启"""

        def _monitor_loop():
            while True:
                result = self.deep_check()
                if result["consecutive_failures"] >= self.max_failures:
                    # 触发实例摘除和自动重启流程
                    self._trigger_restart()
                time.sleep(self.check_interval)

        thread = threading.Thread(target=_monitor_loop, daemon=True)
        thread.start()

    def _trigger_restart(self):
        """摘除节点并重启(实际会调用K8s API)"""
        # 实际场景: 调用K8s删除Pod，让Deployment自动拉起新实例
        print("健康检查连续失败，触发实例重启")
```