Compile:把训练集变成可部署的程序

compile 到底编译了什么

每个 Predictor(Predict/CoT/ReAct 等)内部有几块可调状态:

compile 改的主要是前三个,剩余代码逻辑不变。

保存与加载

# 保存(JSON,人可读)
compiled.save("artifacts/rag_v3.json")

# 加载:要先构造相同 class,再 load state
rag = RAG()
rag.load("artifacts/rag_v3.json")

JSON 文件长什么样

{
  "generate_query.predict": {
    "signature": "context, question -> reasoning, search_query",
    "instructions": "Given relevant context and a question, write a search query...",
    "demos": [
      {"context": ["..."], "question": "...", "reasoning": "...", "search_query": "..."},
      ...
    ]
  },
  "generate_answer.predict": {...}
}

这个文件可以 diff、可以 code review——prompt 工程终于变成了可 PR 的资产。

多版本管理

import hashlib, json, datetime

def save_versioned(mod, score):
    ts = datetime.datetime.now().strftime("%Y%m%d_%H%M")
    path = f"artifacts/rag_{ts}_f1_{score:.3f}.json"
    mod.save(path)
    meta = {
        "file": path,
        "score": score,
        "git": subprocess.check_output(["git", "rev-parse", "HEAD"]).decode().strip(),
        "trainset_hash": hashlib.md5(str(trainset).encode()).hexdigest(),
        "optimizer": "MIPROv2",
        "lm": str(dspy.settings.lm),
    }
    with open(path + ".meta.json", "w") as f:
        json.dump(meta, f, indent=2, ensure_ascii=False)

再配个软链 artifacts/rag_current.json → rag_20260430_1140_f1_0.912.json,上线和回滚就是改软链。

部署到 FastAPI

from fastapi import FastAPI
from pydantic import BaseModel
import dspy

class Query(BaseModel):
    question: str

class Answer(BaseModel):
    answer: str
    reasoning: str | None = None

app = FastAPI()

@app.on_event("startup")
def load():
    dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))
    app.state.rag = RAG()
    app.state.rag.load("artifacts/rag_current.json")

@app.post("/ask")
async def ask(q: Query) -> Answer:
    pred = app.state.rag(question=q.question)
    return Answer(answer=pred.answer, reasoning=getattr(pred, "reasoning", None))

用 uvicorn 起服务后就是普通 HTTP API,前端/其他服务照常接。

异步执行

# 2.5+ Module 原生支持 acall
@app.post("/ask")
async def ask(q: Query) -> Answer:
    pred = await app.state.rag.acall(question=q.question)
    return Answer(answer=pred.answer)

async 模式下并发 QPS 能从几十跳到几百。

批量推理

from dspy import evaluate

outputs = rag.batch(examples=[Example(question=q).with_inputs("question") for q in questions],
                      num_threads=16)

8000 条问题 15 分钟跑完,比 for 循环快 20 倍。

缓存配置

lm = dspy.LM(
    "openai/gpt-4o-mini",
    cache=True,                  # 开内存 + 磁盘缓存(默认)
)

# 想禁用(比如在评估里)
with dspy.context(lm=dspy.LM("openai/gpt-4o-mini", cache=False)):
    evaluate(...)

自定义缓存路径

export DSPY_CACHEDIR=/mnt/cache/dspy

多进程部署共享同一个 DSPY_CACHEDIR,能避免重复调用。

观测与埋点

# 1) settings 级别的 token 计数
dspy.settings.configure(track_usage=True)
pred = rag(question="...")
print(pred.get_lm_usage())  # {'openai/gpt-4o-mini': {'prompt_tokens': 812, 'completion_tokens': 102}}

# 2) Callback 埋点
from dspy.utils.callback import BaseCallback

class LatencyCB(BaseCallback):
    def on_module_end(self, call_id, outputs, exception):
        log.info("call %s duration=%s", call_id, outputs.get("duration_ms"))

dspy.settings.configure(callbacks=[LatencyCB()])

A/B 上线

编译产物是 JSON,做 A/B 只需加载不同文件:

class RAGRouter:
    def __init__(self, a_path, b_path):
        self.a = RAG(); self.a.load(a_path)
        self.b = RAG(); self.b.load(b_path)

    def __call__(self, question, user_id):
        mod = self.b if hash(user_id) % 100 < 10 else self.a   # 10% 流量到 B
        return mod(question=question)

编译产物的 CI 校验

# tests/test_rag.py
def test_rag_regression():
    rag = RAG()
    rag.load("artifacts/rag_current.json")

    eval_fn = dspy.Evaluate(devset=regression_set, metric=metric)
    score = eval_fn(rag)
    assert score >= 0.88, f"分数 {score} 低于底线"

每次改 Module 代码 → 重编译 → CI 跑 regression → 没达标不让合并。

常见坑

坑	症状	解法
加载报 key 缺失	改了 Module 字段名	版本号锁死 + 升级脚本改 JSON key
生产比训练分数差	测试集数据分布变了	每周跑一次线上采样 eval,触发重编译
启动慢	加载 JSON + warmup LM 客户端	健康检查用 /ready,先 dry-run 一次
并发调 LLM 撞 rate-limit	429 错误	配置 LM 的 retry_policy + 限流 semaphore

本章小结