langchain added:

app/main.py

@@ -12,7 +12,6 @@ from app.api.routes_review import router as review_router
 from app.core.config import settings
 from app.core.logging import configure_logging
 from app.db.sqlite import init_db
-from app.rag.bm25 import BM25Index
 from app.rag.ingestion import DocumentIngestionService
 from app.rag.qdrant_store import QdrantVectorStore
 
@@ -46,7 +45,6 @@ def startup() -> None:
     QdrantVectorStore().ensure_collections()
     if settings.auto_ingest_pdfs_on_startup:
         DocumentIngestionService().ingest_pdf_directory(settings.document_dir)
-    BM25Index.load_or_create().save()
 
 
 @app.get("/")

app/rag/bm25.py

@@ -2,74 +2,69 @@ import json
 from pathlib import Path
 from typing import Any
 
-from rank_bm25 import BM25Okapi
+from langchain_community.retrievers import BM25Retriever
+from langchain_core.documents import Document
 
 from app.core.config import settings
-from app.db.sqlite import db
 from app.rag.text import tokenize
 
 
 class BM25Index:
     def __init__(self, docs: list[dict[str, Any]]) -> None:
         self.docs = docs
-        self.tokens = [tokenize(d["text"]) for d in docs]
-        self.index = BM25Okapi(self.tokens) if self.tokens else None
-
-    @classmethod
-    def from_db(cls) -> "BM25Index":
-        with db() as conn:
-            rows = conn.execute(
-                """
-                SELECT c.chunk_id, c.text, c.metadata_json, d.source_name
-                FROM chunks c
-                JOIN documents d ON d.doc_id = c.doc_id
-                """
-            ).fetchall()
-        docs = [
-            {
-                "id": row["chunk_id"],
-                "text": row["text"],
-                "source_name": row["source_name"],
-                "metadata": json.loads(row["metadata_json"]),
-            }
-            for row in rows
+        documents = [
+            Document(
+                page_content=doc["text"],
+                metadata={
+                    **doc.get("metadata", {}),
+                    "id": doc["id"],
+                    "source_name": doc.get("source_name", "unknown"),
+                },
+            )
+            for doc in docs
         ]
-        return cls(docs)
+        self.retriever = BM25Retriever.from_documents(
+            documents,
+            preprocess_func=tokenize,
+        ) if documents else None
 
     @classmethod
     def load_or_create(cls) -> "BM25Index":
         path = Path(settings.bm25_index_path)
         if not path.exists():
-            return cls.from_db()
+            return cls([])
         try:
             payload = json.loads(path.read_text(encoding="utf-8"))
             return cls(payload.get("docs", []))
         except (OSError, json.JSONDecodeError):
-            return cls.from_db()
+            return cls([])
 
     def save(self) -> None:
         path = Path(settings.bm25_index_path)
         path.parent.mkdir(parents=True, exist_ok=True)
         path.write_text(json.dumps({"docs": self.docs}, ensure_ascii=True), encoding="utf-8")
 
-    def rebuild(self) -> None:
-        fresh = self.from_db()
+    def rebuild(self, docs: list[dict[str, Any]] | None = None) -> None:
+        fresh = BM25Index(docs or self.docs)
         self.docs = fresh.docs
-        self.tokens = fresh.tokens
-        self.index = fresh.index
+        self.retriever = fresh.retriever
         self.save()
 
     def search(self, query: str, top_k: int) -> list[dict[str, Any]]:
-        if not self.index or not self.docs:
+        if not self.retriever or not self.docs:
             return []
-        scores = self.index.get_scores(tokenize(query))
-        ranked = sorted(enumerate(scores), key=lambda item: item[1], reverse=True)[:top_k]
-        return [
-            {
-                **self.docs[idx],
-                "score": float(score),
-                "metadata": {**self.docs[idx].get("metadata", {}), "retriever": "bm25"},
-            }
-            for idx, score in ranked
-            if score > 0
-        ]
+        self.retriever.k = top_k
+        results = self.retriever.invoke(query)
+        hits = []
+        for rank, doc in enumerate(results, start=1):
+            metadata = dict(doc.metadata)
+            hits.append(
+                {
+                    "id": str(metadata.get("id", f"bm25-{rank}")),
+                    "text": doc.page_content,
+                    "source_name": str(metadata.get("source_name", "unknown")),
+                    "score": 1.0 / rank,
+                    "metadata": {**metadata, "retriever": "langchain_bm25"},
+                }
+            )
+        return hits

app/rag/ingestion.py

@@ -97,6 +97,7 @@ class DocumentIngestionService:
         split_docs = self.splitter.split_documents(docs)
         chunk_records = []
         new_chunk_texts = []
+        bm25_docs = []
         skipped_chunks = 0
 
         for index, split_doc in enumerate(split_docs):
@@ -113,6 +114,14 @@ class DocumentIngestionService:
                 "source_name": source_name,
                 "text_hash": text_hash,
             }
+            bm25_docs.append(
+                {
+                    "id": chunk_id,
+                    "text": chunk,
+                    "source_name": source_name,
+                    "metadata": chunk_metadata,
+                }
+            )
             if self.cache.chunk_exists(text_hash):
                 skipped_chunks += 1
                 self.cache.save_chunk(chunk_id, doc_id, index, chunk, text_hash, chunk_metadata, embedded=False)
@@ -133,7 +142,7 @@ class DocumentIngestionService:
             )
 
         self.qdrant.upsert_chunks(points)
-        BM25Index.from_db().save()
+        self._merge_bm25_docs(bm25_docs)
 
         return {
             "status": "embedded",
@@ -141,3 +150,14 @@ class DocumentIngestionService:
             "embedded_chunks": len(points),
             "skipped_chunks": skipped_chunks,
         }
+
+    def _merge_bm25_docs(self, docs: list[dict[str, Any]]) -> None:
+        current = BM25Index.load_or_create()
+        by_hash = {
+            str(doc.get("metadata", {}).get("text_hash") or doc.get("id")): doc
+            for doc in current.docs
+        }
+        for doc in docs:
+            key = str(doc.get("metadata", {}).get("text_hash") or doc.get("id"))
+            by_hash[key] = doc
+        BM25Index(list(by_hash.values())).save()