大模型产品经理实战：把"模糊需求"编译成可验收的工程契约

# evals/spec.py —— PM 维护的需求规格，算法与测试共用
from dataclasses import dataclass, field

@dataclass
class Requirement:
    name: str
    metric: str                  # 指标定义
    threshold: float             # 验收线
    cases_path: str              # 黄金集
    blocking: bool = True        # 不达标是否阻断发布
    notes: str = ""

SPECS = [
    Requirement("答案事实正确", "factual_acc", 0.92, "evals/gold/facts.jsonl"),
    Requirement("引用可追溯",   "citation_rate", 0.98, "evals/gold/cite.jsonl"),
    Requirement("拒答越界问题", "refusal_acc", 0.95, "evals/gold/refuse.jsonl"),
    Requirement("首字延迟",     "p95_ttft_ms", 1200, "evals/gold/perf.jsonl", blocking=False),
]

from pydantic import BaseModel, Field, ValidationError
import json

class Answer(BaseModel):
    conclusion: str = Field(max_length=200)              # 先给结论
    evidence: list[str] = Field(min_length=1, max_length=3)  # 必须带证据
    confidence: float = Field(ge=0, le=1)
    followup: str | None = None

def generate(llm, question: str, retries: int = 2) -> Answer:
    for _ in range(retries + 1):
        raw = llm.complete(
            f"问题:{question}\n按 schema 输出 JSON: {Answer.model_json_schema()}",
            temperature=0)
        try:
            return Answer(**json.loads(raw))
        except (json.JSONDecodeError, ValidationError):
            continue
    return Answer(conclusion="暂时无法可靠回答", evidence=["系统降级"],
                  confidence=0.0)                        # 明确降级，不编造

ROUTE = {
    "simple":  {"model": "small",  "max_tokens": 300,  "budget_ms": 800},
    "complex": {"model": "large",  "max_tokens": 2000, "budget_ms": 6000},
}

def route(question: str) -> dict:
    # 规则优先，可解释、可回归、零成本
    if len(question) < 60 and "为什么" not in question:
        return ROUTE["simple"]
    return ROUTE["complex"]

def handle(llm, question: str) -> Answer:
    cfg = route(question)
    ans = generate(llm.bind(cfg["model"], cfg["max_tokens"]), question)
    if ans.confidence < 0.6:
        ans.followup = "是否需要转人工确认？"     # 低置信度触发人工路径
    return ans

def canary_gate(old_metrics: dict, new_metrics: dict, specs: list) -> bool:
    for s in specs:
        if not s.blocking:
            continue
        key = s.metric
        if key.endswith("_ms"):                    # 延迟类指标越小越好
            if new_metrics[key] > s.threshold:
                return False
        elif new_metrics[key] < s.threshold:       # 准确率类越大越好
            return False
        if new_metrics[key] < old_metrics[key] - 0.02:  # 不得显著劣化
            return False
    return True