AI模型工程与隐私实战指南：从版本管理到QLoRA

import mlflow
import mlflow.sklearn

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

with mlflow.start_run():
    # 记录参数
    mlflow.log_params({
        "model_type": "random_forest",
        "n_estimators": 200,
        "max_depth": 10
    })

    # 训练
    model = RandomForestRegressor(n_estimators=200, max_depth=10)
    model.fit(X_train, y_train)

    # 记录指标
    mlflow.log_metrics({
        "rmse": rmse,
        "r2": r2_score
    })

    # 记录模型（自动版本化）
    mlflow.sklearn.log_model(model, "model")

    # 记录训练数据信息
    mlflow.log_artifact("data_info.json")

from mlflow import MlflowClient

client = MlflowClient()

# 注册模型
client.create_registered_model("house_price_predictor")

# 添加版本
client.create_model_version(
    name="house_price_predictor",
    source="mlruns/1/abc123/artifacts/model",
    tags={"dataset_version": "v2.1"}
)

# 阶段转换
client.transition_model_version_stage(
    name="house_price_predictor",
    version=3,
    stage="Production"  # None → Staging → Production → Archived
)

# 模型版本应该记录
model_metadata:
  version: "v2.1.3"
  training_data:
    source: "s3://data/houses.csv"
    hash: "sha256:abc123..."
    size: 100000
    date_range: "2020-01-01 to 2024-12-31"

  training_config:
    framework: "pytorch"
    hyperparameters:
      learning_rate: 1e-4
      batch_size: 64
      epochs: 100

  evaluation:
    dataset: "test_set_v2"
    metrics:
      accuracy: 0.92
      f1: 0.89

  deployment:
    endpoint: "https://api.example.com/v1/predict"
    canary: false

# 搜索模型
versions = client.search_model_versions("name='house_price_predictor'")

for v in versions:
    print(f"Version {v.version}: {v.current_stage}, tags={v.tags}")

# 获取 Production 版本
prod_versions = client.get_latest_versions(
    "house_price_predictor",
    stages=["Production"]
)

# 加载特定版本
import mlflow.pyfunc
model = mlflow.pyfunc.load_model(
    model_uri="models:/house_price_predictor/Production"
)
# 或按版本号
model = mlflow.pyfunc.load_model(
    model_uri="models:/house_price_predictor/3"
)

from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
from trl import SFTTrainer, SFTConfig

# 1. 4-bit 量化加载
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",              # NF4 量化（推荐）
    bnb_4bit_compute_dtype=torch.bfloat16,   # 计算精度
    bnb_4bit_use_double_quant=True           # 二次量化
)

model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-2-7b",
    quantization_config=bnb_config,
    device_map="auto"
)

# 2. 准备 k-bit 训练
model = prepare_model_for_kbit_training(model)

# 3. LoRA 配置
lora_config = LoraConfig(
    r=16,                    # 秩
    lora_alpha=32,           # 缩放因子
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj"],
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM"
)

model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# 输出: trainable params: 13M || all params: 6,758M || trainable%: 0.19%

# 4. 训练
tokenizer = AutoTokenizer.from_pretrained("meta-llama/Llama-2-7b")
tokenizer.pad_token = tokenizer.eos_token

training_config = SFTConfig(
    output_dir="./qlora_output",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    gradient_accumulation_steps=4,
    learning_rate=3e-4,
    bf16=True,
    save_strategy="epoch",
    dataset_text_field="text"
)

trainer = SFTTrainer(
    model=model,
    train_dataset=dataset,
    peft_config=lora_config,
    tokenizer=tokenizer,
    args=training_config
)

trainer.train()

# 5. 保存适配器
model.save_pretrained("./qlora_adapter")

# 方法一：适配器 + 量化模型
from peft import PeftModel
from transformers import AutoModelForCausalLM

base_model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-2-7b",
    quantization_config=bnb_config,
    device_map="auto"
)
model = PeftModel.from_pretrained(base_model, "./qlora_adapter")

# 方法二：合并后推理（更快）
merged_model = model.merge_and_unload()
merged_model.save_pretrained("./merged_model")

from opacus import PrivacyEngine

model = AutoModelForCausalLM.from_pretrained(...)
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4)

# 初始化隐私引擎
privacy_engine = PrivacyEngine()

model, optimizer, dataloader = privacy_engine.make_private(
    module=model,
    optimizer=optimizer,
    dataloader=dataloader,
    noise_multiplier=1.0,    # 噪声大小
    max_grad_norm=1.0,       # 梯度裁剪
    target_epsilon=8.0,      # 目标 ε
    target_delta=1e-5,       # δ
    epochs=10
)

# 正常训练
for epoch in range(10):
    for batch in dataloader:
        loss = model(batch).loss
        loss.backward()
        optimizer.step()
        optimizer.zero_grad()

    # 查看当前 ε
    epsilon = privacy_engine.get_epsilon(1e-5)
    print(f"Epoch {epoch}: ε = {epsilon:.2f}")

from transformers import AutoModelForCausalLM, AutoTokenizer

model = AutoModelForCausalLM.from_pretrained(
    "microsoft/phi-2",
    torch_dtype=torch.float16,
    device_map="auto",
    trust_remote_code=True
)

tokenizer = AutoTokenizer.from_pretrained("microsoft/phi-2", trust_remote_code=True)

# 推理
prompt = "Write a Python function to check if a number is prime"
inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
outputs = model.generate(**inputs, max_new_tokens=200)
print(tokenizer.decode(outputs[0]))

from neo4j import GraphDatabase
from langchain_community.graphs import Neo4jGraph

# 1. 构建知识图谱
graph = Neo4jGraph(
    url="bolt://localhost:7687",
    username="neo4j",
    password="password"
)

# 2. 从文档抽取实体和关系
def extract_and_store(text):
    # LLM 抽取实体
    entities = llm.extract_entities(text)
    # LLM 抽取关系
    relations = llm.extract_relations(text)

    # 存入 Neo4j
    for entity in entities:
        graph.query(f"MERGE (n:{entity.type} {{name: '{entity.name}'}})")

    for rel in relations:
        graph.query(f"""
            MATCH (a {{name: '{rel.source}'}}), (b {{name: '{rel.target}'}})
            MERGE (a)-[:{rel.type}]->(b)
        """)

# 3. 图谱检索
def graph_retrieve(question):
    # 提取问题中的实体
    entities = llm.extract_entities(question)

    # Cypher 查询（多跳）
    cypher = f"""
    MATCH (n)-[r*1..3]-(m)
    WHERE n.name IN {entities}
    RETURN n, r, m
    LIMIT 50
    """
    result = graph.query(cypher)

    # 整理为上下文
    context = format_graph_result(result)
    return context

# 4. 生成答案
answer = llm.generate(question, context=graph_retrieve(question))

class FusionRAG:
    def __init__(self):
        self.vector_search = VectorSearch()
        self.bm25_search = BM25Search()
        self.graph_search = GraphSearch()
        self.reranker = CrossEncoder('BAAI/bge-reranker-base')

    def retrieve(self, query, top_k=5):
        # 多路召回
        vector_results = self.vector_search.search(query, top_k=20)
        bm25_results = self.bm25_search.search(query, top_k=20)
        graph_results = self.graph_search.search(query, top_k=20)

        # 融合（RRF: Reciprocal Rank Fusion）
        all_docs = {}
        for rank, doc in enumerate(vector_results):
            all_docs[doc.id] = all_docs.get(doc.id, 0) + 1 / (rank + 1)
        for rank, doc in enumerate(bm25_results):
            all_docs[doc.id] = all_docs.get(doc.id, 0) + 1 / (rank + 1)
        for rank, doc in enumerate(graph_results):
            all_docs[doc.id] = all_docs.get(doc.id, 0) + 1 / (rank + 1)

        # 排序
        sorted_docs = sorted(all_docs.items(), key=lambda x: x[1], reverse=True)[:20]

        # 重排序
        candidates = [get_doc(doc_id) for doc_id, _ in sorted_docs]
        rerank_scores = self.reranker.predict([(query, doc.text) for doc in candidates])

        # 返回 top_k
        ranked = sorted(zip(candidates, rerank_scores), key=lambda x: x[1], reverse=True)
        return [doc for doc, _ in ranked[:top_k]]

from sklearn.linear_model import LinearRegression, Ridge, Lasso
from sklearn.metrics import mean_squared_error, r2_score

# 线性回归
lr = LinearRegression()
lr.fit(X_train, y_train)

# Ridge（L2 正则化）
ridge = Ridge(alpha=1.0)
ridge.fit(X_train, y_train)

# Lasso（L1 正则化，特征选择）
lasso = Lasso(alpha=0.1)
lasso.fit(X_train, y_train)

# 评估
for name, model in [('LR', lr), ('Ridge', ridge), ('Lasso', lasso)]:
    pred = model.predict(X_test)
    print(f"{name}: RMSE={mean_squared_error(y_test, pred, squared=False):.4f}, R²={r2_score(y_test, pred):.4f}")

import xgboost as xgb
from sklearn.model_selection import cross_val_score

# XGBoost 回归
model = xgb.XGBRegressor(
    n_estimators=500,
    max_depth=6,
    learning_rate=0.01,
    subsample=0.8,
    colsample_bytree=0.8,
    random_state=42
)

# 交叉验证
scores = cross_val_score(model, X, y, cv=5, scoring='neg_root_mean_squared_error')
print(f"CV RMSE: {-scores.mean():.4f} (+/- {scores.std():.4f})")

model.fit(X_train, y_train)

from sklearn.metrics import (
    mean_squared_error, mean_absolute_error,
    r2_score, mean_absolute_percentage_error
)

# 常用指标
mse = mean_squared_error(y_true, y_pred)
rmse = mse ** 0.5
mae = mean_absolute_error(y_true, y_pred)
r2 = r2_score(y_true, y_pred)
mape = mean_absolute_percentage_error(y_true, y_pred)

# 业务指标（根据场景）
# 金融：偏向低 MAPE
# 推荐：偏向高 R²
# 预测：偏向低 RMSE

class OfflineEvaluator:
    def __init__(self, test_data, metrics):
        self.test_data = test_data
        self.metrics = metrics  # ['accuracy', 'f1', 'auc', ...]

    def evaluate(self, model):
        results = {}
        predictions = model.predict(self.test_data.X)
        true_labels = self.test_data.y

        for metric_name in self.metrics:
            if metric_name == 'accuracy':
                results[metric_name] = accuracy_score(true_labels, predictions)
            elif metric_name == 'f1':
                results[metric_name] = f1_score(true_labels, predictions, average='macro')
            elif metric_name == 'auc':
                probabilities = model.predict_proba(self.test_data.X)[:, 1]
                results[metric_name] = roc_auc_score(true_labels, probabilities)

        return results

    def compare_models(self, models):
        """对比多个模型"""
        all_results = {}
        for name, model in models.items():
            all_results[name] = self.evaluate(model)
        return pd.DataFrame(all_results).T

def stratified_evaluation(model, test_data, strata_column):
    """按子群体分别评估"""
    results = {}
    for stratum in test_data[strata_column].unique():
        subset = test_data[test_data[strata_column] == stratum]
        pred = model.predict(subset.X)
        results[stratum] = {
            'count': len(subset),
            'accuracy': accuracy_score(subset.y, pred),
            'f1': f1_score(subset.y, pred, average='macro')
        }
    return pd.DataFrame(results).T