AI 企业规模化部署指南

# PoC 与生产的核心差异：以成本为例
# PoC 阶段（理想情况）
poc_cost = {
    "daily_requests": 100,
    "cost_per_request": 0.05,
    "monthly_cost": 100 * 0.05 * 30,  # $150/月
}

# 生产阶段（真实情况）
prod_cost = {
    "daily_requests": 50000,
    "cost_per_request": 0.08,  # 生产级提示词更长，成本更高
    "monthly_cost": 50000 * 0.08 * 30,  # $120,000/月
}

# 差距：800 倍
print(f"成本差距: {prod_cost['monthly_cost'] / poc_cost['monthly_cost']:.0f}x")

// 模型路由配置：根据请求复杂度选择模型
interface ModelRoute {
  name: string;
  provider: string;
  model: string;
  maxTokens: number;
  costPer1k: number;
}

const ROUTES: Record<string, ModelRoute> = {
  simple: {
    name: "轻量模型",
    provider: "openai",
    model: "gpt-4o-mini",
    maxTokens: 2000,
    costPer1k: 0.00015,
  },
  complex: {
    name: "旗舰模型",
    provider: "openai",
    model: "gpt-4o",
    maxTokens: 8000,
    costPer1k: 0.0025,
  },
  fallback: {
    name: "降级模型",
    provider: "openai",
    model: "gpt-4o-mini",
    maxTokens: 1000,
    costPer1k: 0.00015,
  },
};

function selectRoute(request: { complexity: number; urgency: number }): ModelRoute {
  if (request.complexity > 0.8) return ROUTES.complex;
  if (request.complexity > 0.4) return ROUTES.simple;
  return ROUTES.fallback;
}

# 单位经济效益计算
class UnitEconomics:
    def __init__(self):
        self.total_api_cost = 0.0
        self.total_revenue = 0.0
        self.total_interactions = 0
    
    def record_interaction(self, cost: float, revenue: float):
        self.total_api_cost += cost
        self.total_revenue += revenue
        self.total_interactions += 1
    
    def cost_per_interaction(self) -> float:
        if self.total_interactions == 0:
            return 0.0
        return self.total_api_cost / self.total_interactions
    
    def revenue_per_interaction(self) -> float:
        if self.total_interactions == 0:
            return 0.0
        return self.total_revenue / self.total_interactions
    
    def profit_margin(self) -> float:
        if self.total_revenue == 0:
            return -100.0
        return (self.total_revenue - self.total_api_cost) / self.total_revenue * 100
    
    def is_sustainable(self) -> bool:
        return self.profit_margin() > 0

# 使用示例
economics = UnitEconomics()
for _ in range(10000):
    economics.record_interaction(cost=0.05, revenue=2.00)

print(f"单次交互成本: {economics.cost_per_interaction():.4f}")
print(f"单次交互收入: {economics.revenue_per_interaction():.2f}")
print(f"利润率: {economics.profit_margin():.1f}%")
print(f"可持续性: {economics.is_sustainable()}"

# 数据脱敏管道：在向模型发送前清理敏感信息
import re
from typing import Dict, List

class DataSanitizer:
    """生产级数据脱敏器"""
    
    # 敏感信息模式
    PATTERNS = {
        "email": re.compile(r'[\w.+-]+@[\w-]+\.[\w.-]+'),
        "phone_cn": re.compile(r'1[3-9]\d{9}'),
        "id_card": re.compile(r'\d{17}[\dXx]'),
        "credit_card": re.compile(r'\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}'),
    }
    
    REPLACEMENTS = {
        "email": "[EMAIL_REDACTED]",
        "phone_cn": "[PHONE_REDACTED]",
        "id_card": "[ID_REDACTED]",
        "credit_card": "[CARD_REDACTED]",
    }
    
    def sanitize(self, text: str) -> Dict[str, any]:
        """脱敏并返回脱敏后的文本和脱敏记录"""
        redacted = text
        redaction_log = []
        
        for field, pattern in self.PATTERNS.items():
            matches = pattern.findall(redacted)
            if matches:
                redaction_log.append({
                    "field": field,
                    "count": len(matches),
                    "replacement": self.REPLACEMENTS[field],
                })
                redacted = pattern.sub(self.REPLACEMENTS[field], redacted)
        
        return {
            "text": redacted,
            "redactions": redaction_log,
            "has_sensitive_data": len(redaction_log) > 0,
        }

# 使用示例
sanitizer = DataSanitizer()
result = sanitizer.sanitize("请联系张三，邮箱 zhangsan@example.com，手机 13800138000")
print(f"脱敏后: {result['text']}")
# 输出: 请联系张三，邮箱 [EMAIL_REDACTED]，手机 [PHONE_REDACTED]

# AI 质量监控：漂移检测示例
import numpy as np
from scipy.stats import ks_2statistic

class DriftDetector:
    """检测输入/输出分布漂移"""
    
    def __init__(self, baseline_samples: list, threshold: float = 0.05):
        self.baseline = baseline_samples
        self.threshold = threshold
    
    def detect_drift(self, current_samples: list) -> dict:
        """使用 KS 检验检测分布漂移"""
        if len(current_samples) < 30:
            return {"drift": False, "reason": "样本不足"}
        
        # 提取特征（如文本长度、词数等）
        baseline_features = [len(s.split()) for s in self.baseline]
        current_features = [len(s.split()) for s in current_samples]
        
        # KS 检验
        stat, p_value = ks_2statistic(baseline_features, current_features)
        
        return {
            "drift": p_value < self.threshold,
            "statistic": stat,
            "p_value": p_value,
            "baseline_mean_length": np.mean(baseline_features),
            "current_mean_length": np.mean(current_features),
        }

# 使用示例
detector = DriftDetector(
    baseline_samples=["你好", "今天天气怎么样", "帮我写个邮件"] * 100,
    threshold=0.05
)
result = detector.detect_drift(["请分析这份合同的法律风险并给出修改建议"] * 100)
print(f"检测到漂移: {result['drift']}")

# A/B 测试框架：比较两个模型版本
import numpy as np
from scipy import stats

class ABTest:
    def __init__(self, confidence_level: float = 0.95):
        self.confidence = confidence_level
        self.group_a = []  # 对照组
        self.group_b = []  # 实验组
    
    def record(self, group: str, score: float):
        if group == "a":
            self.group_a.append(score)
        else:
            self.group_b.append(score)
    
    def analyze(self) -> dict:
        if len(self.group_a) < 30 or len(self.group_b) < 30:
            return {"conclusion": "样本不足", "need_more": True}
        
        # t 检验
        t_stat, p_value = stats.ttest_ind(self.group_a, self.group_b)
        
        mean_a = np.mean(self.group_a)
        mean_b = np.mean(self.group_b)
        improvement = (mean_b - mean_a) / mean_a * 100 if mean_a > 0 else 0
        
        significant = p_value < (1 - self.confidence)
        
        return {
            "group_a_mean": round(mean_a, 4),
            "group_b_mean": round(mean_b, 4),
            "improvement_pct": round(improvement, 2),
            "p_value": round(p_value, 6),
            "significant": significant,
            "conclusion": "B 版本显著优于 A" if (significant and improvement > 0) 
                          else "A 版本显著优于 B" if (significant and improvement < 0)
                          else "无显著差异",
        }

# 使用示例
test = ABTest()
# 模拟两组用户评分（1-5分）
for _ in range(500):
    test.record("a", np.random.normal(3.8, 0.8))
    test.record("b", np.random.normal(4.1, 0.7))

result = test.analyze()
print(result["conclusion"])

# 跨职能 AI 小队配置示例
ai_squad:
  name: "AI 客服小队"
  mission: "构建和维护 AI 客服系统"
  
  roles:
    - role: AI工程师
      count: 2
      responsibilities:
        - 模型选型和评估
        - 提示词设计和优化
        - 模型微调（需要时）
    
    - role: 后端工程师
      count: 2
      responsibilities:
        - API 开发和维护
        - 数据库设计
        - 系统集成
    
    - role: MLOps工程师
      count: 1
      responsibilities:
        - 模型部署和监控
        - CI/CD 流水线
        - 成本管理
    
    - role: 安全工程师
      count: 0.5  # 兼职
      responsibilities:
        - 安全审查
        - 合规审计
        - 渗透测试
    
    - role: 产品经理
      count: 1
      responsibilities:
        - 需求分析
        - 优先级排序
        - 用户反馈收集
    
    - role: 数据工程师
      count: 1
      responsibilities:
        - 数据管道
        - 特征工程
        - 训练数据管理

# AI 生产就绪检查清单

## 架构与设计
- [ ] 分层解耦架构（API 网关/路由/缓存/审核/监控独立）
- [ ] 降级策略（主模型不可用时的备用方案）
- [ ] 缓存策略（语义缓存/精确缓存混合）
- [ ] 模型路由（根据复杂度选择模型）
- [ ] 输入预处理（清洗/脱敏/分类）

## 性能与成本
- [ ] P99 延迟 < 5 秒（或业务要求）
- [ ] 成本基线建立（知道每千次请求的成本）
- [ ] 预算上限设定（日度/月度）
- [ ] 缓存命中率 > 30%
- [ ] 模型路由节省 > 30%

## 安全与合规
- [ ] 数据脱敏管道（邮箱/手机/身份证）
- [ ] 输出内容审核（敏感词/偏见检测）
- [ ] 合规审查（GDPR/CCPA/行业规范）
- [ ] 访问控制（RBAC/API Key）
- [ ] 审计日志（完整调用链记录）

## 运维与监控
- [ ] 指标监控（延迟/错误率/成本）
- [ ] 告警机制（阈值告警/升级机制）
- [ ] 日志系统（请求详情/错误堆栈）
- [ ] 质量监控（人工审核/用户反馈）
- [ ] 漂移检测（输入/输出分布监控）

## 团队与流程
- [ ] 跨职能团队（AI/后端/MLOps/安全/产品）
- [ ] 灰度发布流程（实验室→内部→灰度→全量）
- [ ] 回滚能力（5 分钟内回滚）
- [ ] 文档齐全（架构图/API 文档/运维手册）
- [ ] 应急预案（故障处理流程）