用 Python 搭建人工智能系统的深度实践：从数据到上线的一条龙

用 Python 搭建人工智能系统的深度实践：从数据到上线的一条龙

> 这不是“装库即成功”的教程，而是一条能在真实业务里落地、可维护、可观测、可控成本的 AI 工程路径。目标：用 Python 打通数据工程 → 训练/微调 → 评测对齐 → 推理服务化 → 观测与治理，全链路可复制。

1. 问题刻画：先定义“可被学习”的目标函数

  • 业务刻画：明确输入分布、输出约束、误差成本（FP/ FN 哪个更贵）、上线后的反馈回路。
  • 数据策略：分 原始数据、标注数据、合成数据 三层；沉淀数据字典（Schema）、质量度量（覆盖度、重复率、毒性/偏见分数）。
  • 最小闭环：先做一个“可离线评测”的子任务（如 FAQ 检索/摘要），将其评测指标固化。

2. 数据工程：清洗、切片与可追溯

核心目标：可重跑、可审计、可切分 A/B。

  • 标准化：统一编码、去重、切句、脱敏；构建样本难度分布（困惑度/长度/领域标签）。
  • 数据切片（Data Slicing）：按领域、长度、噪声、时间、用户群体等维度建立 slice，评测时逐 slice 报告，便于诊断“平均好但局部差”的问题。
  • 管线化：用 Pydantic 定义样本 Schema，用 Polars/DuckDB 做大表处理；Airflow/Prefect 编排离线流程。
from pydantic import BaseModel, Field
class Sample(BaseModel):
    q: str = Field(min_length=2)
    a: str = Field(min_length=1)
    domain: str

3. 模型路线：检索增强（RAG）与微调（Finetune）何时取舍

  • RAG：数据敏捷、可解释、更新快；适合知识问答、合规内容、文档助手。
  • 微调：对齐特定风格/工具调用/多轮记忆稳定性；适合“长尾一贯性”与“低延迟批量离线生成”。
  • 组合拳：RAG 兜知识边界，微调固化风格与步骤；通过 工具调用（function calling） 接 DB/搜索/计算引擎。

4. 向量检索（RAG）工程要点

  • 向量化：sentence-transformers/text-embedding-*；中文建议使用多语言或中文专用模型。
  • 切片策略：按语义段落/标题层级切块，重叠窗口保证上下文连续。
  • 索引：开发期用 FAISS，生产可迁移 Milvus/PGVector/Elasticsearch。
  • 重排：BM25 + 向量混排，或小重排模型（monoT5）。
  • 上下文构建：模板化 Prompt，控制总 token、插入 citation 与来源。
# 简化的 RAG 检索示例
from sentence_transformers import SentenceTransformer
import faiss, numpy as np


model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
docs = ["……文档1……", "……文档2……"]
emb = model.encode(docs, normalize_embeddings=True)
index = faiss.IndexFlatIP(emb.shape[1]); index.add(np.array(emb))
query = "增值税发票作废条件？"
qv = model.encode([query], normalize_embeddings=True)
D, I = index.search(np.array(qv), k=3)
context = "\n\n".join(docs[i] for i in I[0])

5. 微调与对齐：从指令到偏好优化

  • 指令微调（SFT）：用高质量指令-回复对，快速“讲懂规矩”；配 LoRA/QLoRA 降低显存与成本。
  • 偏好优化（DPO/ORPO/KTO）：让模型“更像人”而不是“更像训练集”；样本来自标注对比或线下点击反馈。
  • 数据配方：高质量少量胜过低质海量；混合“通用指令 + 领域指令 + 工具调用 + 拒答示例”。
# 伪代码：PyTorch 训练主循环（SFT 思路）
for batch in loader:
    out = model(**tokenizer(batch["prompt+answer"], return_tensors="pt").to(device))
    loss = out.loss  # 已按标签 mask 仅在答案区计算
    loss.backward(); optimizer.step(); optimizer.zero_grad()

6. 评测：不做“主观感觉好”，而做“可复现实证”

  • 自动指标：RAG 用 Recall@k / MRR / nDCG；摘要用 ROUGE / BERTScore。
  • 对话任务：构造对抗集（越界、幻觉、逻辑链），打“安全、事实一致性、工具调用正确率”。
  • 人评闭环：抽样多评一致性（Cohen’s Kappa），将人评结果回灌至偏好优化数据池。
  • Slice 报告：逐领域/长度/时间切片输出对比，拒绝“平均分掩盖问题”。

7. 推理服务化与性能工程

  • 框架：FastAPI 作为 HTTP 边界，后端调用 vLLM/TGI/Ollama 或自建 PyTorch 推理；高 QPS 用 vLLM 动态批处理（PAGED attention）。
  • 上下文成本：长上下文模型配 检索压缩（LLMLingua）；频繁相似查询启用 缓存（prompt 缓存/embedding 缓存）。
  • 量化：AWQ/GPTQ/INT4/LLM.int8() 折中质量与吞吐；边缘端可使用 GGUF。
  • 弹性扩缩：HPA + 队列回压，限流/超时/重试三件套。
# FastAPI 推理边界（简化）
from fastapi import FastAPI
app = FastAPI()


@app.post("/chat")
def chat(req: dict):
    q = req["q"]
    ctx = build_context(q)  # RAG
    prompt = f"基于资料回答：\n{ctx}\n\n问题：{q}"
    ans = llm.generate(prompt, max_tokens=512)
    return {"answer": ans}

8. 观测、灰度与安全治理

  • 可观测性：统一 OpenTelemetry 追踪 + Prometheus 指标（QPS、P95 延迟、失败率、拒答率、平均上下文 token）。
  • 质量看板：离线评测与线上采样双轨；暴露“幻觉率、引用命中率、工具调用成功率”。
  • 灰度：分流策略（用户组/流量百分比/问题域）；遇回归一键回滚。
  • 安全：Prompt 注入防护（指令边界、系统提示加密/拼接）、输出审计（PII/不当内容过滤）、合规留痕（可追溯数据与模型版本）。

9. 成本控制：把钱花在“有效增益”的地方

  • 优先 RAG：把“知识更新”从“重新训练”转为“重新索引”。
  • 精调做小而精：只在“稳定风格/流程”上微调；其余交给检索与工具调用。
  • 缓存与复用：Embedding 共享向量库、提示模板与特征缓存、批量离线生成静态答案。
  • 分层硬件：训练与推理分层；推理选择更优性价比的显卡/CPU + 量化模型。

10. 参考落地蓝图（可裁剪）

  1. 数据层：DuckDB/对象存储 + Airflow 管线；Pydantic 统一 Schema。
  2. 索引层：FAISS → Milvus/PGVector；BM25 混排兜底。
  3. 模型层：基础大模型（开源/商用 API） + LoRA 微调（仅在必要处）。
  4. 服务层：FastAPI + vLLM/TGI；工具编排用 LangChain/LlamaIndex（或自研轻量执行器）。
  5. 评测层：离线基准 + 人评采样；逐 slice 报告。
  6. 运维层：OTel+Prometheus+Grafana；灰度、回滚、速率限制、审计留痕。

结语：让系统“可学习、可解释、可演进”

Python 的价值在于 把复杂的 AI 工程拆成可组合的积木：数据即产品、评测即文化、服务即平台。先用 RAG 让知识可更新，用小步微调固化风格与流程，用可观测性约束模型“说真话、做正事”。当你把数据质量和评测文化放在第一位，Python 生态的速度与弹性，才会真正转化为可持续的业务竞争力。
aileenbuha@gmail.com

评论 0