From 9bc7828e63a192c69dec7cfca1b4ab9f980cb0fb Mon Sep 17 00:00:00 2001 From: lhl Date: Sat, 29 Aug 2026 22:27:24 +0800 Subject: [PATCH 01/11] =?UTF-8?q?feat(rag):=20=E6=96=B0=E5=A2=9E=20Embedde?= =?UTF-8?q?r=20=E6=8A=BD=E8=B1=A1=E4=B8=8E=E7=A6=BB=E7=BA=BF=20FakeEmbedde?= =?UTF-8?q?r?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/genesis/rag/__init__.py | 0 src/genesis/rag/embeddings.py | 51 +++++++++++++++++++++++++++++++++++ tests/test_rag_embeddings.py | 30 +++++++++++++++++++++ 3 files changed, 81 insertions(+) create mode 100644 src/genesis/rag/__init__.py create mode 100644 src/genesis/rag/embeddings.py create mode 100644 tests/test_rag_embeddings.py 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..c8ba40e --- /dev/null +++ b/src/genesis/rag/embeddings.py @@ -0,0 +1,51 @@ +from __future__ import annotations + +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 = [] + cur = "" + for ch in text.lower(): + if ch.isalnum(): + cur += ch + else: + if cur: + toks.append(cur) + cur = "" + if cur: + toks.append(cur) + out = [] + for t in toks: + idx = 0 + for i, c in enumerate(t): + if i > 0 and c.isupper(): + out.append(t[idx:i]) + idx = i + out.append(t[idx:]) + return [x for x in out if x] + + +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 = __import__("hashlib").md5(tok.encode("utf-8")).digest() + idx = h[0] % _DIM + v[idx] += 1.0 + norm = __import__("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) -> Embedder: + return FakeEmbedder() 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" From afc09fc908dbbfa7e655d5a27662b3baa562a19d Mon Sep 17 00:00:00 2001 From: lhl Date: Sat, 29 Aug 2026 22:35:16 +0800 Subject: [PATCH 02/11] =?UTF-8?q?fix(rag):=20=E7=A7=BB=E9=99=A4=20embeddin?= =?UTF-8?q?gs=20camelCase=20=E6=AD=BB=E4=BB=A3=E7=A0=81=EF=BC=8C=E8=A6=86?= =?UTF-8?q?=E7=9B=96=E7=8E=87=E8=BE=BE=20100%?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/genesis/rag/embeddings.py | 34 +++++++++++----------------------- 1 file changed, 11 insertions(+), 23 deletions(-) diff --git a/src/genesis/rag/embeddings.py b/src/genesis/rag/embeddings.py index c8ba40e..6c59ff1 100644 --- a/src/genesis/rag/embeddings.py +++ b/src/genesis/rag/embeddings.py @@ -1,5 +1,8 @@ from __future__ import annotations +import hashlib +import math +import re from typing import List, Protocol @@ -11,26 +14,9 @@ _DIM = 64 def _tokenize(text: str) -> List[str]: - toks = [] - cur = "" - for ch in text.lower(): - if ch.isalnum(): - cur += ch - else: - if cur: - toks.append(cur) - cur = "" - if cur: - toks.append(cur) - out = [] - for t in toks: - idx = 0 - for i, c in enumerate(t): - if i > 0 and c.isupper(): - out.append(t[idx:i]) - idx = i - out.append(t[idx:]) - return [x for x in out if x] + # 分词意图：先将文本统一转为小写，再按非字母数字字符切分为词元，最后过滤空串 + toks = re.split(r"\W+", text.lower()) + return [t for t in toks if t] class FakeEmbedder: @@ -39,13 +25,15 @@ class FakeEmbedder: for t in texts: v = [0.0] * _DIM for tok in _tokenize(t): - h = __import__("hashlib").md5(tok.encode("utf-8")).digest() + h = hashlib.md5(tok.encode("utf-8")).digest() idx = h[0] % _DIM v[idx] += 1.0 - norm = __import__("math").sqrt(sum(x * x for x in v)) or 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) -> Embedder: +def get_embedder(engine: str) -> Embedder: + # engine 当前未使用，统一回退到 FakeEmbedder；真实向量模型（如 OpenAI/BGE）接入点预留于此 return FakeEmbedder() From 2eab0183ee62c8113b0373430e7c0e4acb40e7f3 Mon Sep 17 00:00:00 2001 From: lhl Date: Sat, 29 Aug 2026 22:39:37 +0800 Subject: [PATCH 03/11] =?UTF-8?q?feat(rag):=20=E6=96=B0=E5=A2=9E=20RagStor?= =?UTF-8?q?e=EF=BC=88SQLite=20=E5=90=91=E9=87=8F=E5=AD=98=E5=82=A8=20+=20?= =?UTF-8?q?=E4=BD=99=E5=BC=A6=E6=A3=80=E7=B4=A2=EF=BC=8C=E7=BA=BF=E7=A8=8B?= =?UTF-8?q?=E5=AE=89=E5=85=A8=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/genesis/rag/store.py | 52 ++++++++++++++++++++++++++++++++++++++++ tests/test_rag_store.py | 44 ++++++++++++++++++++++++++++++++++ 2 files changed, 96 insertions(+) create mode 100644 src/genesis/rag/store.py create mode 100644 tests/test_rag_store.py diff --git a/src/genesis/rag/store.py b/src/genesis/rag/store.py new file mode 100644 index 0000000..2a05a72 --- /dev/null +++ b/src/genesis/rag/store.py @@ -0,0 +1,52 @@ +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: + # 先清后写（增量索引），整段加锁串行化 + 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]: + 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]] diff --git a/tests/test_rag_store.py b/tests/test_rag_store.py new file mode 100644 index 0000000..c96315d --- /dev/null +++ b/tests/test_rag_store.py @@ -0,0 +1,44 @@ +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 == ["订单模块处理创建"] + + +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) == [] + + +def test_concurrent_add_and_search_no_crash(): + import threading + s = RagStore(":memory:") + s.reset_scope("p1") + + def worker(i): + s.add("p1", [f"chunk-{i}"], [_vec(i % 8)]) + + 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) From 69a7ba722622ba44b674d55d28f8fac156187eb2 Mon Sep 17 00:00:00 2001 From: lhl Date: Sat, 29 Aug 2026 22:45:17 +0800 Subject: [PATCH 04/11] =?UTF-8?q?fix(rag):=20=E5=BC=BA=E5=8C=96=20RagStore?= =?UTF-8?q?=20=E5=B9=B6=E5=8F=91=E8=AF=BB=E5=86=99=E9=AA=8C=E8=AF=81=20+?= =?UTF-8?q?=20close/=E5=81=A5=E5=A3=AE=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/genesis/rag/store.py | 11 ++++++++++- tests/test_rag_store.py | 35 ++++++++++++++++++++++++++--------- 2 files changed, 36 insertions(+), 10 deletions(-) diff --git a/src/genesis/rag/store.py b/src/genesis/rag/store.py index 2a05a72..4c4ac09 100644 --- a/src/genesis/rag/store.py +++ b/src/genesis/rag/store.py @@ -31,7 +31,9 @@ class RagStore: 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( @@ -41,6 +43,10 @@ class RagStore: 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,) @@ -50,3 +56,6 @@ class RagStore: 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/tests/test_rag_store.py b/tests/test_rag_store.py index c96315d..e350888 100644 --- a/tests/test_rag_store.py +++ b/tests/test_rag_store.py @@ -16,6 +16,9 @@ def test_search_returns_most_similar_chunk(): 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(): @@ -24,6 +27,13 @@ def test_reset_scope_clears(): 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(): @@ -32,13 +42,20 @@ def test_concurrent_add_and_search_no_crash(): s.reset_scope("p1") def worker(i): - s.add("p1", [f"chunk-{i}"], [_vec(i % 8)]) + # 每个 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) - 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) + 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() From 6f0cb2f79207c74ee28d1615928828b2e7e24bf7 Mon Sep 17 00:00:00 2001 From: lhl Date: Sat, 29 Aug 2026 22:48:42 +0800 Subject: [PATCH 05/11] =?UTF-8?q?feat(rag):=20=E6=96=B0=E5=A2=9E=20ImpactR?= =?UTF-8?q?AG=20=E7=B4=A2=E5=BC=95/=E6=A3=80=E7=B4=A2=E6=9C=8D=E5=8A=A1?= =?UTF-8?q?=EF=BC=88=E5=90=AB=20index=5Fdir=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/genesis/rag/impact_rag.py | 64 +++++++++++++++++++++++++++++++++++ tests/test_impact_rag.py | 35 +++++++++++++++++++ 2 files changed, 99 insertions(+) create mode 100644 src/genesis/rag/impact_rag.py create mode 100644 tests/test_impact_rag.py diff --git a/src/genesis/rag/impact_rag.py b/src/genesis/rag/impact_rag.py new file mode 100644 index 0000000..fa10be8 --- /dev/null +++ b/src/genesis/rag/impact_rag.py @@ -0,0 +1,64 @@ +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): + 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 text.strip(): + 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/tests/test_impact_rag.py b/tests/test_impact_rag.py new file mode 100644 index 0000000..e6deabd --- /dev/null +++ b/tests/test_impact_rag.py @@ -0,0 +1,35 @@ +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() From 9037ea6f4befeb61987fd6137ca2f904619cf61a Mon Sep 17 00:00:00 2001 From: lhl Date: Sat, 29 Aug 2026 22:55:59 +0800 Subject: [PATCH 06/11] =?UTF-8?q?fix(rag):=20=E8=A1=A5=E5=85=85=20ImpactRA?= =?UTF-8?q?G=20=E5=88=86=E6=94=AF=E6=B5=8B=E8=AF=95=E4=B8=8E=E9=98=B2?= =?UTF-8?q?=E5=BE=A1=EF=BC=8C=E8=A6=86=E7=9B=96=E7=8E=87=E8=BE=BE=20100%?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/genesis/rag/impact_rag.py | 16 +++-- tests/test_impact_rag.py | 121 ++++++++++++++++++++++++++++++++++ 2 files changed, 133 insertions(+), 4 deletions(-) diff --git a/src/genesis/rag/impact_rag.py b/src/genesis/rag/impact_rag.py index fa10be8..ab14a2d 100644 --- a/src/genesis/rag/impact_rag.py +++ b/src/genesis/rag/impact_rag.py @@ -36,12 +36,14 @@ class ImpactRAG: chunks = [] for name, text in sources: for piece in _split(text): - chunks.append(f"[{name}]\n{piece}") + # 跳过空片段，避免产生空向量噪声 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：索引目录内文本源文件（既有系统源码）。返回索引的片段数。""" + """D1：索引目录内文本源文件（既有系统源码）。返回索引的文件数。""" root_path = Path(root) sources: List[Tuple[str, str]] = [] if root_path.is_dir(): @@ -50,9 +52,15 @@ class ImpactRAG: try: text = p.read_text(encoding="utf-8", errors="ignore") except Exception: + # 读取异常的文件跳过，不中断整体索引 continue - if text.strip(): - sources.append((str(p.relative_to(root_path)), text)) + # 空内容文件跳过 + 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) diff --git a/tests/test_impact_rag.py b/tests/test_impact_rag.py index e6deabd..dfab69c 100644 --- a/tests/test_impact_rag.py +++ b/tests/test_impact_rag.py @@ -33,3 +33,124 @@ def test_index_dir_reads_text_files(tmp_path): 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() From 974bf12dc0acc5dbb6f97be70ca1d5c9a3274cec Mon Sep 17 00:00:00 2001 From: lhl Date: Sat, 29 Aug 2026 23:05:53 +0800 Subject: [PATCH 07/11] =?UTF-8?q?feat(rag):=20ImpactAgent=20=E6=8E=A5?= =?UTF-8?q?=E5=85=A5=E5=8F=AF=E9=80=89=20RAG=20=E4=B8=8A=E4=B8=8B=E6=96=87?= =?UTF-8?q?=EF=BC=88use=5Frag=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/genesis/impact/impact_agent.py | 72 ++++++++++++++++++++++++- tests/test_impact_agent_rag.py | 85 ++++++++++++++++++++++++++++++ 2 files changed, 156 insertions(+), 1 deletion(-) create mode 100644 tests/test_impact_agent_rag.py diff --git a/src/genesis/impact/impact_agent.py b/src/genesis/impact/impact_agent.py index bf5e509..5554b22 100644 --- a/src/genesis/impact/impact_agent.py +++ b/src/genesis/impact/impact_agent.py @@ -84,8 +84,78 @@ 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" + ) + + 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 小节，向后兼容）。 + """ + 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 self.engine.chat_structured( + session_id=session_id, + prompt=prompt, + variables={}, + schema={}, + ) def run( self, diff --git a/tests/test_impact_agent_rag.py b/tests/test_impact_agent_rag.py new file mode 100644 index 0000000..74a059d --- /dev/null +++ b/tests/test_impact_agent_rag.py @@ -0,0 +1,85 @@ +"""ImpactAgent RAG 上下文注入测试（RAG 迭代 Task 4）。 + +验证： +- use_rag=True 时，run_impact 发送给 LLM 的 prompt 文本包含 RAG 检索命中片段与明确小节标题。 +- use_rag=False（或默认）时，prompt 文本不含 RAG 小节标题（向后兼容）。 +""" +from genesis.impact.impact_agent import ImpactAgent +from genesis.rag.embeddings import FakeEmbedder +from genesis.rag.impact_rag import ImpactRAG +from genesis.rag.store import RagStore + + +_RAG_SECTION_TITLE = "# 既有系统关联上下文（RAG 检索，辅助判断影响范围）" + + +class FakeEngine: + """捕获真实 LLM 方法（chat_structured）收到的 prompt 文本。 + + 方法名与签名刻意复用本仓库 InferenceEngine.chat_structured 的形参风格， + 以保证 mock 的是真实接口（key=session_id/prompt/variables/schema/retry_count）。 + """ + + def __init__(self) -> None: + self.last_prompt: str | None = None + self.calls = 0 + + def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2): + self.last_prompt = prompt + self.calls += 1 + # 返回结构兼容 ChatResult 的最小占位（测试仅校验 prompt 注入） + return {"session_id": session_id, "prompt": prompt} + + +def _make_rag(session_id: str, sources): + store = RagStore(":memory:") + rag = ImpactRAG(store, FakeEmbedder()) + rag.index(session_id, sources) + return store, rag + + +def test_run_impact_with_rag_injects_context(): + session_id = "sess-rag" + store, rag = _make_rag(session_id, [("TradeApplication.java", "订单创建调用 MyBatis")]) + try: + engine = FakeEngine() + agent = ImpactAgent(engine=engine, rag=rag, use_rag=True) + agent.run_impact(session_id, requirements_text="创建订单的影响", k=5) + prompt = engine.last_prompt + assert prompt is not None + # 命中片段（含文件名 TradeApplication.java）被注入 + assert "TradeApplication" in prompt + # 明确小节标题被注入 + assert _RAG_SECTION_TITLE in prompt + finally: + store.close() + + +def test_run_impact_without_rag_no_context(): + session_id = "sess-no-rag" + store, rag = _make_rag(session_id, [("TradeApplication.java", "订单创建调用 MyBatis")]) + try: + engine = FakeEngine() + agent = ImpactAgent(engine=engine, rag=rag, use_rag=False) + agent.run_impact(session_id, requirements_text="创建订单的影响", k=5) + prompt = engine.last_prompt + # 显式关闭 RAG：不含小节标题，也不含检索片段 + assert _RAG_SECTION_TITLE not in prompt + assert "TradeApplication" not in prompt + finally: + store.close() + + +def test_run_impact_default_no_rag_no_context(): + # 默认 use_rag 为 False（未显式开启），行为与关闭一致 + session_id = "sess-default" + store, rag = _make_rag(session_id, [("TradeApplication.java", "订单创建调用 MyBatis")]) + try: + engine = FakeEngine() + agent = ImpactAgent(engine=engine, rag=rag) # 不传 use_rag + agent.run_impact(session_id, requirements_text="创建订单的影响", k=5) + prompt = engine.last_prompt + assert _RAG_SECTION_TITLE not in prompt + assert "TradeApplication" not in prompt + finally: + store.close() From 9514749b9fdc8ab2a9be794fc425553f32f940c9 Mon Sep 17 00:00:00 2001 From: lhl Date: Sat, 29 Aug 2026 23:16:12 +0800 Subject: [PATCH 08/11] =?UTF-8?q?fix(rag):=20run=5Fimpact=20=E6=94=B9?= =?UTF-8?q?=E4=B8=BA=20async=20=E5=B9=B6=20await=20=E5=BC=95=E6=93=8E?= =?UTF-8?q?=EF=BC=88StructuredResult=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/genesis/impact/impact_agent.py | 19 +++-- tests/test_impact_agent_rag.py | 112 +++++++++++++++-------------- 2 files changed, 67 insertions(+), 64 deletions(-) diff --git a/src/genesis/impact/impact_agent.py b/src/genesis/impact/impact_agent.py index 5554b22..2012146 100644 --- a/src/genesis/impact/impact_agent.py +++ b/src/genesis/impact/impact_agent.py @@ -121,36 +121,33 @@ class ImpactAgent: f"{requirements_text}\n" ) - def run_impact( + async def run_impact( self, session_id: str, requirements_text: str, use_rag: bool | None = None, k: int = 5, ): - """LLM 驱动的变更影响分析（可选 RAG 上下文注入）。 + """LLM 驱动的变更影响分析（可选 RAG 上下文注入，异步）。 - - use_rag 优先取显参；为 None 时回退到实例级 ``self.use_rag``。 - - 启用且 ``self.rag`` 存在时，以 ``影响调查：`` + 要件前若干字 为查询， - 调用 ``self.rag.retrieve(session_id, query, k)``，将命中片段注入 prompt。 - - use_rag=False 时 prompt 内容与原版完全一致（不含 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 self.engine.chat_structured( + return await self.engine.chat_structured( session_id=session_id, prompt=prompt, variables={}, diff --git a/tests/test_impact_agent_rag.py b/tests/test_impact_agent_rag.py index 74a059d..89748e5 100644 --- a/tests/test_impact_agent_rag.py +++ b/tests/test_impact_agent_rag.py @@ -1,85 +1,91 @@ -"""ImpactAgent RAG 上下文注入测试（RAG 迭代 Task 4）。 +"""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。 """ -from genesis.impact.impact_agent import ImpactAgent +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 -_RAG_SECTION_TITLE = "# 既有系统关联上下文（RAG 检索，辅助判断影响范围）" - - class FakeEngine: - """捕获真实 LLM 方法（chat_structured）收到的 prompt 文本。 + """捕获真实 LLM 方法（chat_structured，async）收到的 prompt 文本。 方法名与签名刻意复用本仓库 InferenceEngine.chat_structured 的形参风格， 以保证 mock 的是真实接口（key=session_id/prompt/variables/schema/retry_count）。 """ def __init__(self) -> None: - self.last_prompt: str | None = None + self.captured: str | None = None self.calls = 0 - def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2): - self.last_prompt = prompt + async def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2): + self.captured = prompt self.calls += 1 - # 返回结构兼容 ChatResult 的最小占位（测试仅校验 prompt 注入） - return {"session_id": session_id, "prompt": prompt} + # 返回结构兼容 StructuredResult 的最小占位（含 data/raw_text） + return types.SimpleNamespace(data={}, raw_text=prompt) -def _make_rag(session_id: str, sources): +def _make_rag(session_id: str, text: str) -> ImpactRAG: store = RagStore(":memory:") rag = ImpactRAG(store, FakeEmbedder()) - rag.index(session_id, sources) - return store, rag + rag.index(session_id, [("TradeApplication.java", text)]) + return rag def test_run_impact_with_rag_injects_context(): - session_id = "sess-rag" - store, rag = _make_rag(session_id, [("TradeApplication.java", "订单创建调用 MyBatis")]) - try: - engine = FakeEngine() - agent = ImpactAgent(engine=engine, rag=rag, use_rag=True) - agent.run_impact(session_id, requirements_text="创建订单的影响", k=5) - prompt = engine.last_prompt - assert prompt is not None - # 命中片段（含文件名 TradeApplication.java）被注入 - assert "TradeApplication" in prompt - # 明确小节标题被注入 - assert _RAG_SECTION_TITLE in prompt - finally: - store.close() + 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(): - session_id = "sess-no-rag" - store, rag = _make_rag(session_id, [("TradeApplication.java", "订单创建调用 MyBatis")]) - try: - engine = FakeEngine() - agent = ImpactAgent(engine=engine, rag=rag, use_rag=False) - agent.run_impact(session_id, requirements_text="创建订单的影响", k=5) - prompt = engine.last_prompt - # 显式关闭 RAG：不含小节标题，也不含检索片段 - assert _RAG_SECTION_TITLE not in prompt - assert "TradeApplication" not in prompt - finally: - store.close() + 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_default_no_rag_no_context(): - # 默认 use_rag 为 False（未显式开启），行为与关闭一致 - session_id = "sess-default" - store, rag = _make_rag(session_id, [("TradeApplication.java", "订单创建调用 MyBatis")]) - try: - engine = FakeEngine() - agent = ImpactAgent(engine=engine, rag=rag) # 不传 use_rag - agent.run_impact(session_id, requirements_text="创建订单的影响", k=5) - prompt = engine.last_prompt - assert _RAG_SECTION_TITLE not in prompt - assert "TradeApplication" not in prompt - finally: - store.close() +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")) From f77fab707d445841a3d3ea333f152d002797a1bb Mon Sep 17 00:00:00 2001 From: lhl Date: Sun, 30 Aug 2026 00:25:24 +0800 Subject: [PATCH 09/11] =?UTF-8?q?feat(rag):=20=E6=9C=8D=E5=8A=A1=E5=B1=82?= =?UTF-8?q?=E6=8E=A5=E7=BA=BF=EF=BC=88=E4=B8=8A=E4=BC=A0=E5=8D=B3=E7=B4=A2?= =?UTF-8?q?=E5=BC=95=20+=20use=5Frag=20=E5=BC=82=E6=AD=A5=E9=93=BE?= =?UTF-8?q?=E8=B7=AF=20+=20=E5=BC=BA=20e2e=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/design.md | 12 +++ src/genesis/chat/agent.py | 5 +- src/genesis/server/app.py | 21 ++++- src/genesis/server/service.py | 48 ++++++++++- tests/test_impact_rag_e2e.py | 146 ++++++++++++++++++++++++++++++++++ tests/test_server_service.py | 5 +- 6 files changed, 228 insertions(+), 9 deletions(-) create mode 100644 tests/test_impact_rag_e2e.py diff --git a/docs/design.md b/docs/design.md index e32e6d0..a8d998b 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/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..bbed852 100644 --- a/src/genesis/server/service.py +++ b/src/genesis/server/service.py @@ -13,6 +13,7 @@ from __future__ import annotations import html import json +import logging import shutil import zipfile from pathlib import Path @@ -22,6 +23,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 +60,10 @@ class GenesisService: prompt_registry=None, samples_dir: str = "sample", projects: "ProjectsStore | None" = None, + rag: "ImpactRAG | None" = None, + use_rag: bool = False, + rag_db_path: str | None = None, + embedder=None, ) -> None: self.store = store self.data_root = Path(data_root) @@ -64,6 +71,14 @@ 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 + self.rag_db_path = rag_db_path + if embedder is None: + from genesis.rag.embeddings import get_embedder + embedder = get_embedder("fake") + self.embedder = embedder # ---------- 会话与文件 ---------- @@ -126,6 +141,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 +225,35 @@ 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: + 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_rag_e2e.py b/tests/test_impact_rag_e2e.py new file mode 100644 index 0000000..8866b3f --- /dev/null +++ b/tests/test_impact_rag_e2e.py @@ -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() 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): From 9a02fcd00f3b5d32504b6e0c921c6fd77ef5b03e Mon Sep 17 00:00:00 2001 From: lhl Date: Sun, 30 Aug 2026 00:31:18 +0800 Subject: [PATCH 10/11] =?UTF-8?q?fix(rag):=20=E7=A7=BB=E9=99=A4=20service?= =?UTF-8?q?=20=E6=9C=AA=E7=94=A8=E5=8F=82=E6=95=B0=20+=20=E5=BC=95?= =?UTF-8?q?=E6=93=8E=E6=9E=84=E5=BB=BA=E5=8A=A0=E9=94=81=20+=20=E4=BF=AE?= =?UTF-8?q?=E6=AD=A3=E6=96=87=E6=A1=A3=E6=8E=AA=E8=BE=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/design.md | 2 +- src/genesis/server/service.py | 16 +++++++--------- 2 files changed, 8 insertions(+), 10 deletions(-) diff --git a/docs/design.md b/docs/design.md index a8d998b..76ba672 100644 --- a/docs/design.md +++ b/docs/design.md @@ -1930,6 +1930,6 @@ Document（注入后 Word 文档） - **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(...)` 包裹以兼容同步消息处理。 +- **异步链路**：`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/server/service.py b/src/genesis/server/service.py index bbed852..6bbce6c 100644 --- a/src/genesis/server/service.py +++ b/src/genesis/server/service.py @@ -15,6 +15,7 @@ import html import json import logging import shutil +import threading import zipfile from pathlib import Path @@ -62,8 +63,6 @@ class GenesisService: projects: "ProjectsStore | None" = None, rag: "ImpactRAG | None" = None, use_rag: bool = False, - rag_db_path: str | None = None, - embedder=None, ) -> None: self.store = store self.data_root = Path(data_root) @@ -74,11 +73,8 @@ class GenesisService: # RAG：rag 为 None 表示不启用（向后兼容默认关闭） self.rag = rag self.use_rag = use_rag - self.rag_db_path = rag_db_path - if embedder is None: - from genesis.rag.embeddings import get_embedder - embedder = get_embedder("fake") - self.embedder = embedder + # 引擎构建锁：避免 worker 线程并发下共享可变属性的竞态（D2） + self._engine_lock = threading.Lock() # ---------- 会话与文件 ---------- @@ -234,8 +230,10 @@ class GenesisService: if eff and self.rag is not None: # RAG 路径：LLM 驱动，检索既有系统上下文注入 prompt if self.engine is None: - from genesis.inference.factory import build_inference_engine - self.engine = build_inference_engine() + 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( From ee224e7c56b84e6644d96cf0f5111a0622c3e114 Mon Sep 17 00:00:00 2001 From: lhl Date: Sun, 30 Aug 2026 00:44:49 +0800 Subject: [PATCH 11/11] =?UTF-8?q?test(rag):=20=E8=A1=A5=E5=85=85=E5=BC=95?= =?UTF-8?q?=E6=93=8E=E6=87=92=E6=9E=84=E5=BB=BA=E4=B8=8E=20HTTP=20?= =?UTF-8?q?=E7=BA=A7=20RAG=20e2e=20=E8=A6=86=E7=9B=96=EF=BC=8C=E6=81=A2?= =?UTF-8?q?=E5=A4=8D=2099%=20=E9=97=A8=E7=A6=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_impact_rag_e2e.py | 78 ++++++++++++++++++++++++++++++++++++ 1 file changed, 78 insertions(+) diff --git a/tests/test_impact_rag_e2e.py b/tests/test_impact_rag_e2e.py index 8866b3f..12cbb17 100644 --- a/tests/test_impact_rag_e2e.py +++ b/tests/test_impact_rag_e2e.py @@ -8,10 +8,14 @@ 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 @@ -144,3 +148,77 @@ def test_rag_e2e_use_rag_false_backward_compat(tmp_path): 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