AI工程管理与可解释性实战指南：从API设计到风险管理

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional
import time

app = FastAPI()

class GenerationRequest(BaseModel):
    prompt: str = Field(..., min_length=1, max_length=4000)
    max_tokens: int = Field(default=512, ge=1, le=4096)
    temperature: float = Field(default=0.7, ge=0, le=2)
    top_p: float = Field(default=0.9, ge=0, le=1)
    stream: bool = False
    user_id: Optional[str] = None

class GenerationResponse(BaseModel):
    id: str
    text: str
    usage: dict
    model: str
    created_at: int

@app.post("/v1/generate", response_model=GenerationResponse)
async def generate(request: GenerationRequest):
    start = time.time()

    # 限流检查
    if request.user_id:
        check_rate_limit(request.user_id)

    try:
        # 生成
        if request.stream:
            return StreamingResponse(stream_generate(request))
        else:
            result = await model.generate(
                prompt=request.prompt,
                max_tokens=request.max_tokens,
                temperature=request.temperature
            )

            return GenerationResponse(
                id=generate_id(),
                text=result.text,
                usage=result.usage,
                model="my-model-v1",
                created_at=int(time.time())
            )

    except ModelTimeout:
        raise HTTPException(503, "Model timeout")
    except ContentFilter:
        raise HTTPException(400, "Content blocked")

from fastapi.responses import StreamingResponse
import json

async def stream_generate(request):
    async for chunk in model.stream_generate(request.prompt):
        data = {
            "id": request.id,
            "choices": [{"delta": {"content": chunk}}],
            "done": False
        }
        yield f"data: {json.dumps(data)}\n\n"

    # 结束标记
    yield f"data: {json.dumps({'done': True})}\n\n"

@app.post("/v1/chat/stream")
async def chat_stream(request: GenerationRequest):
    return StreamingResponse(stream_generate(request), media_type="text/event-stream")

# 兼容 OpenAI 格式，方便用户直接切换
@app.post("/v1/chat/completions")
async def chat_completions(request: OpenAIRequest):
    # 兼容 OpenAI 的请求/响应格式
    response = await model.chat(
        messages=request.messages,
        model=request.model,
        temperature=request.temperature,
        stream=request.stream
    )

    if request.stream:
        return StreamingResponse(stream_openai_format(response))
    else:
        return format_openai_response(response)

from lime.lime_text import LimeTextExplainer

explainer = LimeTextExplainer(class_names=['negative', 'positive'])

# 解释单个预测
exp = explainer.explain_instance(
    text_instance="This movie is great",
    classifier_fn=model.predict_proba,
    num_features=10
)

# 可视化哪些词影响了预测
exp.show_in_notebook()

import shap

# Tree 模型
explainer = shap.TreeExplainer(model)
shap_values = explainer.shap_values(X_test)

# 全局特征重要性
shap.summary_plot(shap_values, X_test)

# 单个预测的解释
shap.force_plot(explainer.expected_value[1], shap_values[1][0], X_test.iloc[0])

import matplotlib.pyplot as plt
import seaborn as sns

def plot_attention(attention_weights, tokens):
    """可视化注意力权重"""
    plt.figure(figsize=(10, 8))
    sns.heatmap(
        attention_weights,
        xticklabels=tokens,
        yticklabels=tokens,
        cmap='YlOrRd',
        annot=True,
        fmt='.2f'
    )
    plt.xlabel('Key')
    plt.ylabel('Query')
    plt.title('Attention Weights')
    plt.show()

# 让 LLM 解释自己的推理过程
prompt = """
请逐步解释你是如何得到这个答案的：

问题：{}
你的答案：{}

请说明：
1. 你理解了什么
2. 你考虑了哪些因素
3. 为什么选择了这个答案
"""

def adversarial_train(model, x, y, epsilon=0.01):
    """对抗训练一步"""
    # 1. 生成对抗样本
    x.requires_grad = True
    output = model(x)
    loss = F.cross_entropy(output, y)
    loss.backward()

    x_adv = x + epsilon * x.grad.sign()
    x_adv = x_adv.detach()

    # 2. 用原始 + 对抗样本一起训练
    output_clean = model(x)
    output_adv = model(x_adv)

    loss = F.cross_entropy(output_clean, y) + F.cross_entropy(output_adv, y)
    return loss

def detect_ood(model, x, threshold=0.8):
    """检测超出分布的输入"""
    with torch.no_grad():
        output = model(x)
        confidence = F.softmax(output, dim=-1).max()

    if confidence < threshold:
        return True, "Low confidence: possible OOD"
    return False, "In distribution"

项目模板:
  阶段 1 - 数据:
    - 数据收集
    - 标注
    - 清洗
    - 评估

  阶段 2 - 模型:
    - 基线模型
    - 迭代优化
    - 离线评估

  阶段 3 - 工程:
    - API 设计
    - 性能优化
    - 监控

  阶段 4 - 上线:
    - 影子部署
    - A/B 测试
    - 全量发布

  阶段 5 - 持续:
    - 监控
    - 重训
    - 迭代

# 训练成本估算
def estimate_training_cost(
    model_params,          # 模型参数量
    dataset_size,          # 数据集大小
    epochs,                # 训练轮数
    gpu_type="A100",       # GPU 类型
    gpu_per_hour=2.5       # 美元/小时
):
    # 粗略估算 FLOPs
    flops_per_sample = 6 * model_params
    total_flops = flops_per_sample * dataset_size * epochs

    # A100 大约 300 TFLOPS
    gpu_flops = 300e12
    hours = total_flops / gpu_flops / 3600

    cost = hours * gpu_per_hour
    return {
        "estimated_hours": hours,
        "estimated_cost": cost,
        "gpu_type": gpu_type
    }

# 估算 7B 模型训练
result = estimate_training_cost(
    model_params=7e9,
    dataset_size=1e6,
    epochs=3
)
print(f"训练成本: ${result['estimated_cost']:.0f}")

# 成本对比
cost_comparison = {
    "GPT-4 API": {"per_1k_tokens": 0.03, "daily_calls": 50000},
    "GPT-3.5 API": {"per_1k_tokens": 0.002, "daily_calls": 50000},
    "本地 Llama-2-7B": {"per_1k_tokens": 0.0005, "daily_calls": 50000},
    "本地 Llama-2-7B INT4": {"per_1k_tokens": 0.0002, "daily_calls": 50000}
}

for name, config in cost_comparison.items():
    monthly_cost = config["per_1k_tokens"] * 500 * config["daily_calls"] * 30 / 1000
    print(f"{name}: ${monthly_cost:.0f}/月")

# 实例价格 $2.5/小时
# 运行 1.2 小时 → 计费 2 小时 = $5.0
# 运行 4.1 小时 → 计费 5 小时 = $12.5

import requests, boto3

def collect_openai_usage(api_key, start_date, end_date):
    """收集 OpenAI API 使用情况，按模型+操作聚合"""
    headers = {"Authorization": f"Bearer {api_key}"}
    resp = requests.get(
        "https://api.openai.com/v1/usage",
        headers=headers,
        params={"start_date": start_date, "end_date": end_date}
    )
    summary = {}
    for item in resp.json().get("data", []):
        key = f"{item.get('model','unknown')}_{item.get('operation_type','unknown')}"
        cost = item.get("n_generated_tokens", 0) * item.get("cost_per_1k_tokens", 0) / 1000
        summary[key] = summary.get(key, 0) + cost
    return summary

def collect_aws_billing(region, start_date, end_date):
    """收集 AWS 账单（按服务和用量类型）"""
    client = boto3.client('ce', region_name=region)
    return client.get_cost_and_usage(
        TimePeriod={'Start': start_date, 'End': end_date},
        Granularity='DAILY',
        Metrics=['UnblendedCost'],
        GroupBy=[{'Type': 'DIMENSION', 'Key': 'SERVICE'},
                 {'Type': 'DIMENSION', 'Key': 'USAGE_TYPE'}]
    )['ResultsByTime']

import hashlib, json, os, pickle, time
from functools import wraps
from datetime import datetime, timedelta
from threading import Lock

class CacheLock:
    """按缓存键加锁，避免缓存击穿（失效瞬间大量请求同时打到 API）"""
    def __init__(self):
        self.locks, self.global_lock = {}, Lock()
    def get(self, key):
        with self.global_lock:
            if key not in self.locks:
                self.locks[key] = Lock()
            return self.locks[key]

_cache_locks = CacheLock()

def cache_api_calls(cache_dir="api_cache", cache_minutes=30):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            cache_key = hashlib.md5(
                json.dumps({"args": args, "kwargs": kwargs}, sort_keys=True).encode()
            ).hexdigest()
            cache_file = os.path.join(cache_dir, f"{cache_key}.pkl")
            with _cache_locks.get(cache_key):       # 加锁防击穿
                if os.path.exists(cache_file):
                    with open(cache_file, 'rb') as f:
                        data = pickle.load(f)
                        if datetime.now() - data['timestamp'] < timedelta(minutes=cache_minutes):
                            return data['result']
                result = func(*args, **kwargs)
                os.makedirs(cache_dir, exist_ok=True)
                with open(cache_file, 'wb') as f:
                    pickle.dump({'result': result, 'timestamp': datetime.now()}, f)
                return result
        return wrapper
    return decorator

class GPUManager:
    def __init__(self, instance_type, region):
        self.instance_type = instance_type
        self.ec2 = boto3.client('ec2', region_name=region)
        self.instance_id = None
        self.last_activity = None

    def start_instance(self):
        if self.instance_id is None:
            resp = self.ec2.run_instances(
                ImageId='ami-12345678', InstanceType=self.instance_type,
                MinCount=1, MaxCount=1)
            self.instance_id = resp['Instances'][0]['InstanceId']
        self.last_activity = time.time()

    def stop_if_idle(self, idle_minutes=30):
        if self.instance_id and self.last_activity:
            idle = (time.time() - self.last_activity) / 60
            if idle >= idle_minutes:
                self.ec2.stop_instances(InstanceIds=[self.instance_id])
                self.instance_id = None

def is_spot_termination_warning():
    """检查 Spot 实例回收警告"""
    import urllib.request
    try:
        urllib.request.urlopen(
            'http://169.254.169.254/latest/meta-data/spot/termination-time')
        return True
    except urllib.error.HTTPError as e:
        return e.code != 404

from sklearn.ensemble import RandomForestRegressor
from sklearn.preprocessing import StandardScaler

class CostPredictor:
    def __init__(self):
        self.model = RandomForestRegressor(n_estimators=100)
        self.scaler = StandardScaler()

    def _features(self, day):
        return [day['total_tokens'], day['active_users'],
                day['avg_task_complexity'], day['is_weekend']]

    def train(self, history):
        X = self.scaler.fit_transform([self._features(d) for d in history[:-1]])
        y = [d['cost'] for d in history[1:]]
        self.model.fit(X, y)

    def predict_next_day(self, current):
        return self.model.predict(self.scaler.transform([self._features(current)]))[0]

class AIRiskAssessment:
    def __init__(self, model, test_data):
        self.model = model
        self.test_data = test_data

    def assess_all(self):
        return {
            "performance": self._assess_performance(),
            "fairness": self._assess_fairness(),
            "robustness": self._assess_robustness(),
            "privacy": self._assess_privacy(),
            "safety": self._assess_safety()
        }

    def _assess_fairness(self):
        """公平性评估"""
        results = {}
        for group in self.test_data['sensitive_attr'].unique():
            subset = self.test_data[self.test_data['sensitive_attr'] == group]
            pred = self.model.predict(subset)
            results[group] = {
                'positive_rate': (pred == 1).mean(),
                'accuracy': accuracy_score(subset['label'], pred)
            }

        # 计算差异
        rates = [r['positive_rate'] for r in results.values()]
        max_disparity = max(rates) - min(rates)
        return {'by_group': results, 'max_disparity': max_disparity}

    def _assess_robustness(self):
        """鲁棒性评估"""
        # 对抗样本测试
        x_adv = generate_adversarial(self.test_data.X, epsilon=0.01)
        clean_acc = accuracy_score(self.test_data.y, self.model.predict(self.test_data.X))
        adv_acc = accuracy_score(self.test_data.y, self.model.predict(x_adv))
        return {'clean_accuracy': clean_acc, 'adversarial_accuracy': adv_acc}

# 使用 DeepSeek-Coder
from transformers import AutoModelForCausalLM, AutoTokenizer

model = AutoModelForCausalLM.from_pretrained("deepseek-ai/deepseek-coder-6.7b-instruct")
tokenizer = AutoTokenizer.from_pretrained("deepseek-ai/deepseek-coder-6.7b-instruct")

prompt = """Write a Python function that implements binary search.
Include type hints and docstring."""

inputs = tokenizer(prompt, return_tensors="pt")
outputs = model.generate(**inputs, max_new_tokens=500)
code = tokenizer.decode(outputs[0])

import pandas as pd
import numpy as np

# 1. 数值特征
df['log_price'] = np.log1p(df['price'])
df['price_per_sqm'] = df['price'] / df['area']

# 2. 类别特征
df = pd.get_dummies(df, columns=['city'], prefix='city')

# 3. 时间特征
df['hour'] = df['timestamp'].dt.hour
df['day_of_week'] = df['timestamp'].dt.dayofweek
df['is_weekend'] = df['day_of_week'].isin([5, 6]).astype(int)

# 4. 文本特征
df['text_length'] = df['text'].str.len()
df['word_count'] = df['text'].str.split().str.len()

# 5. 交叉特征
df['price_x_rating'] = df['price'] * df['rating']

from sklearn.feature_selection import SelectKBest, f_classif, mutual_info_classif

# 统计方法
selector = SelectKBest(f_classif, k=10)
X_selected = selector.fit_transform(X, y)

# 模型方法
from sklearn.ensemble import RandomForestClassifier
model = RandomForestClassifier()
model.fit(X, y)
importance = pd.Series(model.feature_importances_, index=X.columns)
top_features = importance.nlargest(10).index

# Feature Store（Feast）
from feast import FeatureStore, Entity, FeatureView, FileSource
from feast.value_type import ValueType

# 定义实体
user = Entity(name="user_id", value_type=ValueType.INT64)

# 定义特征视图
user_features = FeatureView(
    name="user_features",
    entities=["user_id"],
    ttl=timedelta(days=30),
    schema=[
        Field(name="age", dtype=ValueType.INT64),
        Field(name="city", dtype=ValueType.STRING),
    ],
    online=True,
    source=user_source
)

# 在线获取特征
store = FeatureStore(repo_path=".")
features = store.get_online_features(
    features=["user_features:age", "user_features:city"],
    entity_rows=[{"user_id": 1}, {"user_id": 2}]
).to_dict()