diff --git a/docs/design.md b/docs/design.md index e32e6d0..76ba672 100644 --- a/docs/design.md +++ b/docs/design.md @@ -1921,3 +1921,15 @@ Document(注入后 Word 文档) ### 12.8 WebSocket 实时进度流(2026-08-29,分支 feat/websocket-progress) 新增 `ProgressHub` 进程内发布/订阅单例 + `/api/sessions/{sid}/ws` 端点 + `chat_ws.js` 前端实时渲染;进度/错误事件实时推送,既有 `role='progress'/'error'` 持久化兜底保留(重载仍可见)。**单进程假设**:hub 为进程内单例,多 worker 部署下跨进程不互通(后续可迭代 Redis 总线)。 + + +## RAG 影响调查接入说明(Task 5,2026-08-29) + +影响调查 Agent 接入可选 RAG 检索能力(既有系统源码 -> 检索上下文注入 LLM 影响分析 prompt),详见 `src/genesis/rag/` 与 `src/genesis/impact/impact_agent.py` 的 `run_impact`。 + +- **scope = session_id**:每个会话的既有系统源码独立索引到 `RagStore` 的同一 scope,互不串扰。 +- **上传即索引(D1)**:`GenesisService.upload_file` 在 `file_type == "existing_system"` 且 `self.rag is not None` 时,解压完成后立即调用 `self.rag.index_dir(session_id, path)`;索引异常仅记录日志(`_LOGGER.warning`)不阻断上传。 +- **`use_rag` 默认关闭**:`GenesisService` 构造参数 `use_rag` 默认 `False`,`rag=None` 表示不启用(向后兼容)。`run_impact(session_id, use_rag=None)` 中 `eff = self.use_rag if use_rag is None else use_rag`;仅当 `eff 且 self.rag is not None` 时走 LLM+RAG 路径,否则走原确定性 `ImpactAgent().run(...)` 路径(行为不变)。 +- **异步链路**:`run_impact` 为 `async def`,RAG 路径 `await ImpactAgent(engine=..., rag=..., use_rag=True).run_impact(...)`;`app.start_impact` 端点改为 `async def` 并 `await service.run_impact(sid, use_rag=use_rag)`;`chat/agent.py` 调用处以 `asyncio.run(...)` 包裹以兼容同步消息处理。 +- **线程安全(D2)**:`RagStore` 构造使用 `sqlite3.connect(db_path, check_same_thread=False)` 并加 `threading.Lock`,读写均加锁串行化,适配 Web 服务端 worker 线程复用连接。 +- **向后兼容**:`use_rag=False` 时 prompt 不含 RAG 小节标题(`_RAG_CONTEXT_TITLE`),影响报告为确定性 `impact-report.json`,不调用 LLM。 diff --git a/src/genesis/chat/agent.py b/src/genesis/chat/agent.py index adf5e9f..9a675e3 100644 --- a/src/genesis/chat/agent.py +++ b/src/genesis/chat/agent.py @@ -5,6 +5,7 @@ """ from __future__ import annotations +import asyncio import json from typing import Callable @@ -170,7 +171,7 @@ class ChatAgent: self._emit_progress(session_id, progress[-1]) rec = self.service.confirm_parse(session_id) if rec.status == "impact_running": - rec = self.service.run_impact(session_id) + rec = asyncio.run(self.service.run_impact(session_id)) progress.append({"step": "impact", "status": "ok", "detail": f"影响调查完成:{self._impact_brief(rec)}"}) self._emit_progress(session_id, progress[-1]) @@ -224,7 +225,7 @@ class ChatAgent: reply = "当前状态无法进行影响调查。" self.store.add_message(session_id, "assistant", reply, action="impact") return {"reply": reply, "progress": progress, "status": rec.status} - rec = self.service.run_impact(session_id) + rec = asyncio.run(self.service.run_impact(session_id)) progress.append({"step": "impact", "status": "ok", "detail": f"影响调查完成:{self._impact_brief(rec)}"}) self._emit_progress(session_id, progress[-1]) diff --git a/src/genesis/impact/impact_agent.py b/src/genesis/impact/impact_agent.py index bf5e509..2012146 100644 --- a/src/genesis/impact/impact_agent.py +++ b/src/genesis/impact/impact_agent.py @@ -84,8 +84,75 @@ def _header_index(headers: list[str], *keywords: str) -> int | None: return None +# RAG 检索命中片段注入到 prompt 的明确小节标题(向后兼容:use_rag=False 时不出现) +_RAG_CONTEXT_TITLE = "# 既有系统关联上下文(RAG 检索,辅助判断影响范围)" + + class ImpactAgent: - """变更点定位 → 影响调查书(MVP)。""" + """变更点定位 → 影响调查书(MVP)。 + + MVP 的确定性规则路径由 ``run`` 提供(无 LLM 参与)。 + 另提供 LLM 驱动的 ``run_impact``,可接入可选 RAG 上下文辅助判断影响范围。 + """ + + def __init__( + self, + engine=None, + use_rag: bool = False, + rag: "ImpactRAG | None" = None, + ) -> None: + """初始化(向后兼容:无参 ``ImpactAgent()`` 仍可用)。 + + - engine: LLM 引擎(InferenceEngine 兼容接口,提供 chat_structured)。 + - use_rag: 实例级默认是否启用 RAG 上下文注入;``run_impact`` 可用显参覆盖。 + - rag: 可选 ImpactRAG 检索器(scope=session_id)。 + """ + self.engine = engine + self.use_rag = use_rag + self.rag = rag + + def _build_impact_prompt(self, requirements_text: str) -> str: + """拼装发送给 LLM 的基础 prompt(不含 RAG 上下文)。""" + return ( + "你是一名变更影响分析专家。请基于以下要件变更说明,判断本次变更的影响范围" + "(涉及的既有機能/画面/DB/IF/バッチ,以及需要修改或回归验证的对象)," + "并说明判断依据。\n\n" + "# 要件变更说明\n" + f"{requirements_text}\n" + ) + + async def run_impact( + self, + session_id: str, + requirements_text: str, + use_rag: bool | None = None, + k: int = 5, + ): + """LLM 驱动的变更影响分析(可选 RAG 上下文注入,异步)。 + + - use_rag 优先取显参;为 None 时回退实例级 self.use_rag。 + - 启用且 self.rag 存在时,以「影响调查:」+ 要件前若干字 为查询, + 调用 self.rag.retrieve(session_id, query, k),将命中片段注入 prompt。 + - use_rag=False 时 prompt 不含 RAG 小节(向后兼容)。 + - 返回底层引擎的 StructuredResult(含 data/raw_text)。 + """ + if self.engine is None: + raise RuntimeError("run_impact 需要 engine,请在构造 ImpactAgent 时传入") + if use_rag is None: + use_rag = self.use_rag + prompt = self._build_impact_prompt(requirements_text) + if use_rag and self.rag is not None: + query = "影响调查:" + requirements_text[:200] + chunks = self.rag.retrieve(session_id, query, k) + if chunks: + rag_context = "\n".join(chunks) + prompt = prompt + f"\n\n{_RAG_CONTEXT_TITLE}\n{rag_context}" + return await self.engine.chat_structured( + session_id=session_id, + prompt=prompt, + variables={}, + schema={}, + ) def run( self, diff --git a/src/genesis/rag/__init__.py b/src/genesis/rag/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/genesis/rag/embeddings.py b/src/genesis/rag/embeddings.py new file mode 100644 index 0000000..6c59ff1 --- /dev/null +++ b/src/genesis/rag/embeddings.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +import hashlib +import math +import re +from typing import List, Protocol + + +class Embedder(Protocol): + def embed(self, texts: List[str]) -> List[List[float]]: ... + + +_DIM = 64 + + +def _tokenize(text: str) -> List[str]: + # 分词意图:先将文本统一转为小写,再按非字母数字字符切分为词元,最后过滤空串 + toks = re.split(r"\W+", text.lower()) + return [t for t in toks if t] + + +class FakeEmbedder: + def embed(self, texts: List[str]) -> List[List[float]]: + vecs = [] + for t in texts: + v = [0.0] * _DIM + for tok in _tokenize(t): + h = hashlib.md5(tok.encode("utf-8")).digest() + idx = h[0] % _DIM + v[idx] += 1.0 + # L2 归一化;norm 为 0 时(全零向量)用 1.0 防除零 + norm = math.sqrt(sum(x * x for x in v)) or 1.0 + vecs.append([x / norm for x in v]) + return vecs + + +def get_embedder(engine: str) -> Embedder: + # engine 当前未使用,统一回退到 FakeEmbedder;真实向量模型(如 OpenAI/BGE)接入点预留于此 + return FakeEmbedder() diff --git a/src/genesis/rag/impact_rag.py b/src/genesis/rag/impact_rag.py new file mode 100644 index 0000000..ab14a2d --- /dev/null +++ b/src/genesis/rag/impact_rag.py @@ -0,0 +1,72 @@ +from __future__ import annotations + +from pathlib import Path +from typing import List, Tuple + +from genesis.rag.embeddings import Embedder +from genesis.rag.store import RagStore + +_CHUNK = 800 +_TEXT_EXTS = { + ".java", ".py", ".js", ".ts", ".go", ".kt", ".scala", ".xml", ".yml", + ".yaml", ".md", ".txt", ".json", ".csv", ".sql", ".html", ".css", ".sh", +} + + +def _split(text: str, size: int = _CHUNK) -> List[str]: + paras = [p.strip() for p in text.split("\n") if p.strip()] + out, buf = [], "" + for p in paras: + if len(buf) + len(p) > size and buf: + out.append(buf) + buf = p + else: + buf = (buf + "\n" + p).strip() + if buf: + out.append(buf) + return out or [""] + + +class ImpactRAG: + def __init__(self, store: RagStore, embedder: Embedder): + self.store = store + self.embedder = embedder + + def index(self, scope: str, sources: List[Tuple[str, str]]) -> None: + chunks = [] + for name, text in sources: + for piece in _split(text): + # 跳过空片段,避免产生空向量噪声 chunk + if piece.strip(): + chunks.append(f"[{name}]\n{piece}") + embs = self.embedder.embed(chunks) + self.store.add(scope, chunks, embs) + + def index_dir(self, scope: str, root: str) -> int: + """D1:索引目录内文本源文件(既有系统源码)。返回索引的文件数。""" + root_path = Path(root) + sources: List[Tuple[str, str]] = [] + if root_path.is_dir(): + for p in root_path.rglob("*"): + if p.is_file() and p.suffix.lower() in _TEXT_EXTS: + try: + text = p.read_text(encoding="utf-8", errors="ignore") + except Exception: + # 读取异常的文件跳过,不中断整体索引 + continue + # 空内容文件跳过 + if not text.strip(): + continue + # 含 NUL 字节的疑似二进制文件(误带白名单扩展名)跳过 + if "\x00" in text: + continue + sources.append((str(p.relative_to(root_path)), text)) + if sources: + self.index(scope, sources) + return len(sources) + + def retrieve(self, scope: str, query: str, k: int = 5) -> List[str]: + if not query.strip(): + return [] + qv = self.embedder.embed([query])[0] + return self.store.search(scope, qv, k) diff --git a/src/genesis/rag/store.py b/src/genesis/rag/store.py new file mode 100644 index 0000000..4c4ac09 --- /dev/null +++ b/src/genesis/rag/store.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +import json +import math +import sqlite3 +import threading +from typing import List + + +def _cosine(a, b): + dot = sum(x * y for x, y in zip(a, b)) + na = math.sqrt(sum(x * x for x in a)) + nb = math.sqrt(sum(y * y for y in b)) + return dot / (na * nb or 1.0) + + +class RagStore: + def __init__(self, db_path: str): + # D2:服务端在 worker 线程复用连接,须关闭同线程检查 + self.conn = sqlite3.connect(db_path, check_same_thread=False) + self._lock = threading.Lock() + self.conn.execute( + "CREATE TABLE IF NOT EXISTS rag_chunks (" + "id INTEGER PRIMARY KEY, scope TEXT, chunk TEXT, embedding TEXT)" + ) + self.conn.commit() + + def reset_scope(self, scope: str) -> None: + with self._lock: + self.conn.execute("DELETE FROM rag_chunks WHERE scope=?", (scope,)) + self.conn.commit() + + def add(self, scope: str, chunks: List[str], embeddings: List[List[float]]) -> None: + # 按 scope 全量替换(先清后写),非追加;整段加锁串行化 + if len(chunks) != len(embeddings): + raise ValueError("chunks 与 embeddings 长度不一致") + with self._lock: + self.conn.execute("DELETE FROM rag_chunks WHERE scope=?", (scope,)) + self.conn.executemany( + "INSERT INTO rag_chunks(scope, chunk, embedding) VALUES(?,?,?)", + [(scope, c, json.dumps(e)) for c, e in zip(chunks, embeddings)], + ) + self.conn.commit() + + def search(self, scope: str, query_vec: List[float], k: int = 5) -> List[str]: + k = max(0, int(k)) + if k <= 0: + return [] + # 读也加锁,保证跨线程读写串行化 + with self._lock: + rows = self.conn.execute( + "SELECT chunk, embedding FROM rag_chunks WHERE scope=?", (scope,) + ).fetchall() + scored = [] + for chunk, emb in rows: + scored.append((_cosine(query_vec, json.loads(emb)), chunk)) + scored.sort(key=lambda x: x[0], reverse=True) + return [c for _, c in scored[:k]] + + def close(self) -> None: + self.conn.close() diff --git a/src/genesis/server/app.py b/src/genesis/server/app.py index 3778d25..7d85a6f 100644 --- a/src/genesis/server/app.py +++ b/src/genesis/server/app.py @@ -32,7 +32,7 @@ VERSION = "0.1.0" class _FakeEngine: """离线确定性引擎(--engine fake):用于无 API key 的 Web 演示/测试。""" - def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2): + async def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2): from types import SimpleNamespace title = variables.get("title", "x") return SimpleNamespace( @@ -83,6 +83,7 @@ def create_app( store: SessionStore | None = None, data_root: str = "data/server", engine: Any = None, + rag: "ImpactRAG | None" = None, ) -> FastAPI: store = store or SessionStore() is_fake = engine == "fake" @@ -91,8 +92,20 @@ def create_app( elif engine is None: engine = None # 真实模式:generate/qa 时按需 build(避免未配置 key 直接 503) + # RAG 接线:未注入则内部构造并默认关闭(use_rag=False,向后兼容) + if rag is None: + from genesis.rag.embeddings import get_embedder + from genesis.rag.impact_rag import ImpactRAG + from genesis.rag.store import RagStore + rag_db = str(Path(data_root) / "rag.db") + Path(rag_db).parent.mkdir(parents=True, exist_ok=True) + rag = ImpactRAG(RagStore(rag_db), get_embedder("fake")) + projects = ProjectsStore(db_path=str(store._db)) if store else None - service = GenesisService(store=store, data_root=data_root, engine=engine, projects=projects) + service = GenesisService( + store=store, data_root=data_root, engine=engine, projects=projects, + rag=rag, use_rag=False, + ) from genesis.chat.agent import ChatAgent chat_agent = ChatAgent(service=service, fake=is_fake, engine=engine) @@ -232,9 +245,9 @@ def create_app( # ---------- 影响调查 ---------- @app.post("/api/sessions/{sid}/start-impact") - def start_impact(sid: str): + async def start_impact(sid: str, use_rag: bool = False): try: - rec = service.run_impact(sid) + rec = await service.run_impact(sid, use_rag=use_rag) except ServiceStepError as e: raise _error(409, "STATE_TRANSITION_INVALID", str(e)) return {"ok": True, "status": rec.status} diff --git a/src/genesis/server/service.py b/src/genesis/server/service.py index 437ddd0..6bbce6c 100644 --- a/src/genesis/server/service.py +++ b/src/genesis/server/service.py @@ -13,7 +13,9 @@ from __future__ import annotations import html import json +import logging import shutil +import threading import zipfile from pathlib import Path @@ -22,6 +24,8 @@ from docx import Document from genesis.parsers.source_aggregator import SourceParser from genesis.server.store import SessionStore, SessionRecord, ProjectsStore, ProjectConfigError +_LOGGER = logging.getLogger(__name__) + ALLOWED_FILE_TYPES = { "requirements", "template", "write_instruction", "rules", "existing_system", } @@ -57,6 +61,8 @@ class GenesisService: prompt_registry=None, samples_dir: str = "sample", projects: "ProjectsStore | None" = None, + rag: "ImpactRAG | None" = None, + use_rag: bool = False, ) -> None: self.store = store self.data_root = Path(data_root) @@ -64,6 +70,11 @@ class GenesisService: self.prompt_registry = prompt_registry self.samples_dir = samples_dir self.projects = projects + # RAG:rag 为 None 表示不启用(向后兼容默认关闭) + self.rag = rag + self.use_rag = use_rag + # 引擎构建锁:避免 worker 线程并发下共享可变属性的竞态(D2) + self._engine_lock = threading.Lock() # ---------- 会话与文件 ---------- @@ -126,6 +137,12 @@ class GenesisService: zf.extractall(dst) zpath.unlink(missing_ok=True) path = str(dst) + # D1:上传即索引(scope=session_id),索引异常不阻断上传 + if self.rag is not None: + try: + self.rag.index_dir(session_id, path) + except Exception as e: # noqa: BLE001 + _LOGGER.warning("existing_system 索引失败(已忽略): %s", e) else: path = str(sdir / filename) Path(path).write_bytes(content) @@ -204,10 +221,37 @@ class GenesisService: # ---------- 影响调查 ---------- - def run_impact(self, session_id: str) -> SessionRecord: + async def run_impact(self, session_id: str, use_rag: bool | None = None) -> SessionRecord: rec = self.get_session(session_id) if rec.status != "impact_running": raise ServiceStepError(f"当前状态 {rec.status} 不可启动影响调查") + # 解析是否启用 RAG(实例默认 use_rag 可被显参覆盖) + eff = self.use_rag if use_rag is None else use_rag + if eff and self.rag is not None: + # RAG 路径:LLM 驱动,检索既有系统上下文注入 prompt + if self.engine is None: + with self._engine_lock: + if self.engine is None: + from genesis.inference.factory import build_inference_engine + self.engine = build_inference_engine() + from genesis.impact.impact_agent import ImpactAgent + requirements_text = rec.files.get("requirements", {}).get("name") or "要件定義" + result = await ImpactAgent(engine=self.engine, rag=self.rag, use_rag=True).run_impact( + session_id, requirements_text, use_rag=True) + summary = { + "rag_enabled": True, + "impact_llm": getattr(result, "raw_text", None) or getattr(result, "data", None), + } + # 尽量保留既有确定性报告文件(无则保持原值) + impact_report_path = rec.impact_report_path + self.store.update_session( + session_id, + status="awaiting_impact_confirm", + impact_summary=json.dumps(summary, ensure_ascii=False), + impact_report_path=impact_report_path, + ) + return self.get_session(session_id) + # 默认确定性路径(向后兼容) from genesis.impact.impact_agent import ImpactAgent, impact_report_to_dict ss = self._rebuild_source(rec) report = ImpactAgent().run(ss, session_id=session_id) diff --git a/tests/test_impact_agent_rag.py b/tests/test_impact_agent_rag.py new file mode 100644 index 0000000..89748e5 --- /dev/null +++ b/tests/test_impact_agent_rag.py @@ -0,0 +1,91 @@ +"""ImpactAgent RAG 上下文注入测试(RAG 迭代 Task 4,异步形态)。 + +验证: +- use_rag=True 时,run_impact 发送给 LLM 的 prompt 文本包含 RAG 检索命中片段与明确小节标题。 +- use_rag=False(或默认)时,prompt 文本不含 RAG 小节标题(向后兼容)。 +- run_impact 为 async def,真实 await 引擎 chat_structured。 +""" +import asyncio +import types + +import pytest + +from genesis.impact.impact_agent import ImpactAgent, _RAG_CONTEXT_TITLE +from genesis.rag.embeddings import FakeEmbedder +from genesis.rag.impact_rag import ImpactRAG +from genesis.rag.store import RagStore + + +class FakeEngine: + """捕获真实 LLM 方法(chat_structured,async)收到的 prompt 文本。 + + 方法名与签名刻意复用本仓库 InferenceEngine.chat_structured 的形参风格, + 以保证 mock 的是真实接口(key=session_id/prompt/variables/schema/retry_count)。 + """ + + def __init__(self) -> None: + self.captured: str | None = None + self.calls = 0 + + async def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2): + self.captured = prompt + self.calls += 1 + # 返回结构兼容 StructuredResult 的最小占位(含 data/raw_text) + return types.SimpleNamespace(data={}, raw_text=prompt) + + +def _make_rag(session_id: str, text: str) -> ImpactRAG: + store = RagStore(":memory:") + rag = ImpactRAG(store, FakeEmbedder()) + rag.index(session_id, [("TradeApplication.java", text)]) + return rag + + +def test_run_impact_with_rag_injects_context(): + rag = _make_rag("p1", "订单创建调用 MyBatis") + engine = FakeEngine() + agent = ImpactAgent(engine=engine, rag=rag, use_rag=True) + asyncio.run(agent.run_impact("p1", "创建订单的影响")) + assert engine.captured is not None + # 命中片段(含文件名 TradeApplication.java)被注入 + assert "TradeApplication" in engine.captured + # 明确小节标题被注入 + assert _RAG_CONTEXT_TITLE in engine.captured + + +def test_run_impact_without_rag_no_context(): + engine = FakeEngine() + agent = ImpactAgent(engine=engine, rag=_make_rag("p1", "x"), use_rag=False) + asyncio.run(agent.run_impact("p1", "创建订单的影响")) + assert _RAG_CONTEXT_TITLE not in engine.captured + + +def test_run_impact_rag_enabled_but_no_hits(): + # rag 已建但索引在另一 scope,retrieve 命中为空 → 不应注入 RAG 小节 + engine = FakeEngine() + agent = ImpactAgent(engine=engine, rag=_make_rag("other", "订单创建调用 MyBatis"), use_rag=True) + asyncio.run(agent.run_impact("p1", "创建订单的影响")) + assert _RAG_CONTEXT_TITLE not in engine.captured + + +def test_run_impact_use_rag_none_falls_back_to_instance_default(): + # use_rag=None 时回退实例级 self.use_rag(此处为 True)→ 应注入 RAG 小节 + engine = FakeEngine() + agent = ImpactAgent(engine=engine, rag=_make_rag("p1", "订单创建调用 MyBatis"), use_rag=True) + asyncio.run(agent.run_impact("p1", "创建订单的影响", use_rag=None)) + assert _RAG_CONTEXT_TITLE in engine.captured + assert "TradeApplication" in engine.captured + + +def test_run_impact_explicit_false_no_context(): + # 显式 use_rag=False(覆盖 use_rag is None 的 False 分支)→ 不注入 RAG 小节 + engine = FakeEngine() + agent = ImpactAgent(engine=engine, rag=_make_rag("p1", "x"), use_rag=True) + asyncio.run(agent.run_impact("p1", "创建订单的影响", use_rag=False)) + assert _RAG_CONTEXT_TITLE not in engine.captured + + +def test_run_impact_requires_engine(): + agent = ImpactAgent() + with pytest.raises(RuntimeError): + asyncio.run(agent.run_impact("p1", "x")) diff --git a/tests/test_impact_rag.py b/tests/test_impact_rag.py new file mode 100644 index 0000000..dfab69c --- /dev/null +++ b/tests/test_impact_rag.py @@ -0,0 +1,156 @@ +from genesis.rag.embeddings import FakeEmbedder +from genesis.rag.store import RagStore +from genesis.rag.impact_rag import ImpactRAG + + +def test_retrieve_returns_relevant_chunk(): + store = RagStore(":memory:") + try: + rag = ImpactRAG(store, FakeEmbedder()) + sources = [ + ("OrderController.java", "public class OrderController { 创建订单 }"), + ("UserAuth.java", "public class UserAuth { 用户登录认证 }"), + ] + rag.index("p1", sources) + res = rag.retrieve("p1", "OrderController 的影响范围", k=1) + assert res and "OrderController" in res[0] + finally: + store.close() + + +def test_index_dir_reads_text_files(tmp_path): + (tmp_path / "src").mkdir() + (tmp_path / "src" / "OrderController.java").write_text( + "public class OrderController { 创建订单 }", encoding="utf-8" + ) + (tmp_path / "src" / "binary.bin").write_bytes(b"\x00\x01") + store = RagStore(":memory:") + try: + rag = ImpactRAG(store, FakeEmbedder()) + n = rag.index_dir("p1", str(tmp_path / "src")) + assert n == 1 + res = rag.retrieve("p1", "OrderController 的影响范围", k=1) + assert res and "OrderController" in res[0] + finally: + store.close() + + +def test_retrieve_empty_query_returns_empty(): + # 空 query(仅空白)直接返回空列表,不调用向量检索 + store = RagStore(":memory:") + try: + rag = ImpactRAG(store, FakeEmbedder()) + rag.index("p1", [("a.java", "创建订单")]) + assert rag.retrieve("p1", " ", k=3) == [] + finally: + store.close() + + +def test_index_empty_text_source_not_indexed(): + # 空文本源:_split 回退为空片段,过滤后无 chunk + store = RagStore(":memory:") + try: + rag = ImpactRAG(store, FakeEmbedder()) + rag.index("p", [("empty.txt", "")]) + assert rag.retrieve("p", "x", k=3) == [] + finally: + store.close() + + +def test_index_dir_non_dir_root_returns_zero(): + # 非目录 root:is_dir() 为 False,返回 0 + store = RagStore(":memory:") + try: + rag = ImpactRAG(store, FakeEmbedder()) + assert rag.index_dir("p", "不存在的路径") == 0 + finally: + store.close() + + +def test_index_dir_skips_empty_file(tmp_path): + # 空内容 .txt 文件不应被索引 + (tmp_path / "empty.txt").write_text("", encoding="utf-8") + store = RagStore(":memory:") + try: + rag = ImpactRAG(store, FakeEmbedder()) + assert rag.index_dir("p", str(tmp_path)) == 0 + finally: + store.close() + + +def test_index_dir_skips_read_error(tmp_path, monkeypatch): + # 读取异常的文件被 except 跳过,整体不崩 + (tmp_path / "a.txt").write_text("创建订单", encoding="utf-8") + monkeypatch.setattr( + __import__("pathlib").Path, + "read_text", + lambda *a, **k: (_ for _ in ()).throw(Exception), + ) + store = RagStore(":memory:") + try: + rag = ImpactRAG(store, FakeEmbedder()) + assert rag.index_dir("p", str(tmp_path)) == 0 + finally: + store.close() + + +def test_index_dir_uppercase_ext_indexed(tmp_path): + # 大写扩展名(.JAVA)应经 .lower() 命中白名单并被索引 + (tmp_path / "Foo.JAVA").write_text( + "public class Foo { 创建订单 }", encoding="utf-8" + ) + store = RagStore(":memory:") + try: + rag = ImpactRAG(store, FakeEmbedder()) + n = rag.index_dir("p", str(tmp_path)) + assert n == 1 + res = rag.retrieve("p", "创建订单", k=1) + assert res and "Foo" in res[0] + finally: + store.close() + + +def test_index_dir_empty_dir_returns_zero(tmp_path): + # 目录内无文本文件(sources 为空)返回 0 + store = RagStore(":memory:") + try: + rag = ImpactRAG(store, FakeEmbedder()) + assert rag.index_dir("p", str(tmp_path)) == 0 + finally: + store.close() + + +def test_index_dir_skips_nul_binary_text(tmp_path): + # 含 NUL 字节的 .txt 视为二进制误带扩展名,应跳过 + (tmp_path / "bad.txt").write_bytes(b"hello\x00world") + store = RagStore(":memory:") + try: + rag = ImpactRAG(store, FakeEmbedder()) + assert rag.index_dir("p", str(tmp_path)) == 0 + finally: + store.close() + + +def test_index_splits_long_single_paragraph(): + # 单段落超长:触发 len(buf)+len(p) > size 且 buf 为空分支 -> 整体作为 1 个 chunk + store = RagStore(":memory:") + try: + rag = ImpactRAG(store, FakeEmbedder()) + rag.index("p", [("long.txt", "丙" * 900)]) + chunks = rag.retrieve("p", "丙", k=5) + assert len(chunks) == 1 + finally: + store.close() + + +def test_index_splits_multiple_paragraphs(): + # 多段落(含空行):触发拆行、空白行过滤、and buf 真分支 -> 2 个 chunk + store = RagStore(":memory:") + try: + rag = ImpactRAG(store, FakeEmbedder()) + text = "甲" * 500 + "\n\n" + "乙" * 500 + rag.index("p", [("multi.txt", text)]) + chunks = rag.retrieve("p", "甲", k=5) + assert len(chunks) == 2 + finally: + store.close() diff --git a/tests/test_impact_rag_e2e.py b/tests/test_impact_rag_e2e.py new file mode 100644 index 0000000..12cbb17 --- /dev/null +++ b/tests/test_impact_rag_e2e.py @@ -0,0 +1,224 @@ +"""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 json +import zipfile +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from genesis.server.app import create_app + +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() + + +def test_service_rag_lazy_engine_build(tmp_path, monkeypatch): + """覆盖 service.py:233-236:构造未传 engine 且 run_impact(use_rag=True) 时懒构建引擎。""" + store = SessionStore(db_path=str(tmp_path / "s.db")) + rag = ImpactRAG(RagStore(str(tmp_path / "rag.db")), FakeEmbedder()) + # 注入的懒构建产物:一个会捕获 prompt 的 FakeEngine + injected_engine = FakeEngine() + monkeypatch.setattr( + "genesis.inference.factory.build_inference_engine", lambda: injected_engine + ) + svc = GenesisService( + store=store, + data_root=str(tmp_path / "data"), + rag=rag, + use_rag=False, # 构造时不传 engine、RAG 默认关闭 + ) + assert svc.engine is None # 尚未构建 + + 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),existing_system 含 KnownOrder 源码 + svc.upload_file(sid, "existing_system", "existing.zip", _make_existing_zip()) + svc.run_parse(sid) + svc.confirm_parse(sid) + assert svc.get_session(sid).status == "impact_running" + + # 显参开启 RAG → 触发 engine 为 None 的懒构建分支 + rec = asyncio.run(svc.run_impact(sid, use_rag=True)) + assert rec.status == "awaiting_impact_confirm" + # 懒构建的引擎被注入并实际使用 + assert svc.engine is injected_engine + summary = json.loads(rec.impact_summary) + assert summary.get("rag_enabled") is True + # 引擎捕获的 prompt 含 RAG 检索注入的 KnownOrder 片段 + assert injected_engine.captured is not None + assert "KnownOrder" in injected_engine.captured + + +def test_http_start_impact_use_rag_e2e(tmp_path): + """覆盖 app.py start_impact 异步端点 + _FakeEngine.chat_structured(RAG 路径,HTTP 级)。""" + store = SessionStore(db_path=str(tmp_path / "s.db")) + rag = ImpactRAG(RagStore(str(tmp_path / "rag.db")), FakeEmbedder()) + app = create_app( + store=store, + data_root=str(tmp_path / "data"), + engine="fake", # 触发内部 _FakeEngine(异步 chat_structured) + rag=rag, + ) + client = TestClient(app) + + sid = client.post("/api/sessions", json={"user_id": "u1"}).json()["session_id"] + # multipart 上传:要件/模板/既有系统 + client.post(f"/api/sessions/{sid}/files", + data={"file_type": "requirements"}, + files={"file": ("requirements_newdev.xlsx", _sample_req())}) + client.post(f"/api/sessions/{sid}/files", + data={"file_type": "template"}, + files={"file": ("template_design_ja.docx", _sample_tpl())}) + client.post(f"/api/sessions/{sid}/files", + data={"file_type": "existing_system"}, + files={"file": ("existing.zip", _make_existing_zip())}) + + assert client.post(f"/api/sessions/{sid}/start-parse").status_code == 200 + # 有既有系统 → impact_running + assert client.post(f"/api/sessions/{sid}/confirm-parse").json()["status"] == "impact_running" + + # 异步端点:start-impact?use_rag=True(显参开启 RAG) + r = client.post(f"/api/sessions/{sid}/start-impact?use_rag=True") + assert r.status_code == 200 + ir = client.get(f"/api/sessions/{sid}/impact-result").json() + # 经 HTTP 真实走到 RAG 路径,摘要标记 rag_enabled + assert ir.get("rag_enabled") is True diff --git a/tests/test_rag_embeddings.py b/tests/test_rag_embeddings.py new file mode 100644 index 0000000..b1f9770 --- /dev/null +++ b/tests/test_rag_embeddings.py @@ -0,0 +1,30 @@ +# tests/test_rag_embeddings.py +from genesis.rag.embeddings import FakeEmbedder, get_embedder + + +def test_fake_embedder_deterministic(): + e = FakeEmbedder() + a = e.embed(["OrderController 创建订单"])[0] + b = e.embed(["OrderController 创建订单"])[0] + assert a == b + + +def test_fake_embedder_similar_closer_than_unrelated(): + e = FakeEmbedder() + base = e.embed(["OrderController 处理创建订单请求"])[0] + sim = e.embed(["OrderController 保存订单到数据库"])[0] + dif = e.embed(["用户登录认证模块"])[0] + import math + + def cos(x, y): + dot = sum(p * q for p, q in zip(x, y)) + nx = math.sqrt(sum(p * p for p in x)) + ny = math.sqrt(sum(q * q for q in y)) + return dot / (nx * ny or 1.0) + + assert cos(base, sim) > cos(base, dif) + + +def test_get_embedder_fake_engine_returns_fake(): + e = get_embedder("fake") + assert e.__class__.__name__ == "FakeEmbedder" diff --git a/tests/test_rag_store.py b/tests/test_rag_store.py new file mode 100644 index 0000000..e350888 --- /dev/null +++ b/tests/test_rag_store.py @@ -0,0 +1,61 @@ +import math +from genesis.rag.store import RagStore + + +def _vec(*ones): + v = [0.0] * 8 + for i in ones: + v[i] = 1.0 + n = math.sqrt(sum(x * x for x in v)) + return [x / n for x in v] + + +def test_search_returns_most_similar_chunk(): + s = RagStore(":memory:") + s.reset_scope("p1") + s.add("p1", ["订单模块处理创建", "用户认证登录"], [_vec(0, 1), _vec(4, 5)]) + res = s.search("p1", _vec(0, 1), k=1) + assert res == ["订单模块处理创建"] + # k<=0 健壮性 + assert s.search("p1", _vec(0, 1), k=0) == [] + s.close() + + +def test_reset_scope_clears(): + s = RagStore(":memory:") + s.reset_scope("p1") + s.add("p1", ["a"], [_vec(0)]) + s.reset_scope("p1") + assert s.search("p1", _vec(0), k=3) == [] + # chunks 与 embeddings 长度不一致应抛 ValueError + try: + import pytest + with pytest.raises(ValueError): + s.add("p1", ["a", "b"], [_vec(0)]) + finally: + s.close() + + +def test_concurrent_add_and_search_no_crash(): + import threading + s = RagStore(":memory:") + s.reset_scope("p1") + + def worker(i): + # 每个 worker 线程内交替执行 add 与 search,让读写真正并发跨线程运行 + for j in range(5): + s.add("p1", [f"chunk-{i}-{j}"], [_vec(i % 8)]) + res = s.search("p1", _vec(i % 8), k=3) + assert all(isinstance(r, str) for r in res) + + try: + threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + res = s.search("p1", _vec(0), k=3) + assert len(res) <= 3 + assert all(isinstance(r, str) for r in res) + finally: + s.close() diff --git a/tests/test_server_service.py b/tests/test_server_service.py index e9215a1..c2814c3 100644 --- a/tests/test_server_service.py +++ b/tests/test_server_service.py @@ -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):