feat(rag): 服务层接线（上传即索引 + use_rag 异步链路 + 强 e2e）

2026-08-30 00:25:24 +08:00
parent 9514749b9f
commit f77fab707d
6 changed files with 228 additions and 9 deletions
@@ -0,0 +1,146 @@
+"""Task 5（D3 强端到端）：服务层 RAG 接线验证。
+
+真实驱动完整链路：上传 → 解压 → index_dir → retrieve → LLM prompt（RAG 注入），
+不 mock 任何环节。验证上传即索引（D1）与 use_rag 默认关闭的向后兼容。
+"""
+from __future__ import annotations
+
+import asyncio
+import glob
+import io
+import zipfile
+from pathlib import Path
+
+import pytest
+
+from genesis.impact.impact_agent import _RAG_CONTEXT_TITLE
+from genesis.rag.embeddings import FakeEmbedder
+from genesis.rag.impact_rag import ImpactRAG
+from genesis.rag.store import RagStore
+from genesis.server.service import GenesisService
+from genesis.server.store import SessionStore
+
+_REPO = Path(__file__).resolve().parents[1]
+
+
+def _sample_req() -> bytes:
+    hits = glob.glob(str(_REPO / "sample" / "requirements_*.xlsx"))
+    assert hits, "sample 下未找到 requirements xlsx"
+    return Path(hits[0]).read_bytes()
+
+
+def _sample_tpl() -> bytes:
+    hits = glob.glob(str(_REPO / "sample" / "template_*.docx"))
+    assert hits, "sample 下未找到 template docx"
+    return Path(hits[0]).read_bytes()
+
+
+def _make_existing_zip() -> bytes:
+    """构造含已知源码的既有系统 zip（src/KnownOrder.java 含『订单创建调用 MyBatis』）。"""
+    buf = io.BytesIO()
+    with zipfile.ZipFile(buf, "w") as zf:
+        zf.writestr(
+            "src/KnownOrder.java",
+            "package demo;\n"
+            "public class KnownOrder {\n"
+            "  // 订单创建调用 MyBatis\n"
+            "  public void createOrder() { /* ... */ }\n}\n",
+        )
+    return buf.getvalue()
+
+
+class FakeEngine:
+    """异步确定性引擎：捕获 LLM 收到的 prompt。"""
+
+    def __init__(self) -> None:
+        self.captured: str | None = None
+
+    async def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2):
+        from types import SimpleNamespace
+        self.captured = prompt
+        return SimpleNamespace(
+            data={"title": variables.get("title", "x"), "blocks": []},
+            raw_text=prompt,
+            status="ok",
+        )
+
+
+@pytest.fixture
+def svc(tmp_path):
+    store = SessionStore(db_path=str(tmp_path / "s.db"))
+    rag = ImpactRAG(RagStore(str(tmp_path / "rag.db")), FakeEmbedder())
+    return GenesisService(
+        store=store,
+        data_root=str(tmp_path / "data"),
+        engine=FakeEngine(),
+        rag=rag,
+        use_rag=False,  # 默认关闭，由 run_impact 显参开启
+    )
+
+
+def test_rag_e2e_upload_then_impact_injects_source(tmp_path):
+    """D3 强验证：上传源码被 RAG 检索并注入 LLM prompt。"""
+    store = SessionStore(db_path=str(tmp_path / "s.db"))
+    engine = FakeEngine()
+    rag = ImpactRAG(RagStore(str(tmp_path / "rag.db")), FakeEmbedder())
+    svc = GenesisService(
+        store=store,
+        data_root=str(tmp_path / "data"),
+        engine=engine,
+        rag=rag,
+        use_rag=False,
+    )
+
+    sid = svc.create_session("u1").session_id
+    svc.upload_file(sid, "requirements", "requirements_newdev.xlsx", _sample_req())
+    svc.upload_file(sid, "template", "template_design_ja.docx", _sample_tpl())
+    # 上传即索引（D1）
+    svc.upload_file(sid, "existing_system", "existing.zip", _make_existing_zip())
+
+    # 上传后直接检索应命中非空
+    hits = rag.retrieve(sid, "订单创建", k=1)
+    assert hits, "上传即索引后应能检索到源码片段"
+    assert "KnownOrder" in hits[0]
+
+    svc.run_parse(sid)
+    svc.confirm_parse(sid)
+    assert svc.get_session(sid).status == "impact_running"
+
+    # 真实驱动 RAG 路径
+    rec = asyncio.run(svc.run_impact(sid, use_rag=True))
+    assert rec.status == "awaiting_impact_confirm"
+    # 核心断言：RAG 检索片段（含 KnownOrder）注入 LLM prompt
+    assert engine.captured is not None
+    assert "KnownOrder" in engine.captured
+    # 摘要标记 rag_enabled
+    import json
+    assert json.loads(rec.impact_summary).get("rag_enabled") is True
+
+
+def test_rag_e2e_use_rag_false_backward_compat(tmp_path):
+    """向后兼容：rag=None 且 use_rag 默认关闭时，不调用 LLM、无 RAG 小节。"""
+    store = SessionStore(db_path=str(tmp_path / "s.db"))
+    engine = FakeEngine()
+    svc = GenesisService(
+        store=store,
+        data_root=str(tmp_path / "data"),
+        engine=engine,
+        rag=None,  # 不启用 RAG
+        use_rag=False,
+    )
+
+    sid = svc.create_session("u1").session_id
+    svc.upload_file(sid, "requirements", "requirements_newdev.xlsx", _sample_req())
+    svc.upload_file(sid, "template", "template_design_ja.docx", _sample_tpl())
+    svc.upload_file(sid, "existing_system", "existing.zip", _make_existing_zip())
+
+    svc.run_parse(sid)
+    svc.confirm_parse(sid)
+
+    rec = asyncio.run(svc.run_impact(sid))  # 默认 use_rag=None → 走确定性路径
+    assert rec.status == "awaiting_impact_confirm"
+    # 向后兼容：确定性路径不调用 LLM（captured 保持 None），prompt 不含 RAG 小节
+    assert engine.captured is None
+    assert _RAG_CONTEXT_TITLE not in (engine.captured or "")
+    # 确定性报告存在
+    assert rec.impact_report_path and Path(rec.impact_report_path).exists()
@@ -7,6 +7,7 @@ from __future__ import annotations
 from pathlib import Path
 from types import SimpleNamespace

+import asyncio
 import pytest

 from genesis.server.store import SessionStore, ProjectsStore
@@ -248,7 +249,7 @@ def test_existing_system_zip_triggers_impact_flow(tmp_path):
    svc.confirm_parse(s.session_id)
    assert svc.get_session(s.session_id).status == "impact_running"

-    svc.run_impact(s.session_id)
+    asyncio.run(svc.run_impact(s.session_id))
    got = svc.get_session(s.session_id)
    assert got.status == "awaiting_impact_confirm"
    assert got.impact_summary
@@ -269,7 +270,7 @@ def test_impact_state_guard(svc):
    svc.confirm_parse(s.session_id)
    # 无既有系统 → writing，start-impact 应报错
    with pytest.raises(ServiceStepError):
-        svc.run_impact(s.session_id)
+        asyncio.run(svc.run_impact(s.session_id))


 def test_qa_requires_generated(svc):