diff --git a/docs/rag_upgrade_audit.md b/docs/rag_upgrade_audit.md new file mode 100644 index 00000000..4480b79e --- /dev/null +++ b/docs/rag_upgrade_audit.md @@ -0,0 +1,67 @@ +# RAG Upgrade Audit — HURC1 + +## Scope + +Review of the offline Copilot retrieval path, document ingestion pipeline, and the central TrustGraph orchestration in `src/lib/services/ai.ts`. + +## High-priority weaknesses found + +1. **Offline Copilot was a demo, not a production retriever.** `rag_engine.py` used four hard-coded records and counted exact token overlaps. It did not load the indexed knowledge base, rank multiple sources, normalize confidence, or return provenance. +2. **The ingestion job did not create a vector index.** `document_ingestion_v2.py` only wrapped whole documents in JSON. There was no semantic chunking, overlap, stable IDs, content hashes, duplicate control, atomic publish, or `latest` pointer. +3. **Grounding can be discarded when memory exists.** In `askAI`, the prompt construction chooses memory context before grounding context. When both are present, the retrieved evidence is omitted from the model input. +4. **Retrieved text is fused as instructions.** The ensemble path concatenates GraphRAG and DocumentRAG responses into a user prompt. Retrieved documents should be treated as untrusted data, bounded in size, and separated from instructions to reduce prompt-injection and context-overflow risk. +5. **RAG provenance is lost at the service boundary.** TrustGraph responses support `sources`, but `askWithRAG` returns only response, intent, and source backend. The UI and audit layer cannot show document code, version, page, retrieval score, or conflicting sources. +6. **Retrieval limits are static and reranking is absent.** DocumentRAG requests up to 20 documents and GraphRAG uses fixed graph limits. There is no query-dependent budget, deduplication, diversity selection, or evidence-quality score before generation. +7. **Intent typing is inconsistent.** The implementation returns `ensemble` through `as any`, while `QueryIntent` does not include this state. This weakens compile-time checks and telemetry consistency. +8. **No retrieval-quality test set.** Existing scripts test connectivity and governance, but there is no golden question set measuring hit rate, citation correctness, abstention, latency, or hallucination rate. + +## Changes implemented in this branch + +### Offline Copilot retrieval + +- Replaced exact overlap counting with a standard-library hybrid retriever: BM25, phrase match, metadata/category bonus, query coverage, and character-trigram similarity. +- Added Unicode and Vietnamese accent normalization. +- Added bounded query size, top-k limits, calibrated confidence, low-confidence refusal, and deterministic JSON errors. +- Added loading from `RAG_KNOWLEDGE_PATH`, `data/vector_db/latest.json`, or the newest versioned index. +- Preserved the existing `answer`, `confidence`, and `engine` response fields and added `sources` plus `knowledge_source`. +- Added provenance fields such as document code, category, matched terms, retrieval score, and metadata. + +### Document ingestion + +- Added paragraph/sentence-aware chunking with overlap. +- Added stable document/chunk IDs and SHA-256 content hashes. +- Added duplicate chunk removal while retaining duplicate-source provenance. +- Removed fabricated metadata defaults such as a fixed issue date and contractor. +- Added broader metro-system classification using title, code, content, asset, and aliases. +- Added immutable versioned indexes, atomic writes, integrity hash, and an atomic `latest.json` pointer. +- Explicitly labels the output as a source index awaiting external vectorization instead of claiming that JSON alone is a vector database. + +### Tests + +Run: + +```bash +python src/lib/ai/test_rag_engine.py +python infra/ai-server/rag/test_document_ingestion_v2.py +``` + +The tests cover accent-insensitive retrieval, abstention on unrelated queries, provenance loading, chunk creation, system classification, latest-pointer generation, duplicate removal, and empty-document rejection. + +## Recommended phase 2 for central TrustGraph RAG + +1. Build a typed `RagEvidence` contract carrying document code, version, page/section, collection, retrieval score, and content hash. +2. Fix prompt assembly so memory and grounding are both included, each with independent token budgets. +3. Treat retrieved content as untrusted evidence: delimit it, strip instruction-like content, and never place it above immutable system policy. +4. Add hybrid retrieval: metadata filters + sparse search + dense search + reranker, followed by diversity selection. +5. Replace fixed limits with a retrieval budget based on query type and available context window. +6. Return citations to the UI and require citation coverage for technical or safety conclusions. +7. Add a golden evaluation set for PSD, AFC, VLD, DNF, hazards, maintenance manuals, and Vietnamese abbreviations. Track Recall@5, MRR, citation precision, abstention precision, p95 latency, and grounded-answer rate. +8. Add ingestion lifecycle controls: document version supersession, soft delete, re-index status, failed-page reporting, and permission-aware collection filters. + +## Acceptance targets + +- Recall@5 ≥ 0.85 on the approved technical question set. +- Citation precision ≥ 0.95 for document-code and page references. +- Abstention precision ≥ 0.90 for questions outside the indexed corpus. +- No answer used for safety/maintenance decisions without at least one retrievable source. +- p95 retrieval latency under 2 seconds locally, excluding final generation. diff --git a/infra/ai-server/rag/document_ingestion_v2.py b/infra/ai-server/rag/document_ingestion_v2.py index 7b9569eb..52b5cd20 100644 --- a/infra/ai-server/rag/document_ingestion_v2.py +++ b/infra/ai-server/rag/document_ingestion_v2.py @@ -1,88 +1,396 @@ -import os +"""Versioned, provenance-aware ingestion for the HURC1 RAG knowledge base. + +Despite the historical filename, this is schema version 3. It creates stable +semantic chunks that can be consumed immediately by the standard-library +fallback retriever and later vectorized by TrustGraph or another embedding +service. The module does not claim to create embeddings locally. +""" + +from __future__ import annotations + +import argparse +import hashlib import json -import uuid -from datetime import datetime +import os +import re +import tempfile +import unicodedata +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Mapping, Sequence + +SCHEMA_VERSION = "3.0" +DEFAULT_CHUNK_SIZE = 1_200 +DEFAULT_CHUNK_OVERLAP = 180 +MIN_CHUNK_SIZE = 80 + +SYSTEM_ALIASES: dict[str, tuple[str, ...]] = { + "AFC": ("afc", "automatic fare collection", "cổng soát vé", "vé tự động", "tvm"), + "PSD": ("psd", "platform screen door", "cửa chắn ke ga", "cửa chắn sân ga"), + "Rolling Stock": ("rolling stock", "đoàn tàu", "toa xe", "train", "traction motor", "bcu"), + "Trackwork": ("trackwork", "đường ray", "ray", "turnout", "ghi đường sắt"), + "Power Supply": ("power supply", "traction power", "điện kéo", "tss", "rtss", "vld"), + "Signalling": ("signalling", "signal", "tín hiệu", "atc", "ats", "interlocking"), + "Telecom": ("telecom", "viễn thông", "radio", "cctv", "pa", "pis"), +} + + +def _normalize_text(value: Any) -> str: + text = unicodedata.normalize("NFKC", str(value or "")) + text = text.replace("\r\n", "\n").replace("\r", "\n") + lines = [re.sub(r"[ \t]+", " ", line).strip() for line in text.split("\n")] + # Preserve paragraph boundaries while removing excessive empty lines. + return re.sub(r"\n{3,}", "\n\n", "\n".join(lines)).strip() + + +def _fold(value: Any) -> str: + text = _normalize_text(value).lower().replace("đ", "d") + decomposed = unicodedata.normalize("NFD", text) + return "".join(ch for ch in decomposed if unicodedata.category(ch) != "Mn") + + +def _sha256(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def _utc_now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _atomic_write_json(path: Path, payload: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + fd, temp_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(payload, handle, ensure_ascii=False, indent=2) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temp_name, path) + except Exception: + try: + os.unlink(temp_name) + except OSError: + pass + raise + class DocumentRAGUpgrade: - def __init__(self, raw_docs_dir="data/raw/docs", index_dir="data/vector_db"): - self.raw_docs_dir = raw_docs_dir - self.index_dir = index_dir - os.makedirs(self.raw_docs_dir, exist_ok=True) - os.makedirs(self.index_dir, exist_ok=True) - - # Danh mục phân loại kỹ thuật Metro chuẩn - self.metro_systems = [ - "AFC", "PSD", "Rolling Stock", "Trackwork", - "Power Supply", "Signalling", "Telecom" - ] + def __init__( + self, + raw_docs_dir: str = "data/raw/docs", + index_dir: str = "data/vector_db", + *, + chunk_size: int = DEFAULT_CHUNK_SIZE, + chunk_overlap: int = DEFAULT_CHUNK_OVERLAP, + ) -> None: + if chunk_size < 300: + raise ValueError("chunk_size must be at least 300 characters") + if chunk_overlap < 0 or chunk_overlap >= chunk_size // 2: + raise ValueError("chunk_overlap must be non-negative and smaller than half of chunk_size") - def add_metadata(self, doc_record): - """ - Bổ sung siêu dữ liệu (Metadata) kỹ thuật chuyên sâu thay vì chỉ có text thô. - """ - # Logic giả lập tự động phân loại hệ thống dựa vào title - title = doc_record.get("title", "").upper() - detected_sys = "General" - for sys in self.metro_systems: - if sys.upper() in title: - detected_sys = sys - break + self.raw_docs_dir = Path(raw_docs_dir) + self.index_dir = Path(index_dir) + self.chunk_size = chunk_size + self.chunk_overlap = chunk_overlap + self.raw_docs_dir.mkdir(parents=True, exist_ok=True) + self.index_dir.mkdir(parents=True, exist_ok=True) + + def _detect_system(self, doc_record: Mapping[str, Any]) -> tuple[str, list[str]]: + haystack = _fold( + "\n".join( + str(doc_record.get(field, "")) + for field in ("title", "code", "content", "asset", "system", "category") + ) + ) + matches: list[str] = [] + for system, aliases in SYSTEM_ALIASES.items(): + if any(_fold(alias) in haystack for alias in aliases): + matches.append(system) + return (matches[0] if matches else "General", matches) - # Gắn metadata - enhanced_doc = { - "id": str(uuid.uuid4()), - "content": doc_record.get("content", ""), - "metadata": { - "document_code": doc_record.get("code", "DOC-UNKNOWN"), - "version": doc_record.get("version", "1.0"), - "issue_date": doc_record.get("issue_date", "2024-01-01"), - "contractor": doc_record.get("contractor", "Hitachi/Nippon Koei"), - "related_system": detected_sys, - "related_asset": doc_record.get("asset", "All") + def add_metadata(self, doc_record: Mapping[str, Any]) -> dict[str, Any]: + """Validate a source record and attach stable, non-fabricated metadata.""" + content = _normalize_text(doc_record.get("content") or doc_record.get("text")) + if not content: + raise ValueError("Document content is empty") + + title = _normalize_text(doc_record.get("title")) or "Untitled document" + document_code = _normalize_text(doc_record.get("code") or doc_record.get("document_code")) + if not document_code: + document_code = f"DOC-{_sha256(title + content)[:12].upper()}" + + version = _normalize_text(doc_record.get("version")) or "unknown" + source_file = _normalize_text(doc_record.get("source_file") or doc_record.get("filename")) or None + related_system, system_matches = self._detect_system(doc_record) + document_hash = _sha256(f"{document_code}\n{version}\n{content}") + + metadata = { + "document_code": document_code, + "title": title, + "version": version, + "issue_date": _normalize_text(doc_record.get("issue_date")) or None, + "contractor": _normalize_text(doc_record.get("contractor")) or None, + "related_system": related_system, + "related_system_matches": system_matches, + "related_asset": _normalize_text( + doc_record.get("asset") or doc_record.get("related_asset") + ) or None, + "source_file": source_file, + "source_page": doc_record.get("page") or doc_record.get("source_page"), + "source_section": _normalize_text( + doc_record.get("section") or doc_record.get("source_section") + ) or None, + "language": _normalize_text(doc_record.get("language")) or "vi", + "document_hash": document_hash, + } + metadata.update( + { + str(key): value + for key, value in dict(doc_record.get("metadata") or {}).items() + if key not in metadata } + ) + + return { + "id": f"doc_{document_hash[:20]}", + "content": content, + "metadata": metadata, } - return enhanced_doc - def create_new_vector_index(self, docs): - """ - Nguyên tắc: Không xóa Index cũ. Tạo Index version mới. + def _split_long_unit(self, unit: str) -> list[str]: + if len(unit) <= self.chunk_size: + return [unit] + pieces: list[str] = [] + start = 0 + while start < len(unit): + end = min(len(unit), start + self.chunk_size) + if end < len(unit): + boundary = max( + unit.rfind(". ", start, end), + unit.rfind("; ", start, end), + unit.rfind(", ", start, end), + unit.rfind(" ", start, end), + ) + if boundary > start + self.chunk_size // 2: + end = boundary + 1 + piece = unit[start:end].strip() + if piece: + pieces.append(piece) + if end >= len(unit): + break + start = max(end - self.chunk_overlap, start + 1) + return pieces + + def chunk_text(self, content: str) -> list[str]: + """Create paragraph/sentence-aware chunks with bounded overlap.""" + content = _normalize_text(content) + if not content: + return [] + + units: list[str] = [] + for paragraph in re.split(r"\n\s*\n", content): + paragraph = paragraph.strip() + if not paragraph: + continue + sentence_units = re.split(r"(?<=[.!?;:])\s+(?=[A-ZÀ-Ỹ0-9])", paragraph) + for sentence in sentence_units: + units.extend(self._split_long_unit(sentence.strip())) + + chunks: list[str] = [] + current = "" + for unit in units: + candidate = f"{current}\n{unit}".strip() if current else unit + if len(candidate) <= self.chunk_size: + current = candidate + continue + + if current: + chunks.append(current) + overlap = current[-self.chunk_overlap:].lstrip() if self.chunk_overlap else "" + current = f"{overlap}\n{unit}".strip() if overlap else unit + else: + chunks.append(unit) + current = "" + + if current: + chunks.append(current) + + if len(chunks) > 1 and len(chunks[-1]) < MIN_CHUNK_SIZE: + tail = chunks.pop() + merged = f"{chunks[-1]}\n{tail}".strip() + if len(merged) <= self.chunk_size + self.chunk_overlap: + chunks[-1] = merged + else: + chunks.append(tail) + return chunks + + def _build_chunks(self, enhanced_doc: Mapping[str, Any]) -> list[dict[str, Any]]: + doc_id = str(enhanced_doc["id"]) + metadata = dict(enhanced_doc["metadata"]) + chunks = self.chunk_text(str(enhanced_doc["content"])) + output: list[dict[str, Any]] = [] + for index, chunk in enumerate(chunks): + content_hash = _sha256(chunk) + chunk_id = f"chunk_{_sha256(f'{doc_id}:{index}:{content_hash}')[:24]}" + chunk_metadata = { + **metadata, + "parent_document_id": doc_id, + "chunk_index": index, + "chunk_count": len(chunks), + "content_hash": content_hash, + "character_count": len(chunk), + } + output.append({"id": chunk_id, "content": chunk, "metadata": chunk_metadata}) + return output + + def create_new_vector_index(self, docs: Sequence[Mapping[str, Any]]) -> str: + """Create an immutable source index and atomically update ``latest.json``. + + The method name is retained for compatibility. The generated file holds + chunked source text and provenance; actual embeddings are expected to be + generated by the configured TrustGraph ingestion flow. """ - version = datetime.now().strftime("%Y%m%d_%H%M") - index_name = f"metro_rag_index_v2_{version}" - - processed_docs = [self.add_metadata(d) for d in docs] - - output_payload = { + if not isinstance(docs, Sequence) or isinstance(docs, (str, bytes)): + raise TypeError("docs must be a sequence of document objects") + + enhanced_documents: list[dict[str, Any]] = [] + rejected: list[dict[str, Any]] = [] + for index, record in enumerate(docs): + try: + if not isinstance(record, Mapping): + raise TypeError("record is not an object") + enhanced_documents.append(self.add_metadata(record)) + except (TypeError, ValueError) as exc: + rejected.append({"position": index, "reason": str(exc)}) + + chunk_by_hash: dict[str, dict[str, Any]] = {} + duplicate_count = 0 + for document in enhanced_documents: + for chunk in self._build_chunks(document): + content_hash = str(chunk["metadata"]["content_hash"]) + existing = chunk_by_hash.get(content_hash) + if existing: + duplicate_count += 1 + duplicate_sources = existing["metadata"].setdefault("duplicate_sources", []) + duplicate_sources.append( + { + "parent_document_id": chunk["metadata"]["parent_document_id"], + "document_code": chunk["metadata"]["document_code"], + "source_file": chunk["metadata"].get("source_file"), + } + ) + else: + chunk_by_hash[content_hash] = chunk + + chunks = list(chunk_by_hash.values()) + created_at = _utc_now() + version = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") + index_name = f"metro_rag_index_v3_{version}" + output_path = self.index_dir / f"{index_name}.json" + + document_summaries = [ + { + "id": doc["id"], + "metadata": doc["metadata"], + "character_count": len(doc["content"]), + "chunk_count": sum( + 1 + for chunk in chunks + if chunk["metadata"].get("parent_document_id") == doc["id"] + ), + } + for doc in enhanced_documents + ] + + payload = { + "schema_version": SCHEMA_VERSION, "index_name": index_name, - "created_at": datetime.now().isoformat(), - "total_documents": len(processed_docs), - "documents": processed_docs + "index_type": "hybrid_lexical_source_index", + "embedding_status": "pending_external_vectorization", + "created_at": created_at, + "configuration": { + "chunk_size": self.chunk_size, + "chunk_overlap": self.chunk_overlap, + }, + "statistics": { + "submitted_documents": len(docs), + "accepted_documents": len(enhanced_documents), + "rejected_documents": len(rejected), + "unique_chunks": len(chunks), + "duplicate_chunks_removed": duplicate_count, + }, + "rejected": rejected, + "documents": document_summaries, + "chunks": chunks, } - - out_path = os.path.join(self.index_dir, f"{index_name}.json") - with open(out_path, "w", encoding="utf-8") as f: - json.dump(output_payload, f, ensure_ascii=False, indent=2) - - print(f"✅ [RAG UPGRADE] Tối ưu hóa xong {len(processed_docs)} tài liệu. Đã lưu vào Vector Index MỚI: {index_name}") - return out_path + _atomic_write_json(output_path, payload) -if __name__ == "__main__": - # Mock data tài liệu O&M - mock_docs = [ - { - "title": "Hướng dẫn bảo trì Cổng soát vé AFC tự động", - "code": "OM-AFC-001", - "content": "Kiểm tra vành răng định kỳ 3 tháng/lần...", - "asset": "GATE" - }, - { - "title": "Cẩm nang xử lý sự cố Tàu Rolling Stock", - "code": "OM-RS-042", - "version": "1.2", - "content": "Nếu má phanh mòn quá 2mm, lập tức thay thế...", - "asset": "TRAIN" + latest_payload = { + "schema_version": SCHEMA_VERSION, + "index_name": index_name, + "filename": output_path.name, + "created_at": created_at, + "sha256": _sha256(output_path.read_text(encoding="utf-8")), } - ] - - rag_upgrader = DocumentRAGUpgrade() - rag_upgrader.create_new_vector_index(mock_docs) + _atomic_write_json(self.index_dir / "latest.json", latest_payload) + + print( + f"[RAG UPGRADE] accepted={len(enhanced_documents)} " + f"chunks={len(chunks)} duplicates={duplicate_count} index={output_path}" + ) + return str(output_path) + + +def _load_input(path: str | None) -> list[Mapping[str, Any]]: + if not path: + return [ + { + "title": "Hướng dẫn bảo trì Cổng soát vé AFC tự động", + "code": "OM-AFC-001", + "content": "Kiểm tra vành răng định kỳ 3 tháng/lần. Ghi nhận kết quả và mã thiết bị.", + "asset": "GATE", + }, + { + "title": "Cẩm nang xử lý sự cố Rolling Stock", + "code": "OM-RS-042", + "version": "1.2", + "content": "Kiểm tra giới hạn mòn má phanh theo cẩm nang được phê duyệt trước khi thay thế.", + "asset": "TRAIN", + }, + ] + + with Path(path).open("r", encoding="utf-8") as handle: + payload = json.load(handle) + if isinstance(payload, list): + records = payload + elif isinstance(payload, Mapping): + records = payload.get("documents") or payload.get("records") or payload.get("docs") + else: + records = None + if not isinstance(records, list): + raise ValueError("Input JSON must be a list or contain documents/records/docs") + return [record for record in records if isinstance(record, Mapping)] + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Build a versioned HURC1 RAG source index") + parser.add_argument("--input", help="JSON file containing source documents") + parser.add_argument("--raw-docs-dir", default="data/raw/docs") + parser.add_argument("--index-dir", default="data/vector_db") + parser.add_argument("--chunk-size", type=int, default=DEFAULT_CHUNK_SIZE) + parser.add_argument("--chunk-overlap", type=int, default=DEFAULT_CHUNK_OVERLAP) + args = parser.parse_args(argv) + + documents = _load_input(args.input) + upgrader = DocumentRAGUpgrade( + raw_docs_dir=args.raw_docs_dir, + index_dir=args.index_dir, + chunk_size=args.chunk_size, + chunk_overlap=args.chunk_overlap, + ) + upgrader.create_new_vector_index(documents) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/infra/ai-server/rag/test_document_ingestion_v2.py b/infra/ai-server/rag/test_document_ingestion_v2.py new file mode 100644 index 00000000..d2502b4a --- /dev/null +++ b/infra/ai-server/rag/test_document_ingestion_v2.py @@ -0,0 +1,67 @@ +import importlib.util +import json +import sys +import tempfile +import unittest +from pathlib import Path + +MODULE_PATH = Path(__file__).with_name("document_ingestion_v2.py") +SPEC = importlib.util.spec_from_file_location("hurc_document_ingestion", MODULE_PATH) +assert SPEC and SPEC.loader +ingestion = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = ingestion +SPEC.loader.exec_module(ingestion) + + +class DocumentIngestionTests(unittest.TestCase): + def test_creates_chunks_metadata_and_latest_pointer(self): + with tempfile.TemporaryDirectory() as temp_dir: + index_dir = Path(temp_dir) / "indexes" + raw_dir = Path(temp_dir) / "raw" + builder = ingestion.DocumentRAGUpgrade( + str(raw_dir), + str(index_dir), + chunk_size=300, + chunk_overlap=50, + ) + source = { + "title": "Quy trình kiểm tra PSD tại ga Bến Thành", + "code": "OM-PSD-001", + "version": "2.0", + "source_file": "OM-PSD-001.pdf", + "content": "Kiểm tra cách điện cửa chắn ke ga. " * 40, + } + output_path = Path(builder.create_new_vector_index([source])) + payload = json.loads(output_path.read_text(encoding="utf-8")) + latest = json.loads((index_dir / "latest.json").read_text(encoding="utf-8")) + + self.assertEqual(payload["schema_version"], "3.0") + self.assertGreater(payload["statistics"]["unique_chunks"], 1) + self.assertEqual(payload["chunks"][0]["metadata"]["related_system"], "PSD") + self.assertEqual(payload["chunks"][0]["metadata"]["source_file"], "OM-PSD-001.pdf") + self.assertEqual(latest["filename"], output_path.name) + + def test_deduplicates_identical_chunks_and_rejects_empty_documents(self): + with tempfile.TemporaryDirectory() as temp_dir: + builder = ingestion.DocumentRAGUpgrade( + index_dir=str(Path(temp_dir) / "indexes"), + raw_docs_dir=str(Path(temp_dir) / "raw"), + chunk_size=300, + chunk_overlap=40, + ) + docs = [ + {"title": "AFC A", "code": "AFC-1", "content": "Nội dung kiểm tra AFC."}, + {"title": "AFC B", "code": "AFC-2", "content": "Nội dung kiểm tra AFC."}, + {"title": "Empty", "content": ""}, + ] + output_path = Path(builder.create_new_vector_index(docs)) + payload = json.loads(output_path.read_text(encoding="utf-8")) + + self.assertEqual(payload["statistics"]["accepted_documents"], 2) + self.assertEqual(payload["statistics"]["rejected_documents"], 1) + self.assertEqual(payload["statistics"]["unique_chunks"], 1) + self.assertEqual(payload["statistics"]["duplicate_chunks_removed"], 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/src/lib/ai/rag_engine.py b/src/lib/ai/rag_engine.py index d559ce22..70c4660b 100644 --- a/src/lib/ai/rag_engine.py +++ b/src/lib/ai/rag_engine.py @@ -1,58 +1,446 @@ -import sys +"""Offline retrieval engine used by the HURC1 Copilot fallback. + +The module intentionally depends only on Python's standard library so it can run +inside the existing Next.js server action without installing an ML stack. It +implements a small hybrid lexical retriever (BM25 + phrase/category bonuses), +loads versioned ingestion indexes when available, and preserves the historical +JSON response contract consumed by ``askCopilot``. +""" + +from __future__ import annotations + +import glob +import hashlib import json import math +import os import re +import sys +import unicodedata +from collections import Counter +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Mapping, Sequence -# Giả lập Kho tài liệu Kỹ thuật Bảo trì (Knowledge Base) -MANUAL_PAGES = [ - {"id": "doc_1", "text": "Mã lỗi E-404 trên động cơ kéo (Traction Motor) biểu thị tình trạng quá nhiệt stator. Cách khắc phục: 1. Dừng tàu khẩn cấp. 2. Kiểm tra quạt làm mát. 3. Vệ sinh bộ lọc bụi.", "category": "Motor"}, - {"id": "doc_2", "text": "Bảo dưỡng định kỳ hộp số (Gearbox) mức 2: Yêu cầu thay dầu bôi trơn sau mỗi 50,000 km. Sử dụng dầu Castrol Syntrans. Kiểm tra độ rơ của bánh răng, độ rung cho phép dưới 4.5 mm/s.", "category": "Gearbox"}, - {"id": "doc_3", "text": "Hệ thống phanh hãm (Brake System) báo đèn đỏ: Cảm biến áp suất khí nén có thể bị rò rỉ. Cần xả e, thay thế van một chiều và reset bộ điều khiển trung tâm (BCU).", "category": "Brake"}, - {"id": "doc_4", "text": "Thay thế phụ tùng vòng bi (Bearing): Yêu cầu sử dụng kích thủy lực, tháo trục bánh xe. Không dùng búa đập trực tiếp vào ca ngoài của vòng bi.", "category": "Wheel"} +ENGINE_NAME = "Hybrid BM25 Offline RAG v3" +MAX_QUERY_CHARS = 4_000 +MAX_ANSWER_CHARS = 3_500 +DEFAULT_TOP_K = 3 +DEFAULT_MIN_CONFIDENCE = 0.32 + +# Safe last-resort content. Production deployments should point +# RAG_KNOWLEDGE_PATH to an index generated by document_ingestion_v2.py. +FALLBACK_MANUAL_PAGES: list[dict[str, Any]] = [ + { + "id": "doc_1", + "text": ( + "Mã lỗi E-404 trên động cơ kéo (Traction Motor) biểu thị tình trạng " + "quá nhiệt stator. Cách khắc phục: dừng tàu theo quy trình an toàn, " + "kiểm tra quạt làm mát và vệ sinh bộ lọc bụi." + ), + "category": "Motor", + "document_code": "DEMO-MOTOR-001", + }, + { + "id": "doc_2", + "text": ( + "Bảo dưỡng định kỳ hộp số (Gearbox) mức 2: thay dầu bôi trơn sau mỗi " + "50.000 km, kiểm tra độ rơ bánh răng và độ rung theo giới hạn của tài liệu bảo trì." + ), + "category": "Gearbox", + "document_code": "DEMO-GEARBOX-001", + }, + { + "id": "doc_3", + "text": ( + "Hệ thống phanh báo đèn đỏ có thể liên quan đến áp suất khí nén hoặc " + "rò rỉ. Cần cô lập thiết bị theo quy trình, kiểm tra cảm biến, van một chiều và BCU." + ), + "category": "Brake", + "document_code": "DEMO-BRAKE-001", + }, + { + "id": "doc_4", + "text": ( + "Khi thay vòng bi, sử dụng dụng cụ chuyên dùng và kích thủy lực. " + "Không dùng búa đập trực tiếp vào vòng ngoài của vòng bi." + ), + "category": "Wheel", + "document_code": "DEMO-WHEEL-001", + }, ] -def tokenize(text): - return re.findall(r'\w+', text.lower()) - -def tf_idf_search(query, docs): - query_tokens = set(tokenize(query)) - - best_match = None - highest_score = 0 - - for doc in docs: - doc_tokens = tokenize(doc["text"]) - score = 0 - for q in query_tokens: - if q in doc_tokens: - score += 1 # Mô phỏng BM25 / TF-IDF đơn giản - - if score > highest_score: - highest_score = score - best_match = doc - - return best_match, highest_score +BASE_STOPWORDS = { + "a", "an", "and", "are", "as", "at", "be", "by", "for", "from", "how", "in", + "is", "it", "of", "on", "or", "that", "the", "this", "to", "what", "when", "where", + "which", "with", "và", "là", "của", "cho", "trong", "trên", "dưới", "với", "một", + "các", "những", "này", "đó", "được", "bị", "có", "không", "như", "khi", "tại", + "về", "theo", "từ", "đến", "hãy", "giúp", "tôi", "cần", "nào", "gì", "sao", +} -if __name__ == "__main__": - if len(sys.argv) > 1: +WORD_RE = re.compile(r"[^\W_]+(?:[-./][^\W_]+)*", re.UNICODE) + + +def normalize_text(text: str, *, fold_accents: bool = False) -> str: + normalized = unicodedata.normalize("NFKC", str(text or "")).lower().strip() + normalized = re.sub(r"\s+", " ", normalized) + if fold_accents: + decomposed = unicodedata.normalize("NFD", normalized.replace("đ", "d")) + normalized = "".join(ch for ch in decomposed if unicodedata.category(ch) != "Mn") + return normalized + + +FOLDED_STOPWORDS = {normalize_text(word, fold_accents=True) for word in BASE_STOPWORDS} + + +def tokenize(text: str) -> list[str]: + """Return Unicode-aware, accent-insensitive terms without double counting.""" + folded = normalize_text(text, fold_accents=True) + terms: list[str] = [] + for token in WORD_RE.findall(folded): + token = token.strip("-./") + if len(token) < 2 or token in FOLDED_STOPWORDS: + continue + terms.append(token) + return terms + + +def _stable_id(content: str) -> str: + return hashlib.sha256(content.encode("utf-8")).hexdigest()[:20] + + +@dataclass(frozen=True) +class Document: + id: str + text: str + metadata: dict[str, Any] + + @property + def category(self) -> str: + return str( + self.metadata.get("related_system") + or self.metadata.get("category") + or self.metadata.get("system") + or "General" + ) + + @property + def document_code(self) -> str: + return str(self.metadata.get("document_code") or self.metadata.get("code") or self.id) + + +def _coerce_document(raw: Mapping[str, Any], position: int) -> Document | None: + text = str(raw.get("content") or raw.get("text") or "").strip() + if not text: + return None + + metadata = dict(raw.get("metadata") or {}) + for key in ( + "document_code", "code", "category", "related_system", "system", "title", + "version", "issue_date", "contractor", "related_asset", "source_file", "page", + "section", "chunk_index", "content_hash", + ): + if key in raw and key not in metadata: + metadata[key] = raw[key] + + doc_id = str(raw.get("id") or metadata.get("chunk_id") or _stable_id(f"{position}:{text}")) + return Document(id=doc_id, text=text, metadata=metadata) + + +def _extract_records(payload: Any) -> list[Mapping[str, Any]]: + if isinstance(payload, list): + return [item for item in payload if isinstance(item, Mapping)] + if not isinstance(payload, Mapping): + return [] + + for key in ("chunks", "documents", "records"): + value = payload.get(key) + if isinstance(value, list): + return [item for item in value if isinstance(item, Mapping)] + return [] + + +def _candidate_index_paths() -> list[Path]: + candidates: list[Path] = [] + env_path = os.getenv("RAG_KNOWLEDGE_PATH", "").strip() + if env_path: + candidates.append(Path(env_path)) + + project_root = Path(__file__).resolve().parents[3] + index_dir = project_root / "data" / "vector_db" + candidates.extend( + [ + index_dir / "latest.json", + index_dir / "current_index.json", + ] + ) + + versioned = sorted( + ( + Path(path) + for pattern in ("metro_rag_index_v3_*.json", "metro_rag_index_v2_*.json") + for path in glob.glob(str(index_dir / pattern)) + ), + key=lambda path: path.stat().st_mtime if path.exists() else 0, + reverse=True, + ) + candidates.extend(versioned) + return candidates + + +def _resolve_latest_pointer(path: Path, payload: Any) -> tuple[Path, Any]: + if path.name != "latest.json" or not isinstance(payload, Mapping): + return path, payload + target = payload.get("path") or payload.get("index_path") or payload.get("filename") + if not target: + return path, payload + target_path = Path(str(target)) + if not target_path.is_absolute(): + target_path = path.parent / target_path + with target_path.open("r", encoding="utf-8") as handle: + return target_path, json.load(handle) + + +def load_documents(path: str | os.PathLike[str] | None = None) -> tuple[list[Document], str]: + paths = [Path(path)] if path else _candidate_index_paths() + for candidate in paths: + if not candidate.exists() or not candidate.is_file(): + continue try: - query_json = sys.argv[1] - data = json.loads(query_json) - query = data.get("query", "") - - best_match, score = tf_idf_search(query, MANUAL_PAGES) - - if score > 0: - answer = f"Theo Cẩm nang Bảo trì ({best_match['category']}):\n{best_match['text']}" - else: - answer = "Xin lỗi, tôi không tìm thấy thông tin phù hợp trong kho tài liệu kỹ thuật." - - print(json.dumps({ - "answer": answer, - "confidence": score, - "engine": "TF-IDF Offline RAG" - })) - except Exception as e: - print(json.dumps({"error": str(e)})) + with candidate.open("r", encoding="utf-8") as handle: + payload = json.load(handle) + resolved_path, payload = _resolve_latest_pointer(candidate, payload) + records = _extract_records(payload) + documents = [doc for i, raw in enumerate(records) if (doc := _coerce_document(raw, i))] + if documents: + return documents, str(resolved_path) + except (OSError, ValueError, TypeError): + continue + + fallback_docs = [ + doc for i, raw in enumerate(FALLBACK_MANUAL_PAGES) if (doc := _coerce_document(raw, i)) + ] + return fallback_docs, "embedded-demo-fallback" + + +@dataclass(frozen=True) +class SearchHit: + document: Document + raw_score: float + confidence: float + matched_terms: tuple[str, ...] + + +def _char_trigrams(text: str) -> set[str]: + compact = re.sub(r"\s+", " ", normalize_text(text, fold_accents=True)) + if len(compact) < 3: + return {compact} if compact else set() + return {compact[i:i + 3] for i in range(len(compact) - 2)} + + +def hybrid_search( + query: str, + docs: Sequence[Document], + *, + top_k: int = DEFAULT_TOP_K, + category: str | None = None, +) -> list[SearchHit]: + query_terms = tokenize(query) + if not query_terms or not docs: + return [] + + filtered_docs = [ + doc + for doc in docs + if not category + or normalize_text(doc.category, fold_accents=True) + == normalize_text(category, fold_accents=True) + ] + if not filtered_docs: + return [] + + searchable_texts = [ + " ".join( + part + for part in ( + str(doc.metadata.get("title") or ""), + doc.document_code, + doc.category, + doc.text, + ) + if part + ) + for doc in filtered_docs + ] + tokenized_docs = [tokenize(text) for text in searchable_texts] + doc_term_counts = [Counter(tokens) for tokens in tokenized_docs] + avg_doc_len = sum(max(1, len(tokens)) for tokens in tokenized_docs) / len(tokenized_docs) + + document_frequency: Counter[str] = Counter() + for counts in doc_term_counts: + document_frequency.update(counts.keys()) + + normalized_query = normalize_text(query, fold_accents=True) + query_trigrams = _char_trigrams(query) + query_term_set = set(query_terms) + total_docs = len(filtered_docs) + hits: list[SearchHit] = [] + + for doc, searchable_text, counts, tokens in zip( + filtered_docs, searchable_texts, doc_term_counts, tokenized_docs + ): + doc_len = max(1, len(tokens)) + score = 0.0 + matched: set[str] = set() + + for term in query_term_set: + tf = counts.get(term, 0) + if not tf: + continue + matched.add(term) + df = document_frequency.get(term, 0) + idf = math.log(1 + (total_docs - df + 0.5) / (df + 0.5)) + denominator = tf + 1.5 * (1 - 0.75 + 0.75 * doc_len / avg_doc_len) + score += idf * (tf * 2.5) / denominator + + folded_doc = normalize_text(searchable_text, fold_accents=True) + folded_code = normalize_text(doc.document_code, fold_accents=True) + folded_category = normalize_text(doc.category, fold_accents=True) + + if len(normalized_query) >= 5 and normalized_query in folded_doc: + score += 1.25 + if folded_code and folded_code in normalized_query: + score += 1.0 + if folded_category and folded_category in normalized_query: + score += 0.65 + + doc_trigrams = _char_trigrams(searchable_text[:2_000]) + if query_trigrams and doc_trigrams: + trigram_similarity = len(query_trigrams & doc_trigrams) / max(1, len(query_trigrams)) + score += 0.8 * trigram_similarity + + coverage = len(matched) / max(1, len(query_term_set)) + score += 0.9 * coverage + + # Confidence is bounded and interpretable; it is not a probability. + confidence = (1 - math.exp(-score / 3.2)) * (0.65 + 0.35 * coverage) + confidence *= coverage**0.7 + confidence = max(0.0, min(1.0, confidence)) + if score > 0: + hits.append( + SearchHit( + document=doc, + raw_score=round(score, 6), + confidence=round(confidence, 4), + matched_terms=tuple(sorted(matched)), + ) + ) + + hits.sort(key=lambda hit: (hit.confidence, hit.raw_score), reverse=True) + return hits[: max(1, min(int(top_k), 10))] + + +def _clean_excerpt(text: str, limit: int = 900) -> str: + excerpt = re.sub(r"\s+", " ", text).strip() + if len(excerpt) <= limit: + return excerpt + return excerpt[: limit - 1].rstrip() + "…" + + +def build_answer(hits: Sequence[SearchHit]) -> str: + if not hits: + return "Xin lỗi, tôi không tìm thấy thông tin phù hợp trong kho tài liệu kỹ thuật." + + lines = ["Theo các tài liệu kỹ thuật phù hợp nhất:"] + for index, hit in enumerate(hits, start=1): + label = hit.document.document_code + category = hit.document.category + lines.append(f"{index}. [{label} | {category}] {_clean_excerpt(hit.document.text)}") + return "\n".join(lines)[:MAX_ANSWER_CHARS] + + +def answer_query( + query: str, + *, + top_k: int = DEFAULT_TOP_K, + category: str | None = None, + min_confidence: float = DEFAULT_MIN_CONFIDENCE, + knowledge_path: str | None = None, +) -> dict[str, Any]: + clean_query = str(query or "").strip() + if not clean_query: + raise ValueError("Query is required") + if len(clean_query) > MAX_QUERY_CHARS: + raise ValueError(f"Query exceeds {MAX_QUERY_CHARS} characters") + + documents, knowledge_source = load_documents(knowledge_path) + hits = hybrid_search(clean_query, documents, top_k=top_k, category=category) + if hits: + relative_floor = max(min_confidence, hits[0].confidence * 0.55) + accepted_hits = [hit for hit in hits if hit.confidence >= relative_floor] else: - print(json.dumps({"error": "No query provided"})) + accepted_hits = [] + + if not accepted_hits: + return { + "answer": "Xin lỗi, tôi không tìm thấy thông tin đủ tin cậy trong kho tài liệu kỹ thuật.", + "confidence": 0.0, + "engine": ENGINE_NAME, + "knowledge_source": knowledge_source, + "sources": [], + } + + top_confidence = accepted_hits[0].confidence + return { + "answer": build_answer(accepted_hits), + "confidence": top_confidence, + "engine": ENGINE_NAME, + "knowledge_source": knowledge_source, + "sources": [ + { + "id": hit.document.id, + "document_code": hit.document.document_code, + "category": hit.document.category, + "score": hit.raw_score, + "confidence": hit.confidence, + "matched_terms": list(hit.matched_terms), + "metadata": hit.document.metadata, + } + for hit in accepted_hits + ], + } + + +def _read_cli_payload(argv: Sequence[str]) -> Mapping[str, Any]: + if len(argv) <= 1: + raise ValueError("No query provided") + payload = json.loads(argv[1]) + if not isinstance(payload, Mapping): + raise ValueError("Input must be a JSON object") + return payload + + +def main(argv: Sequence[str] | None = None) -> int: + argv = argv or sys.argv + try: + payload = _read_cli_payload(argv) + result = answer_query( + str(payload.get("query", "")), + top_k=int(payload.get("top_k", DEFAULT_TOP_K)), + category=str(payload["category"]) if payload.get("category") else None, + min_confidence=float(payload.get("min_confidence", DEFAULT_MIN_CONFIDENCE)), + knowledge_path=str(payload["knowledge_path"]) if payload.get("knowledge_path") else None, + ) + print(json.dumps(result, ensure_ascii=False)) + return 0 + except (ValueError, TypeError, json.JSONDecodeError) as exc: + print(json.dumps({"error": str(exc), "engine": ENGINE_NAME}, ensure_ascii=False)) + return 2 + except Exception: + # Keep server output deterministic and avoid exposing filesystem or stack details. + print(json.dumps({"error": "Offline RAG engine failed", "engine": ENGINE_NAME}, ensure_ascii=False)) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/lib/ai/test_rag_engine.py b/src/lib/ai/test_rag_engine.py new file mode 100644 index 00000000..563df3c1 --- /dev/null +++ b/src/lib/ai/test_rag_engine.py @@ -0,0 +1,63 @@ +import importlib.util +import json +import sys +import tempfile +import unittest +from pathlib import Path + +MODULE_PATH = Path(__file__).with_name("rag_engine.py") +SPEC = importlib.util.spec_from_file_location("hurc_rag_engine", MODULE_PATH) +assert SPEC and SPEC.loader +rag_engine = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = rag_engine +SPEC.loader.exec_module(rag_engine) + + +class RagEngineTests(unittest.TestCase): + def test_vietnamese_query_is_accent_insensitive(self): + result = rag_engine.answer_query("qua nhiet stator dong co keo") + self.assertGreater(result["confidence"], 0.5) + self.assertEqual(result["sources"][0]["document_code"], "DEMO-MOTOR-001") + + def test_irrelevant_query_is_refused(self): + result = rag_engine.answer_query("noi dung khong lien quan xyz") + self.assertEqual(result["confidence"], 0.0) + self.assertEqual(result["sources"], []) + + def test_loads_versioned_chunk_index_and_returns_provenance(self): + with tempfile.TemporaryDirectory() as temp_dir: + index_path = Path(temp_dir) / "index.json" + index_path.write_text( + json.dumps( + { + "schema_version": "3.0", + "chunks": [ + { + "id": "chunk-1", + "content": "Kiểm tra vành răng định kỳ 3 tháng một lần.", + "metadata": { + "title": "Hướng dẫn bảo trì cổng soát vé AFC", + "document_code": "OM-AFC-001", + "related_system": "AFC", + "source_file": "manual.pdf", + }, + } + ], + }, + ensure_ascii=False, + ), + encoding="utf-8", + ) + + result = rag_engine.answer_query( + "bảo trì cổng soát vé AFC", + knowledge_path=str(index_path), + ) + + self.assertGreater(result["confidence"], 0.5) + self.assertEqual(result["sources"][0]["document_code"], "OM-AFC-001") + self.assertEqual(result["sources"][0]["metadata"]["source_file"], "manual.pdf") + + +if __name__ == "__main__": + unittest.main()