┌─────────── 第 1 关：单元/集成测试（pytest） ───────────────┐
│  "每个零件都对吗？拼起来对吗？"                             │
│  确定性 mock → 可重复 → CI 里自动跑                         │
└───────────────────────┬─────────────────────────────────────┘
                        │ 全绿 ┌───────────────────────┴─── 第 2 关：评测（Eval） ──────────┐
│  "端到端的质量达标吗？"                                      │
│  Golden dataset → 打分 → pass_rate ≥ 阈值                   │
│  不是"对不对"而是"好不好"                                    │
└───────────────────────┬─────────────────────────────────────┘
                        │ 通过阈值 ┌───────────────────────┴─── 第 3 关：部署（Deploy） ────────┐
│  "生产环境能安全跑起来吗？"                                  │
│  Docker 打包 → CI/CD 流水线 → 健康检查 → 灰度上线           │
└──────────────────────────────────────────────────────────────┘

[pytest]
testpaths = tests
asyncio_mode = strict
addopts = -v --strict-markers --tb=short --disable-warnings
markers =
    slow: 耗时超过 1s 的测试（默认跳过用 -m "not slow"）
    integration: 依赖外部服务（OpenAI / DB 等）的集成测试
    unit: 纯逻辑单元测试（默认跑这一组）

tests/
├── conftest.py              # 公共 fixture
├── test_chat.py / test_qa.py / test_agent.py / test_knowledge.py
├── test_rag.py / test_hybrid.py / test_rag_tuning.py
├── test_rag_graph.py / test_verified_qa.py / test_verifier.py
├── test_auth_rbac.py / test_observability_ch16.py
└── eval/                    # 评测框架
    ├── golden/              # 黄金数据集
    ├── metrics.py           # 打分函数
    └── run_eval.py          # 评测入口

@dataclass
class FakeMessage:
    content: str = "mock-answer"

@dataclass
class FakeChoice:
    message: FakeMessage = field(default_factory=FakeMessage)

@dataclass
class FakeChatCompletion:
    choices: list[FakeChoice] = field(default_factory=lambda: [FakeChoice()])
    usage: FakeUsage = field(default_factory=FakeUsage)

@pytest.fixture
def mock_openai(monkeypatch: pytest.MonkeyPatch) -> None:
    async def _fake_create(*_args, **_kwargs):
        return FakeChatCompletion()

    monkeypatch.setattr(
        llm_service._client.chat.completions, "create", _fake_create
    )

@pytest.fixture
def mock_embedding(monkeypatch: pytest.MonkeyPatch) -> None:
    import hashlib
    def _fake_vector(text: str) -> list[float]:
        seed = int(hashlib.sha256(text.encode()).hexdigest()[:8], 16)
        base = [(seed >> (i * 2)) & 0xFF for i in range(16)]
        norm = [b / max(sum(b*b for b in base)**0.5, 1e-6) for b in base]
        return (norm * (1536 // 16))[:1536]
    monkeypatch.setattr(embedding_service, "embed", _fake_embed)

@pytest.fixture
def temp_vector_store(monkeypatch, tmp_path):
    store = VectorStore(persist_dir=str(tmp_path / "chroma"))
    monkeypatch.setattr(_vs, "vector_store", store)
    # FTS + conversations + approvals + users + ACL 全部用 tmp_path/db
    fts_db_path = str(tmp_path / "fts.db")
    _fts.init_fts(db_path=fts_db_path)
    _conv_db.init_conversations(db_path=fts_db_path)
    _users_db.init_users(db_path=fts_db_path)
    _acl_db.init_acl(db_path=fts_db_path)
    yield store

@pytest.mark.unit
async def test_rewrite_node_skips_when_disabled() -> None:
    out = await rewrite_node(
        {"question": "什么是 RAG？", "enable_rewrite": False}
    )
    assert out["rewritten_query"] == "什么是 RAG？"
    assert out["meta"]["rewrite_skipped"] is True

async def test_graph_hit_path(monkeypatch) -> None:
    monkeypatch.setattr(nodes_mod.hybrid_retrieval_service, "retrieve", ...)
    monkeypatch.setattr(nodes_mod.llm_service, "chat_messages", _fake_chat)
    out = await rag_graph_service.ask("什么是 RAG？")
    assert out["finish_reason"] == "answered"

def test_api_rag_graph_ask_happy(client, monkeypatch) -> None:
    resp = client.post("/api/v1/rag-graph/ask", json={"question": "什么是 RAG？"})
    assert resp.status_code == 200

        △  API 测试（少量）→ 验证 HTTP + 中间件
       ╱ ╲
      ╱   ╲
     ╱ 整图 ╲ → 验证节点串联 + 路由
    ╱  e2e   ╲
   ╱───────────╲
  ╱  单节点测试  ╲ → 验证函数输入/输出（最多最快）
 ╱─────────────────╲

{"id":"rag-1","question":"什么是 RAG？","expected_keywords":["检索","生成"],"expected_citations_min":1,"min_trust_level":"medium"}
{"id":"rag-4","question":"哪里能买到月亮？","expected_keywords":["无法回答"],"expected_citations_min":0,"min_trust_level":"low"}

def rag_score(sample: dict, resp: dict) -> dict:
    ans = (resp.get("answer") or "").lower()
    keywords = [k.lower() for k in sample.get("expected_keywords", [])]
    kw_hit = all(k in ans for k in keywords)
    cit_ok = len(resp.get("citations") or []) >= sample.get("expected_citations_min", 0)
    trust_ok = _trust_ge(resp.get("summary", {}).get("trust_level", "low"),
                          sample.get("min_trust_level", "low"))
    return {"id": sample["id"], "passed": kw_hit and cit_ok and trust_ok,
            "kw_hit": kw_hit, "cit_ok": cit_ok, "trust_ok": trust_ok}

def acl_score(sample: dict, resp: dict) -> dict:
    blob = resp.get("answer", "") + " ".join(e.get("text","") for e in resp.get("evidences", []))
    leaked = [w for w in sample.get("must_not_contain", []) if w in blob]
    return {"id": sample["id"], "passed": len(leaked) == 0, "leaked": leaked}

FROM python:3.11-slim AS deps
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

FROM deps AS runtime
COPY . .
ENV PYTHONPATH=/app
HEALTHCHECK --interval=30s --timeout=5s \
  CMD python -c "import httpx; httpx.get('http://localhost:8000/health').raise_for_status()"
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

代码推送 → pytest -m unit → 全绿 → 构建 Docker 镜像
→ 跑评测 --subset verifier → 推送镜像 → 灰度部署

开发: .env 文件（gitignore）    测试: CI secrets    生产: K8s ConfigMap + Secret

OPENAI_API_KEY=sk-...    # 必须从 Secret 注入，禁止硬编码
JWT_SECRET=...            # 必须覆盖默认值
ENABLE_AUTH=true          # 生产必须开启

# 每次 CI 跑 = 花钱
resp = await openai.chat.completions.create(...)
# mock = $0
monkeypatch.setattr(llm_service._client.chat.completions, "create", _fake_create)

async def test_generate_node_parses_citations(mock_openai, monkeypatch):
    async def _fake_create(*a, **kw):
        return FakeChatCompletion(choices=[FakeChoice(message=FakeMessage(
            content="RAG 是检索增强生成 [1]。结合检索和生成 [1][2]。"))])
    monkeypatch.setattr(llm_service._client.chat.completions, "create", _fake_create)
    out = await generate_node({"question": "...", "hits": [...]})
    assert "[1]" in out["answer"]

{"id":"v-conflict-3","question":"向量库需要GPU吗","answer":"看场景",
 "claims":[{"text":"需要GPU","verification":"supported","evidence_ids":["e1"]},
           {"text":"不需要GPU","verification":"supported","evidence_ids":["e2"]}],
 "expected_verdict":"conflict"}

SYNONYMS = {"检索":["检索","获取","搜索"],"生成":["生成","创建","产生"],"增强":["增强","强化","提升"]}

{"id":"rag-steps","expected_keywords_sets":[["检索","生成","增强"],["获取","创建","强化"],["retrieval","generation","augmentation"]]}

def kw_hit_any_set(ans, keyword_sets):
    return any(all(k.lower() in ans.lower() for k in ks) for ks in keyword_sets)