added

.dockerignore

@@ -0,0 +1,12 @@
+.git
+.venv
+puku
+__pycache__
+*.pyc
+.pytest_cache
+.env
+data/qdrant
+data/copilot.db
+data/bm25_index.json
+data/uploads
+*.log

.env.example

@@ -0,0 +1,30 @@
+APP_NAME=Insurance Claims Copilot
+APP_ENV=development
+APP_HOST=127.0.0.1
+APP_PORT=8000
+
+GROQ_API_KEY=
+GROQ_MODEL=llama-3.3-70b-versatile
+
+QDRANT_URL=local:data/qdrant
+QDRANT_API_KEY=
+QDRANT_COLLECTION=insurance_claims
+QDRANT_CACHE_COLLECTION=semantic_answer_cache
+
+EMBEDDING_MODEL=sentence-transformers/all-MiniLM-L6-v2
+EMBEDDING_DIM=384
+
+SQLITE_PATH=data/copilot.db
+DOCUMENT_DIR=data
+UPLOAD_DIR=data/uploads
+BM25_INDEX_PATH=data/bm25_index.json
+AUTO_INGEST_PDFS_ON_STARTUP=true
+
+SEMANTIC_CACHE_THRESHOLD=0.88
+SELF_RAG_MAX_LOOPS=2
+RETRIEVAL_TOP_K=8
+RERANK_TOP_K=5
+LOW_LATENCY_MODE=true
+ENABLE_QUERY_REWRITE=true
+MAX_SOURCES_TO_LLM=5
+MAX_EVIDENCE_CHARS_PER_SOURCE=900

.gitignore

@@ -0,0 +1,12 @@
+.venv/
+puku/
+__pycache__/
+*.pyc
+.env
+.pytest_cache/
+
+data/qdrant/
+data/copilot.db
+data/bm25_index.json
+data/uploads/
+data/eval/ragas_results.json

Dockerfile

@@ -0,0 +1,33 @@
+FROM python:3.11-bookworm
+
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV PYTHONUNBUFFERED=1
+ENV PIP_NO_CACHE_DIR=1
+ENV PIP_DEFAULT_TIMEOUT=120
+ENV HF_HOME=/app/.cache/huggingface
+
+WORKDIR /app
+
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends curl \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY requirements-docker.txt .
+RUN python -m pip install --upgrade pip \
+    && python -m pip install --index-url https://download.pytorch.org/whl/cpu torch \
+    && python -m pip install -r requirements-docker.txt
+
+COPY app ./app
+COPY frontend ./frontend
+COPY scripts ./scripts
+COPY data/sample_insurance_claim_guide.pdf ./data/sample_insurance_claim_guide.pdf
+COPY .env.example ./.env.example
+
+RUN mkdir -p /app/data/uploads /app/.cache/huggingface
+
+EXPOSE 8000
+
+HEALTHCHECK --interval=30s --timeout=10s --start-period=90s --retries=3 \
+    CMD curl -fsS http://127.0.0.1:8000/api/health || exit 1
+
+CMD ["python", "-m", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Md Shoaib Shahriar Ibrahim
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

app/__init__.py

@@ -0,0 +1 @@
+"""Insurance claims copilot backend."""

app/api/__init__.py

@@ -0,0 +1 @@
+"""API route package."""

app/api/routes_health.py

@@ -0,0 +1,8 @@
+from fastapi import APIRouter
+
+router = APIRouter(tags=["health"])
+
+
+@router.get("/health")
+def health() -> dict[str, str]:
+    return {"status": "ok"}

app/api/routes_ingest.py

@@ -0,0 +1,46 @@
+from typing import Any
+
+from fastapi import APIRouter, File, Form, HTTPException, UploadFile
+from pydantic import BaseModel, Field
+
+from app.rag.ingestion import DocumentIngestionService
+
+router = APIRouter(tags=["documents"])
+
+
+class IngestTextRequest(BaseModel):
+    text: str = Field(min_length=1)
+    source_name: str = "manual_input"
+    metadata: dict[str, Any] | None = None
+
+
+@router.post("/documents/ingest")
+def ingest_text(payload: IngestTextRequest) -> dict[str, Any]:
+    return DocumentIngestionService().ingest_text(
+        text=payload.text,
+        source_name=payload.source_name,
+        metadata=payload.metadata,
+    )
+
+
+@router.post("/documents/upload")
+async def upload_document(
+    file: UploadFile = File(...),
+    policy_type: str = Form(default=""),
+    jurisdiction: str = Form(default=""),
+) -> dict[str, Any]:
+    raw = await file.read()
+    try:
+        text = raw.decode("utf-8")
+    except UnicodeDecodeError as exc:
+        raise HTTPException(status_code=400, detail="Only UTF-8 text files are supported in this scaffold.") from exc
+
+    metadata = {
+        "policy_type": policy_type,
+        "jurisdiction": jurisdiction,
+    }
+    return DocumentIngestionService().ingest_text(
+        text=text,
+        source_name=file.filename or "uploaded_document.txt",
+        metadata=metadata,
+    )

app/api/routes_query.py

@@ -0,0 +1,72 @@
+import json
+import time
+from functools import lru_cache
+from typing import Any
+
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel, Field
+
+from app.db.sqlite import db
+from app.rag.graph import ClaimsRAGGraph
+
+router = APIRouter(tags=["query"])
+
+
+@lru_cache
+def get_rag_graph() -> ClaimsRAGGraph:
+    return ClaimsRAGGraph()
+
+
+class QueryRequest(BaseModel):
+    query: str = Field(min_length=1)
+    user_id: str = "default_user"
+    metadata_filter: dict[str, Any] | None = None
+
+
+@router.post("/query")
+def query_copilot(payload: QueryRequest) -> dict[str, Any]:
+    started = time.perf_counter()
+    try:
+        result = get_rag_graph().run(payload.query, payload.metadata_filter, user_id=payload.user_id)
+        latency_ms = round((time.perf_counter() - started) * 1000, 2)
+        raw_sources = result.get("reranked_sources") or result.get("sources", [])
+        sources = [
+            {
+                "source_name": source.get("source_name", "unknown"),
+                "text": source.get("text", ""),
+                "score": source.get("score", source.get("rrf_score", 0.0)),
+                "page": source.get("metadata", {}).get("page"),
+            }
+            for source in raw_sources
+        ]
+        return {
+            "request_id": result["request_id"],
+            "answer": result.get("answer", ""),
+            "confidence": result.get("confidence", 0.0),
+            "latency_ms": latency_ms,
+            "sources": sources,
+            "from_cache": result.get("cache_hit", False),
+            "retrieval_used": bool(result.get("should_retrieve", True)),
+            "self_rag": {
+                **result.get("self_rag", {}),
+                "iterations": result.get("iteration", 0),
+            },
+            "memory_used": result.get("memory_context", "") != "No prior memory for this user.",
+            "trace": result.get("trace", []),
+        }
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+
+
+@router.get("/traces/{request_id}")
+def get_trace(request_id: str) -> dict[str, Any]:
+    with db() as conn:
+        row = conn.execute("SELECT * FROM traces WHERE request_id = ?", (request_id,)).fetchone()
+    if not row:
+        raise HTTPException(status_code=404, detail="Trace not found")
+    return {
+        "request_id": row["request_id"],
+        "query": row["query"],
+        "trace": json.loads(row["trace_json"]),
+        "created_at": row["created_at"],
+    }

app/api/routes_review.py

@@ -0,0 +1,86 @@
+import json
+from typing import Any
+
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel, Field
+
+from app.db.sqlite import db
+from app.rag.graph import ClaimsRAGGraph
+from app.rag.text import new_id
+
+router = APIRouter(tags=["review"])
+
+
+class ApproveRequest(BaseModel):
+    request_id: str
+    original_answer: str
+    approved_answer: str
+    reviewer: str = "human_adjuster"
+
+
+class RegenerateRequest(BaseModel):
+    query: str = Field(min_length=1)
+    metadata_filter: dict[str, Any] | None = None
+
+
+@router.post("/review/approve")
+def approve(payload: ApproveRequest) -> dict[str, Any]:
+    review_id = new_id("review")
+    with db() as conn:
+        conn.execute(
+            """
+            INSERT INTO reviews(review_id, request_id, original_answer, approved_answer, reviewer, status)
+            VALUES (?, ?, ?, ?, ?, ?)
+            """,
+            (
+                review_id,
+                payload.request_id,
+                payload.original_answer,
+                payload.approved_answer,
+                payload.reviewer,
+                "approved",
+            ),
+        )
+    return {"status": "approved", "review_id": review_id}
+
+
+@router.post("/review/regenerate")
+def regenerate(payload: RegenerateRequest) -> dict[str, Any]:
+    result = ClaimsRAGGraph().run(payload.query, payload.metadata_filter)
+    return {
+        "request_id": result["request_id"],
+        "answer": result.get("answer", ""),
+        "confidence": result.get("confidence", 0.0),
+        "sources": result.get("reranked_sources") or result.get("sources", []),
+        "self_rag": result.get("self_rag", {}),
+    }
+
+
+@router.get("/claims/{claim_id}")
+def get_claim(claim_id: str) -> dict[str, Any]:
+    return {
+        "claim_id": claim_id,
+        "customer": "Sample Customer",
+        "status": "Needs Review",
+        "loss_type": "Property",
+        "summary": "No live claims system is connected yet. This placeholder is ready for plan lookup integration.",
+    }
+
+
+@router.get("/reviews")
+def list_reviews() -> dict[str, Any]:
+    with db() as conn:
+        rows = conn.execute("SELECT * FROM reviews ORDER BY created_at DESC LIMIT 25").fetchall()
+    return {
+        "reviews": [
+            {
+                "review_id": row["review_id"],
+                "request_id": row["request_id"],
+                "approved_answer": row["approved_answer"],
+                "reviewer": row["reviewer"],
+                "status": row["status"],
+                "created_at": row["created_at"],
+            }
+            for row in rows
+        ]
+    }

app/core/__init__.py

@@ -0,0 +1 @@
+"""Core application services."""

app/core/config.py

@@ -0,0 +1,53 @@
+from functools import lru_cache
+from pathlib import Path
+
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    app_name: str = "Insurance Claims Copilot"
+    app_env: str = "development"
+    app_host: str = "127.0.0.1"
+    app_port: int = 8000
+
+    groq_api_key: str = ""
+    groq_model: str = "llama-3.3-70b-versatile"
+
+    qdrant_url: str = "local:data/qdrant"
+    qdrant_api_key: str = ""
+    qdrant_collection: str = "insurance_claims"
+    qdrant_cache_collection: str = "semantic_answer_cache"
+
+    embedding_model: str = "sentence-transformers/all-MiniLM-L6-v2"
+    embedding_dim: int = 384
+
+    sqlite_path: str = "data/copilot.db"
+    document_dir: str = "data"
+    upload_dir: str = "data/uploads"
+    bm25_index_path: str = "data/bm25_index.json"
+    auto_ingest_pdfs_on_startup: bool = True
+
+    semantic_cache_threshold: float = 0.88
+    self_rag_max_loops: int = 2
+    retrieval_top_k: int = 8
+    rerank_top_k: int = 5
+    low_latency_mode: bool = True
+    enable_query_rewrite: bool = True
+    max_sources_to_llm: int = 5
+    max_evidence_chars_per_source: int = 900
+
+    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
+
+    def ensure_directories(self) -> None:
+        Path(self.sqlite_path).parent.mkdir(parents=True, exist_ok=True)
+        Path(self.document_dir).mkdir(parents=True, exist_ok=True)
+        Path(self.upload_dir).mkdir(parents=True, exist_ok=True)
+        Path(self.bm25_index_path).parent.mkdir(parents=True, exist_ok=True)
+
+
+@lru_cache
+def get_settings() -> Settings:
+    return Settings()
+
+
+settings = get_settings()

app/core/logging.py

@@ -0,0 +1,8 @@
+import logging
+
+
+def configure_logging() -> None:
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s %(message)s",
+    )

app/db/__init__.py

@@ -0,0 +1 @@
+"""Database package."""

app/db/sqlite.py

@@ -0,0 +1,162 @@
+import sqlite3
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Iterator
+
+from app.core.config import settings
+
+
+SCHEMA = """
+PRAGMA journal_mode=WAL;
+
+CREATE TABLE IF NOT EXISTS documents (
+    doc_id TEXT PRIMARY KEY,
+    source_name TEXT NOT NULL,
+    file_hash TEXT NOT NULL,
+    normalized_hash TEXT NOT NULL,
+    simhash TEXT NOT NULL,
+    status TEXT NOT NULL,
+    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE TABLE IF NOT EXISTS chunks (
+    chunk_id TEXT PRIMARY KEY,
+    doc_id TEXT NOT NULL,
+    chunk_index INTEGER NOT NULL,
+    text TEXT NOT NULL,
+    text_hash TEXT NOT NULL,
+    metadata_json TEXT NOT NULL,
+    embedded_at TEXT,
+    FOREIGN KEY(doc_id) REFERENCES documents(doc_id)
+);
+
+CREATE TABLE IF NOT EXISTS answer_cache (
+    cache_id TEXT PRIMARY KEY,
+    query TEXT NOT NULL,
+    normalized_query TEXT NOT NULL,
+    answer TEXT NOT NULL,
+    confidence REAL NOT NULL,
+    sources_json TEXT NOT NULL,
+    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE TABLE IF NOT EXISTS traces (
+    request_id TEXT PRIMARY KEY,
+    query TEXT NOT NULL,
+    trace_json TEXT NOT NULL,
+    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE TABLE IF NOT EXISTS reviews (
+    review_id TEXT PRIMARY KEY,
+    request_id TEXT NOT NULL,
+    original_answer TEXT NOT NULL,
+    approved_answer TEXT NOT NULL,
+    reviewer TEXT NOT NULL,
+    status TEXT NOT NULL,
+    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE TABLE IF NOT EXISTS memories (
+    memory_id TEXT PRIMARY KEY,
+    user_id TEXT NOT NULL,
+    kind TEXT NOT NULL,
+    content TEXT NOT NULL,
+    metadata_json TEXT NOT NULL,
+    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE TABLE IF NOT EXISTS customers (
+    customer_id TEXT PRIMARY KEY,
+    name TEXT NOT NULL,
+    preferred_contact TEXT NOT NULL,
+    risk_notes TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS policies (
+    policy_id TEXT PRIMARY KEY,
+    customer_id TEXT NOT NULL,
+    policy_type TEXT NOT NULL,
+    active INTEGER NOT NULL,
+    coverages_json TEXT NOT NULL,
+    exclusions_json TEXT NOT NULL,
+    deductible REAL NOT NULL,
+    policy_limit REAL NOT NULL,
+    endorsements_json TEXT NOT NULL,
+    FOREIGN KEY(customer_id) REFERENCES customers(customer_id)
+);
+
+CREATE TABLE IF NOT EXISTS claims (
+    claim_id TEXT PRIMARY KEY,
+    customer_id TEXT NOT NULL,
+    policy_id TEXT NOT NULL,
+    claim_type TEXT NOT NULL,
+    status TEXT NOT NULL,
+    date_of_loss TEXT NOT NULL,
+    missing_documents_json TEXT NOT NULL,
+    notes TEXT NOT NULL,
+    FOREIGN KEY(customer_id) REFERENCES customers(customer_id),
+    FOREIGN KEY(policy_id) REFERENCES policies(policy_id)
+);
+
+CREATE TABLE IF NOT EXISTS ticket_queues (
+    queue_name TEXT PRIMARY KEY,
+    open_tickets INTEGER NOT NULL,
+    estimated_review_time TEXT NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_documents_hash ON documents(file_hash, normalized_hash);
+CREATE INDEX IF NOT EXISTS idx_chunks_hash ON chunks(text_hash);
+CREATE INDEX IF NOT EXISTS idx_memories_user_kind ON memories(user_id, kind, created_at);
+"""
+
+
+SEED_SQL = """
+INSERT OR IGNORE INTO customers(customer_id, name, preferred_contact, risk_notes)
+VALUES
+('CUS-1001', 'Maya Rahman', 'email', 'Prior water claim required mitigation documentation.'),
+('CUS-1002', 'Omar Khan', 'phone', 'No special risk notes.'),
+('CUS-1003', 'Nadia Islam', 'email', 'Previously submitted theft claim with missing receipts.');
+
+INSERT OR IGNORE INTO policies(policy_id, customer_id, policy_type, active, coverages_json, exclusions_json, deductible, policy_limit, endorsements_json)
+VALUES
+('POL-3001', 'CUS-1001', 'property', 1, '["sudden accidental water discharge", "fire and smoke", "wind and hail", "theft of personal property"]', '["gradual leakage", "mold from long-term seepage", "flood without endorsement", "wear and tear"]', 1000, 25000, '["limited sewer backup"]'),
+('POL-3002', 'CUS-1002', 'auto', 1, '["collision", "comprehensive theft", "liability"]', '["intentional damage", "unlisted commercial use"]', 500, 40000, '[]'),
+('POL-3003', 'CUS-1003', 'property', 1, '["fire and smoke", "theft of personal property", "wind and hail"]', '["flood", "gradual leakage", "high-value unscheduled jewelry above sublimit"]', 1500, 50000, '["scheduled jewelry required above sublimit"]');
+
+INSERT OR IGNORE INTO claims(claim_id, customer_id, policy_id, claim_type, status, date_of_loss, missing_documents_json, notes)
+VALUES
+('CLM-1007', 'CUS-1001', 'POL-3001', 'water_damage', 'documents_pending', '2026-05-10', '["mitigation invoice", "repair estimate"]', 'Kitchen burst pipe; photos and plumber report received.'),
+('CLM-2011', 'CUS-1003', 'POL-3003', 'theft', 'human_review', '2026-04-28', '["receipts", "serial numbers"]', 'Laptop and camera stolen from vehicle; police report received.'),
+('CLM-3020', 'CUS-1001', 'POL-3001', 'storm', 'new', '2026-05-12', '["contractor estimate", "weather event confirmation"]', 'Roof hail damage reported.');
+
+INSERT OR IGNORE INTO ticket_queues(queue_name, open_tickets, estimated_review_time)
+VALUES
+('property_claims', 14, '1 business day'),
+('auto_claims', 8, 'same day'),
+('special_investigation', 6, '2 business days'),
+('liability_claims', 11, '1-2 business days');
+"""
+
+
+def connect() -> sqlite3.Connection:
+    Path(settings.sqlite_path).parent.mkdir(parents=True, exist_ok=True)
+    conn = sqlite3.connect(settings.sqlite_path)
+    conn.row_factory = sqlite3.Row
+    return conn
+
+
+@contextmanager
+def db() -> Iterator[sqlite3.Connection]:
+    conn = connect()
+    try:
+        yield conn
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def init_db() -> None:
+    with db() as conn:
+        conn.executescript(SCHEMA)
+        conn.executescript(SEED_SQL)

app/guardrails/__init__.py

@@ -0,0 +1 @@
+"""Guardrail package."""

app/guardrails/pii.py

@@ -0,0 +1,82 @@
+import re
+from dataclasses import dataclass
+
+
+@dataclass
+class GuardrailResult:
+    text: str
+    blocked: bool
+    findings: list[str]
+
+
+def build_langchain_pii_middlewares() -> list[object]:
+    """Create LangChain PII middleware objects for agent-compatible deployments."""
+    try:
+        from langchain.agents.middleware import PIIMiddleware
+
+        return [
+            PIIMiddleware("email", strategy="redact", apply_to_input=True, apply_to_output=True),
+            PIIMiddleware("credit_card", strategy="mask", apply_to_input=True, apply_to_output=True),
+            PIIMiddleware("ip", strategy="redact", apply_to_input=True, apply_to_output=True),
+            PIIMiddleware(
+                "policy_number",
+                detector=r"\b(?:POL[-\s]?(?=[A-Z0-9-]*\d)[A-Z0-9]{5,}|Policy[-\s]*(?:ID|No\.?|Number)[-:\s]*[A-Z0-9]*\d[A-Z0-9-]{4,})\b",
+                strategy="hash",
+                apply_to_input=True,
+                apply_to_output=True,
+            ),
+            PIIMiddleware(
+                "claim_id",
+                detector=r"\b(?:CLM[-\s]?(?=[A-Z0-9-]*\d)[A-Z0-9]{5,}|Claim[-\s]?(?:ID|No\.?|Number)?[-\s]*[A-Z0-9]*\d[A-Z0-9-]{4,})\b",
+                strategy="hash",
+                apply_to_input=True,
+                apply_to_output=True,
+            ),
+        ]
+    except Exception:
+        return []
+
+
+class PIIGuardrails:
+    def __init__(self) -> None:
+        self.langchain_middlewares = build_langchain_pii_middlewares()
+        self.patterns: list[tuple[str, re.Pattern[str], str]] = [
+            ("email", re.compile(r"\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b", re.I), "[REDACTED_EMAIL]"),
+            ("credit_card", re.compile(r"\b(?:\d[ -]*?){13,19}\b"), "[MASKED_CARD]"),
+            ("phone", re.compile(r"\b(?:\+?\d{1,3}[-.\s]?)?(?:\(?\d{3}\)?[-.\s]?)\d{3}[-.\s]?\d{4}\b"), "[REDACTED_PHONE]"),
+            (
+                "policy_number",
+                re.compile(r"\b(?:POL[-\s]?(?=[A-Z0-9-]*\d)[A-Z0-9]{5,}|Policy[-\s]*(?:ID|No\.?|Number)[-:\s]*[A-Z0-9]*\d[A-Z0-9-]{4,})\b", re.I),
+                "[HASHED_POLICY_NUMBER]",
+            ),
+            (
+                "claim_id",
+                re.compile(r"\b(?:CLM[-\s]?(?=[A-Z0-9-]*\d)[A-Z0-9]{5,}|Claim[-\s]?(?:ID|No\.?|Number)?[-\s]*[A-Z0-9]*\d[A-Z0-9-]{4,})\b", re.I),
+                "[HASHED_CLAIM_ID]",
+            ),
+        ]
+
+    def sanitize(self, text: str) -> GuardrailResult:
+        findings: list[str] = []
+        sanitized = text
+        for name, pattern, replacement in self.patterns:
+            if pattern.search(sanitized):
+                findings.append(name)
+                sanitized = pattern.sub(replacement, sanitized)
+        return GuardrailResult(text=sanitized, blocked=False, findings=findings)
+
+    def clean_legacy_false_positive_placeholders(self, text: str) -> str:
+        """Repair old cached answers where normal policy words were over-masked."""
+        cleaned = text
+        replacements = [
+            (r"standard insurance \[HASHED_POLICY_NUMBER\]", "standard insurance policy"),
+            (r"insurance \[HASHED_POLICY_NUMBER\]", "insurance policy"),
+            (r"\b[Tt]he \[HASHED_POLICY_NUMBER\]'s", "the policyholder's"),
+            (r"\b[Tt]he \[HASHED_POLICY_NUMBER\]", "the policyholder"),
+            (r"\[HASHED_POLICY_NUMBER\]'s specific policy documents", "the policyholder's specific policy documents"),
+            (r"\[HASHED_POLICY_NUMBER\] documents", "policy documents"),
+            (r"\[HASHED_POLICY_NUMBER\] or endorsements", "policy documents or endorsements"),
+        ]
+        for pattern, replacement in replacements:
+            cleaned = re.sub(pattern, replacement, cleaned)
+        return cleaned

app/guardrails/prompt_injection.py

@@ -0,0 +1,25 @@
+import re
+
+
+INJECTION_PATTERNS = [
+    r"ignore\s+(all\s+)?previous\s+instructions",
+    r"reveal\s+(the\s+)?system\s+prompt",
+    r"developer\s+message",
+    r"act\s+as\s+dan",
+    r"jailbreak",
+]
+
+
+def detect_prompt_injection(text: str) -> list[str]:
+    findings = []
+    for pattern in INJECTION_PATTERNS:
+        if re.search(pattern, text, flags=re.I):
+            findings.append(pattern)
+    return findings
+
+
+def strip_unsafe_retrieved_text(text: str) -> str:
+    cleaned = text
+    for pattern in INJECTION_PATTERNS:
+        cleaned = re.sub(pattern, "[REMOVED_PROMPT_INJECTION]", cleaned, flags=re.I)
+    return cleaned

app/main.py

@@ -0,0 +1,54 @@
+from pathlib import Path
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.staticfiles import StaticFiles
+from fastapi.responses import FileResponse
+
+from app.api.routes_health import router as health_router
+from app.api.routes_ingest import router as ingest_router
+from app.api.routes_query import router as query_router
+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
+
+
+configure_logging()
+
+app = FastAPI(title=settings.app_name)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+app.include_router(health_router, prefix="/api")
+app.include_router(query_router, prefix="/api")
+app.include_router(ingest_router, prefix="/api")
+app.include_router(review_router, prefix="/api")
+
+frontend_dir = Path(__file__).resolve().parent.parent / "frontend"
+assets_dir = frontend_dir / "assets"
+app.mount("/assets", StaticFiles(directory=assets_dir), name="assets")
+
+
+@app.on_event("startup")
+def startup() -> None:
+    settings.ensure_directories()
+    init_db()
+    QdrantVectorStore().ensure_collections()
+    if settings.auto_ingest_pdfs_on_startup:
+        DocumentIngestionService().ingest_pdf_directory(settings.document_dir)
+    BM25Index.load_or_create().save()
+
+
+@app.get("/")
+def index() -> FileResponse:
+    return FileResponse(frontend_dir / "index.html")

app/rag/__init__.py

@@ -0,0 +1 @@
+"""RAG pipeline package."""

app/rag/bm25.py

@@ -0,0 +1,75 @@
+import json
+from pathlib import Path
+from typing import Any
+
+from rank_bm25 import BM25Okapi
+
+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
+        ]
+        return cls(docs)
+
+    @classmethod
+    def load_or_create(cls) -> "BM25Index":
+        path = Path(settings.bm25_index_path)
+        if not path.exists():
+            return cls.from_db()
+        try:
+            payload = json.loads(path.read_text(encoding="utf-8"))
+            return cls(payload.get("docs", []))
+        except (OSError, json.JSONDecodeError):
+            return cls.from_db()
+
+    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()
+        self.docs = fresh.docs
+        self.tokens = fresh.tokens
+        self.index = fresh.index
+        self.save()
+
+    def search(self, query: str, top_k: int) -> list[dict[str, Any]]:
+        if not self.index 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
+        ]

app/rag/document_cache.py

@@ -0,0 +1,97 @@
+import json
+from dataclasses import dataclass
+
+from app.db.sqlite import db
+from app.rag.text import hamming_distance, normalize_text, sha256_text, simhash
+
+
+@dataclass
+class DocumentCacheDecision:
+    should_embed: bool
+    status: str
+    matched_doc_id: str | None = None
+    reason: str = ""
+
+
+class DocumentCache:
+    near_duplicate_hamming_threshold = 4
+
+    def inspect(self, raw_text: str) -> DocumentCacheDecision:
+        normalized = normalize_text(raw_text)
+        file_hash = sha256_text(raw_text)
+        normalized_hash = sha256_text(normalized)
+        signature = simhash(normalized)
+
+        with db() as conn:
+            exact = conn.execute(
+                "SELECT doc_id FROM documents WHERE file_hash = ? OR normalized_hash = ?",
+                (file_hash, normalized_hash),
+            ).fetchone()
+            if exact:
+                return DocumentCacheDecision(
+                    should_embed=False,
+                    status="skipped_exact_duplicate",
+                    matched_doc_id=exact["doc_id"],
+                    reason="Document hash already exists.",
+                )
+
+            rows = conn.execute("SELECT doc_id, simhash FROM documents").fetchall()
+            for row in rows:
+                distance = hamming_distance(signature, int(row["simhash"]))
+                if distance <= self.near_duplicate_hamming_threshold:
+                    return DocumentCacheDecision(
+                        should_embed=False,
+                        status="skipped_near_duplicate",
+                        matched_doc_id=row["doc_id"],
+                        reason=f"Near-duplicate SimHash distance {distance}.",
+                    )
+
+        return DocumentCacheDecision(should_embed=True, status="new_document")
+
+    def chunk_exists(self, text_hash: str) -> bool:
+        with db() as conn:
+            row = conn.execute("SELECT chunk_id FROM chunks WHERE text_hash = ?", (text_hash,)).fetchone()
+            return row is not None
+
+    def save_document(
+        self,
+        doc_id: str,
+        source_name: str,
+        raw_text: str,
+        status: str,
+    ) -> None:
+        normalized = normalize_text(raw_text)
+        with db() as conn:
+            conn.execute(
+                """
+                INSERT INTO documents(doc_id, source_name, file_hash, normalized_hash, simhash, status)
+                VALUES (?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    doc_id,
+                    source_name,
+                    sha256_text(raw_text),
+                    sha256_text(normalized),
+                    str(simhash(normalized)),
+                    status,
+                ),
+            )
+
+    def save_chunk(
+        self,
+        chunk_id: str,
+        doc_id: str,
+        chunk_index: int,
+        text: str,
+        text_hash: str,
+        metadata: dict,
+        embedded: bool,
+    ) -> None:
+        with db() as conn:
+            conn.execute(
+                """
+                INSERT OR IGNORE INTO chunks(chunk_id, doc_id, chunk_index, text, text_hash, metadata_json, embedded_at)
+                VALUES (?, ?, ?, ?, ?, ?, CASE WHEN ? THEN CURRENT_TIMESTAMP ELSE NULL END)
+                """,
+                (chunk_id, doc_id, chunk_index, text, text_hash, json.dumps(metadata), int(embedded)),
+            )

app/rag/embeddings.py

@@ -0,0 +1,26 @@
+from functools import lru_cache
+
+from app.core.config import settings
+
+
+class EmbeddingModel:
+    def __init__(self) -> None:
+        from langchain_huggingface import HuggingFaceEmbeddings
+
+        self.model = HuggingFaceEmbeddings(
+            model_name=settings.embedding_model,
+            encode_kwargs={"normalize_embeddings": True},
+        )
+
+    def embed_query(self, text: str) -> list[float]:
+        return self.model.embed_query(text)
+
+    def embed_documents(self, texts: list[str]) -> list[list[float]]:
+        if not texts:
+            return []
+        return self.model.embed_documents(texts)
+
+
+@lru_cache
+def get_embedding_model() -> EmbeddingModel:
+    return EmbeddingModel()

app/rag/graph.py

@@ -0,0 +1,497 @@
+import json
+import re
+from typing import Any
+
+from langgraph.graph import END, StateGraph
+
+from app.core.config import settings
+from app.db.sqlite import db
+from app.guardrails.pii import PIIGuardrails
+from app.guardrails.prompt_injection import detect_prompt_injection, strip_unsafe_retrieved_text
+from app.rag.hybrid_retriever import HybridRetriever
+from app.rag.reranker import FlashRankReranker
+from app.rag.semantic_cache import SemanticAnswerCache
+from app.rag.self_rag import SelfRAG
+from app.rag.state import RagState
+from app.rag.text import new_id, normalize_text
+from app.services.groq_llm import GroqLLM
+from app.services.memory import ClaimMemoryService
+
+
+class ClaimsRAGGraph:
+    def __init__(self) -> None:
+        self.cache = SemanticAnswerCache()
+        self.guardrails = PIIGuardrails()
+        self.retriever = HybridRetriever()
+        self.reranker = FlashRankReranker()
+        self.self_rag = SelfRAG()
+        self.llm = GroqLLM()
+        self.memory = ClaimMemoryService()
+        self.graph = self._build_graph()
+
+    def run(
+        self,
+        query: str,
+        metadata_filter: dict[str, Any] | None = None,
+        user_id: str = "default_user",
+        use_cache: bool = True,
+    ) -> RagState:
+        request_id = new_id("request")
+        state: RagState = {
+            "request_id": request_id,
+            "query": query,
+            "user_id": user_id,
+            "sanitized_query": query,
+            "retrieval_query": query,
+            "normalized_query": normalize_text(query),
+            "memory_context": "",
+            "iteration": 0,
+            "cache_hit": False,
+            "use_cache": use_cache,
+            "trace": [{"node": "start", "query": query, "metadata_filter": metadata_filter or {}}],
+        }
+        if metadata_filter:
+            state["metadata_filter"] = metadata_filter  # type: ignore[typeddict-unknown-key]
+        result = self.graph.invoke(state)
+        self._save_trace(result)
+        return result
+
+    def _build_graph(self):
+        workflow = StateGraph(RagState)
+        workflow.add_node("planner", self._planner)
+        workflow.add_node("semantic_cache", self._semantic_cache)
+        workflow.add_node("guardrails", self._guardrails)
+        workflow.add_node("load_memory", self._load_memory)
+        workflow.add_node("direct_response", self._direct_response)
+        workflow.add_node("retrieve", self._retrieve)
+        workflow.add_node("rerank", self._rerank)
+        workflow.add_node("generate", self._generate)
+        workflow.add_node("critique", self._critique)
+        workflow.add_node("rewrite", self._rewrite)
+        workflow.add_node("finalize", self._finalize)
+
+        workflow.set_entry_point("planner")
+        workflow.add_edge("planner", "semantic_cache")
+        workflow.add_conditional_edges(
+            "semantic_cache",
+            self._route_cache,
+            {"hit": "finalize", "miss": "guardrails"},
+        )
+        workflow.add_conditional_edges(
+            "guardrails",
+            lambda state: "direct" if state.get("answer") and "override system" in state["answer"] else "memory",
+            {"direct": "direct_response", "memory": "load_memory"},
+        )
+        workflow.add_conditional_edges(
+            "load_memory",
+            self._route_retrieval,
+            {"direct": "direct_response", "retrieve": "retrieve"},
+        )
+        workflow.add_edge("direct_response", "finalize")
+        workflow.add_edge("retrieve", "rerank")
+        workflow.add_edge("rerank", "generate")
+        workflow.add_edge("generate", "critique")
+        workflow.add_conditional_edges(
+            "critique",
+            self._route_critique,
+            {"accept": "finalize", "retry": "rewrite"},
+        )
+        workflow.add_edge("rewrite", "retrieve")
+        workflow.add_edge("finalize", END)
+        return workflow.compile()
+
+    def _planner(self, state: RagState) -> RagState:
+        decision = self.self_rag.grade_retrieval_need(state["query"])
+        state["should_retrieve"] = bool(decision.get("should_retrieve", True))
+        state["intent"] = str(decision.get("intent", "claims_question"))
+        state["risk_level"] = decision.get("risk_level", "medium")  # type: ignore[assignment]
+        self._trace(state, "planner", decision)
+        return state
+
+    def _load_memory(self, state: RagState) -> RagState:
+        memory_context = self.memory.search(
+            user_id=state.get("user_id", "default_user"),
+            query=state["sanitized_query"],
+        )
+        state["memory_context"] = memory_context
+        self._trace(
+            state,
+            "load_memory",
+            {
+                "langmem_available": self.memory.langmem_available,
+                "has_memory": memory_context != "No prior memory for this user.",
+            },
+        )
+        return state
+
+    def _semantic_cache(self, state: RagState) -> RagState:
+        if not state.get("use_cache", True):
+            state["cache_hit"] = False
+            self._trace(state, "semantic_cache", {"hit": False, "disabled": True})
+            return state
+        hit = self.cache.lookup(state["query"])
+        if hit:
+            state["cache_hit"] = True
+            state["answer"] = hit["answer"]
+            state["confidence"] = hit["confidence"]
+            state["sources"] = hit["sources"]
+            has_sources = bool(hit["sources"])
+            state["self_rag"] = {
+                "passed": True,
+                "retrieve": True,
+                "isrel": has_sources,
+                "issup": has_sources,
+                "isuse": bool(hit["answer"]),
+                "confidence": hit["confidence"],
+                "issues": ["Served from semantic answer cache."],
+            }
+            self._trace(state, "semantic_cache", {"hit": True, "score": hit["score"]})
+            return state
+        state["cache_hit"] = False
+        self._trace(state, "semantic_cache", {"hit": False})
+        return state
+
+    def _guardrails(self, state: RagState) -> RagState:
+        pii = self.guardrails.sanitize(state["query"])
+        injection = detect_prompt_injection(state["query"])
+        state["sanitized_query"] = pii.text
+        if injection:
+            state["should_retrieve"] = False
+            state["answer"] = "I cannot help with requests that try to override system or safety instructions."
+            state["confidence"] = 0.99
+        self._trace(
+            state,
+            "guardrails",
+            {
+                "pii_findings": pii.findings,
+                "prompt_injection_findings": injection,
+                "langchain_pii_middlewares": len(self.guardrails.langchain_middlewares),
+            },
+        )
+        return state
+
+    def _direct_response(self, state: RagState) -> RagState:
+        if state.get("answer"):
+            return state
+        answer = self.llm.invoke_text(
+            system=(
+                "You are an insurance claims support copilot. Answer simple non-policy questions "
+                "briefly. Do not invent coverage, policy terms, claim outcomes, or payments."
+            ),
+            user=state["sanitized_query"],
+        )
+        state["answer"] = answer
+        state["confidence"] = 0.72
+        state["sources"] = []
+        state["self_rag"] = {
+            "passed": True,
+            "confidence": 0.72,
+            "issues": ["Direct response path. Retrieval was not required."],
+        }
+        self._trace(state, "direct_response", {"confidence": state["confidence"]})
+        return state
+
+    def _retrieve(self, state: RagState) -> RagState:
+        query = self._prepare_retrieval_query(state)
+        metadata_filter = state.get("metadata_filter")  # type: ignore[typeddict-item]
+        sources = self.retriever.retrieve(query, metadata_filter=metadata_filter)
+        cleaned_sources = []
+        for source in sources:
+            cleaned_sources.append({**source, "text": strip_unsafe_retrieved_text(source.get("text", ""))})
+        state["sources"] = cleaned_sources
+        self._trace(state, "retrieve", {"count": len(cleaned_sources), "retrieval_query": query})
+        return state
+
+    def _rerank(self, state: RagState) -> RagState:
+        reranked = self.reranker.rerank(
+            state.get("retrieval_query", state["sanitized_query"]),
+            state.get("sources", []),
+            top_k=settings.rerank_top_k,
+        )
+        state["reranked_sources"] = reranked
+        self._trace(state, "rerank", {"count": len(reranked)})
+        return state
+
+    def _prepare_retrieval_query(self, state: RagState) -> str:
+        if not settings.enable_query_rewrite:
+            state["retrieval_query"] = state["sanitized_query"]
+            return state["retrieval_query"]
+        if state.get("retrieval_query") and state.get("retrieval_query") != state.get("query"):
+            return state["retrieval_query"]
+        result = self.llm.invoke_json(
+            system=(
+                "You rewrite user insurance questions into concise retrieval queries for a hybrid "
+                "BM25 + vector RAG system. Preserve all facts from the user. Do not answer the "
+                "question. Add only helpful insurance terminology that improves retrieval, such as "
+                "coverage part, exclusion, deductible, endorsement, claim documents, fault, valuation, "
+                "or policy limit when relevant.\n\n"
+                "Important retrieval discipline:\n"
+                "- Keep the rewritten query focused on the user's requested claim issue.\n"
+                "- Do not add benefits, services, or subtopics the user did not ask about.\n"
+                "- If the user asks about damage to insured property, focus on the coverage for that "
+                "damage, required evidence, deductible, and exclusions.\n"
+                "- If a term in the user question is vague, rewrite it into precise insurance language, "
+                "but do not change the claim type.\n"
+                "- Prefer compact keyword-rich wording over a sentence.\n\n"
+                "Return JSON only with keys: query, changed, rationale."
+            ),
+            user=(
+                f"Intent: {state.get('intent', 'unknown')}\n"
+                f"Original user question:\n{state['sanitized_query']}"
+            ),
+            fallback={"query": state["sanitized_query"], "changed": False, "rationale": "fallback"},
+        )
+        rewritten = str(result.get("query") or state["sanitized_query"]).strip()
+        if not rewritten:
+            rewritten = state["sanitized_query"]
+        state["retrieval_query"] = rewritten
+        self._trace(
+            state,
+            "query_rewrite",
+            {
+                "original_query": state["sanitized_query"],
+                "retrieval_query": rewritten,
+                "changed": bool(result.get("changed", rewritten != state["sanitized_query"])),
+                "rationale": str(result.get("rationale", ""))[:300],
+            },
+        )
+        return rewritten
+
+    def _generate(self, state: RagState) -> RagState:
+        sources = state.get("reranked_sources", [])
+        llm_sources = sources[: settings.max_sources_to_llm]
+        evidence = "\n\n".join(
+            (
+                f"Source {i + 1}: {src.get('source_name', 'unknown')}\n"
+                f"Text: {src.get('text', '')[: settings.max_evidence_chars_per_source]}"
+            )
+            for i, src in enumerate(llm_sources)
+        )
+        if state.get("intent") == "general_insurance_concept":
+            answer = self.llm.invoke_text(
+                system=(
+                    "You are an insurance education assistant. The user is asking a general "
+                    "insurance concept question, not requesting a claim payment decision. Use only "
+                    "the provided evidence. Answer briefly and clearly in plain language. Do not use "
+                    "the claim triage structure. Do not say Likely covered, Likely not covered, or "
+                    "Needs human review unless the user asks about a claim scenario. Cite sources as "
+                    "[Source 1], [Source 2], etc. Do not use outside source names."
+                ),
+                user=(
+                    f"Question:\n{state['sanitized_query']}\n\n"
+                    f"Retrieved evidence:\n{evidence}"
+                ),
+            )
+            state["answer"] = self._ensure_source_citation(answer, sources)
+            self._trace(state, "generate", {"source_count": len(sources), "mode": "concept_llm"})
+            return state
+        answer = self._generate_claim_json_answer(state, evidence, sources)
+        state["answer"] = self._ensure_source_citation(answer, sources)
+        self._trace(state, "generate", {"source_count": len(sources), "mode": "llm"})
+        return state
+
+    def _ensure_source_citation(self, answer: str, sources: list[dict[str, Any]]) -> str:
+        if not sources or re.search(r"\[Source\s+\d+\]", answer, flags=re.IGNORECASE):
+            return answer
+        return answer.rstrip() + " [Source 1]"
+
+    def _generate_claim_json_answer(self, state: RagState, evidence: str, sources: list[dict[str, Any]]) -> str:
+        result = self.llm.invoke_json(
+            system=(
+                "You are an insurance claim support AI agent. The user describes a claim scenario. "
+                "Use only the retrieved evidence. Do not use outside knowledge. Do not invent final "
+                "payment approval, denial, claim status, policy terms, or source names.\n\n"
+                "Return JSON only with exactly these keys:\n"
+                "- decision: one of Likely covered, Likely not covered, Needs human review\n"
+                "- reason: one or two evidence-grounded sentences\n"
+                "- missing_evidence: short string listing missing documents/facts or None identified\n"
+                "- recommended_action: short string with next step, escalation, or review action\n"
+                "- sources: short string with citations like [Source 1], [Source 2]\n\n"
+                "Rubric:\n"
+                "- Likely covered: use only when evidence directly says this cause of loss or scenario is "
+                "normally covered by the relevant coverage part and the user's facts do not leave a major "
+                "coverage dependency unresolved.\n"
+                "- Likely not covered: use only when evidence directly says this cause of loss is excluded, "
+                "not covered by the standard policy, or requires separate coverage that the user says they "
+                "do not have.\n"
+                "- Needs human review: use when payment or coverage depends on unresolved policy-specific "
+                "facts, endorsements, sublimits, deductibles, fault, valuation, contestability, fraud "
+                "review, regulatory timing, guaranty fund state limits, settlement amount disputes, or "
+                "other claim-file details.\n\n"
+                "If evidence is incomplete, still choose the best triage label and explain what is missing. "
+                "Allowed citations are only [Source 1], [Source 2], etc."
+            ),
+            user=(
+                f"User memory context:\n{state.get('memory_context', 'No prior memory.')}\n\n"
+                f"Claim scenario:\n{state['sanitized_query']}\n\n"
+                f"Retrieved evidence:\n{evidence}"
+            ),
+            fallback={
+                "decision": "Needs human review",
+                "reason": "The available evidence is not sufficient to make a final coverage triage.",
+                "missing_evidence": "Policy-specific details and claim-file documentation.",
+                "recommended_action": "Escalate for human review with the retrieved evidence.",
+                "sources": "[Source 1]" if sources else "",
+            },
+        )
+        decision = str(result.get("decision", "Needs human review"))
+        if decision not in {"Likely covered", "Likely not covered", "Needs human review"}:
+            decision = "Needs human review"
+        sources_text = str(result.get("sources", "")).strip()
+        if sources and not re.search(r"\[Source\s+\d+\]", sources_text, flags=re.IGNORECASE):
+            sources_text = "[Source 1]"
+        return (
+            f"Decision: {decision}\n"
+            f"Reason: {str(result.get('reason', '')).strip()}\n"
+            f"Missing evidence: {str(result.get('missing_evidence', '')).strip()}\n"
+            f"Recommended action: {str(result.get('recommended_action', '')).strip()}\n"
+            f"Sources: {sources_text}"
+        )
+
+    def _critique(self, state: RagState) -> RagState:
+        if state.get("intent") == "general_insurance_concept":
+            sources = state.get("reranked_sources", [])
+            answer = state.get("answer", "")
+            critique = {
+                "passed": bool(sources) and bool(answer),
+                "retrieve": True,
+                "isrel": bool(sources),
+                "issup": bool(sources) and "[source" in answer.lower(),
+                "isuse": len(answer.strip()) > 20,
+                "confidence": 0.84 if sources and answer else 0.45,
+                "relevance_score": 0.85 if sources else 0.0,
+                "faithfulness_score": 0.8 if sources and "[source" in answer.lower() else 0.35,
+                "evidence_score": min(0.9, 0.35 + len(sources) * 0.1),
+                "needs_rewrite": False,
+                "rewrite_query": None,
+                "issues": ["General insurance concept answer with retrieved sources."],
+            }
+            state["self_rag"] = critique
+            state["confidence"] = float(critique["confidence"])
+            self._trace(state, "critique", critique)
+            return state
+        critique = self.self_rag.critique(
+            query=state["sanitized_query"],
+            answer=state.get("answer", ""),
+            sources=state.get("reranked_sources", []),
+            iteration=int(state.get("iteration", 0)),
+        )
+        state["self_rag"] = critique
+        state["confidence"] = float(critique.get("confidence", 0.0))
+        self._trace(state, "critique", critique)
+        return state
+
+    def _rewrite(self, state: RagState) -> RagState:
+        critique = state.get("self_rag", {})
+        fallback_query = critique.get("rewrite_query") or state.get("retrieval_query") or state["sanitized_query"]
+        result = self.llm.invoke_json(
+            system=(
+                "You are rewriting a failed insurance RAG retrieval query for a retry. Use the "
+                "critique issues and original user question to create a better retrieval query. "
+                "Preserve the user's facts. Do not answer the question. Return JSON only with "
+                "keys: query, rationale."
+            ),
+            user=(
+                f"Original user question:\n{state['sanitized_query']}\n\n"
+                f"Previous retrieval query:\n{state.get('retrieval_query', state['sanitized_query'])}\n\n"
+                f"Critique issues:\n{json.dumps(critique.get('issues', []), ensure_ascii=True)}"
+            ),
+            fallback={"query": fallback_query, "rationale": "fallback"},
+        )
+        rewrite = str(result.get("query") or fallback_query).strip() or str(fallback_query)
+        state["retrieval_query"] = rewrite
+        state["iteration"] = int(state.get("iteration", 0)) + 1
+        self._trace(
+            state,
+            "rewrite",
+            {
+                "retrieval_query": state["retrieval_query"],
+                "iteration": state["iteration"],
+                "rationale": str(result.get("rationale", ""))[:300],
+            },
+        )
+        return state
+
+    def _finalize(self, state: RagState) -> RagState:
+        if state.get("answer"):
+            sanitized = self.guardrails.sanitize(state["answer"]).text
+            state["answer"] = self.guardrails.clean_legacy_false_positive_placeholders(sanitized)
+        if (
+            state.get("use_cache", True)
+            and not state.get("cache_hit")
+            and state.get("answer")
+            and not self._has_unsupported_citation(state)
+        ):
+            self.cache.save(
+                query=state["query"],
+                answer=state["answer"],
+                confidence=float(state.get("confidence", 0.0)),
+                sources=state.get("reranked_sources") or state.get("sources", []),
+            )
+        if state.get("answer") and state.get("reranked_sources"):
+            self.memory.save_interaction(
+                user_id=state.get("user_id", "default_user"),
+                query=state["query"],
+                answer=state["answer"],
+                critique=state.get("self_rag", {}),
+                sources=state.get("reranked_sources", []),
+            )
+        self._trace(
+            state,
+            "finalize",
+            {
+                "cache_hit": state.get("cache_hit", False),
+                "confidence": state.get("confidence", 0.0),
+                "iterations": state.get("iteration", 0),
+            },
+        )
+        return state
+
+    def _has_unsupported_citation(self, state: RagState) -> bool:
+        answer = state.get("answer", "").lower()
+        blocked_markers = [
+            "insurance information institute",
+            "source: none",
+            "source: insurance",
+            "according to standard insurance terminology",
+        ]
+        return any(marker in answer for marker in blocked_markers)
+
+    def _route_cache(self, state: RagState) -> str:
+        return "hit" if state.get("cache_hit") else "miss"
+
+    def _route_retrieval(self, state: RagState) -> str:
+        if state.get("answer") and "override system" in state["answer"]:
+            return "direct"
+        return "retrieve" if state.get("should_retrieve", True) else "direct"
+
+    def _route_critique(self, state: RagState) -> str:
+        critique = state.get("self_rag", {})
+        passed = bool(critique.get("passed", False))
+        confidence = float(critique.get("confidence", 0.0))
+        iteration = int(state.get("iteration", 0))
+        if passed and confidence >= 0.68:
+            return "accept"
+        if iteration >= settings.self_rag_max_loops:
+            return "accept"
+        if critique.get("needs_rewrite", True):
+            return "retry"
+        return "accept"
+
+    def _trace(self, state: RagState, node: str, payload: dict[str, Any]) -> None:
+        trace = state.setdefault("trace", [])
+        trace.append({"node": node, **payload})
+
+    def _save_trace(self, state: RagState) -> None:
+        with db() as conn:
+            conn.execute(
+                """
+                INSERT OR REPLACE INTO traces(request_id, query, trace_json)
+                VALUES (?, ?, ?)
+                """,
+                (
+                    state["request_id"],
+                    state["query"],
+                    json.dumps(state.get("trace", []), ensure_ascii=True),
+                ),
+            )

app/rag/hybrid_retriever.py

@@ -0,0 +1,29 @@
+from typing import Any
+
+from app.core.config import settings
+from app.rag.bm25 import BM25Index
+from app.rag.embeddings import get_embedding_model
+from app.rag.qdrant_store import QdrantVectorStore
+from app.rag.rrf import reciprocal_rank_fusion
+
+
+class HybridRetriever:
+    def __init__(self) -> None:
+        self.embeddings = get_embedding_model()
+        self.qdrant = QdrantVectorStore()
+        self.bm25 = BM25Index.load_or_create()
+
+    def retrieve(self, query: str, metadata_filter: dict[str, Any] | None = None) -> list[dict[str, Any]]:
+        vector = self.embeddings.embed_query(query)
+        vector_hits = self.qdrant.search_chunks(
+            vector=vector,
+            top_k=settings.retrieval_top_k,
+            metadata_filter=metadata_filter,
+        )
+        for hit in vector_hits:
+            hit["metadata"] = {**hit.get("metadata", {}), "retriever": "qdrant"}
+        bm25_hits = self.bm25.search(query, top_k=settings.retrieval_top_k)
+        return reciprocal_rank_fusion(
+            [bm25_hits, vector_hits],
+            top_k=max(settings.retrieval_top_k, settings.rerank_top_k),
+        )

app/rag/ingestion.py

@@ -0,0 +1,143 @@
+import logging
+from pathlib import Path
+from typing import Any
+
+from langchain_community.document_loaders import PyPDFLoader
+from langchain_core.documents import Document
+from langchain_text_splitters import RecursiveCharacterTextSplitter
+
+from app.core.config import settings
+from app.rag.bm25 import BM25Index
+from app.rag.document_cache import DocumentCache
+from app.rag.embeddings import get_embedding_model
+from app.rag.qdrant_store import QdrantVectorStore
+from app.rag.text import new_id, normalize_text, sha256_text
+
+logger = logging.getLogger(__name__)
+
+
+class DocumentIngestionService:
+    def __init__(self) -> None:
+        self.cache = DocumentCache()
+        self._embeddings = None
+        self.qdrant = QdrantVectorStore()
+        self.splitter = RecursiveCharacterTextSplitter(
+            chunk_size=900,
+            chunk_overlap=140,
+            separators=["\n\n", "\n", ". ", " ", ""],
+        )
+
+    @property
+    def embeddings(self):
+        if self._embeddings is None:
+            self._embeddings = get_embedding_model()
+        return self._embeddings
+
+    def ingest_text(self, text: str, source_name: str, metadata: dict[str, Any] | None = None) -> dict[str, Any]:
+        docs = [Document(page_content=text, metadata=metadata or {})]
+        return self.ingest_documents(docs, source_name=source_name)
+
+    def ingest_pdf(self, pdf_path: Path) -> dict[str, Any]:
+        loader = PyPDFLoader(str(pdf_path))
+        docs = loader.load()
+        metadata = {
+            "document_type": "pdf",
+            "source_path": str(pdf_path),
+            "source_name": pdf_path.name,
+        }
+        enriched = [
+            Document(page_content=doc.page_content, metadata={**metadata, **doc.metadata})
+            for doc in docs
+        ]
+        return self.ingest_documents(enriched, source_name=pdf_path.name)
+
+    def ingest_pdf_directory(self, directory: str | Path | None = None) -> dict[str, Any]:
+        root = Path(directory or settings.document_dir)
+        root.mkdir(parents=True, exist_ok=True)
+        pdfs = sorted(
+            path for path in root.rglob("*.pdf")
+            if not any(part.startswith(".") for part in path.parts)
+        )
+        results = []
+        for pdf in pdfs:
+            try:
+                results.append({"file": str(pdf), **self.ingest_pdf(pdf)})
+            except Exception as exc:
+                logger.exception("Failed to ingest PDF %s", pdf)
+                results.append({"file": str(pdf), "status": "failed", "reason": str(exc)})
+        return {
+            "status": "scanned",
+            "directory": str(root),
+            "pdf_count": len(pdfs),
+            "results": results,
+        }
+
+    def ingest_documents(self, docs: list[Document], source_name: str) -> dict[str, Any]:
+        text = "\n\n".join(doc.page_content for doc in docs if doc.page_content.strip())
+        if not text.strip():
+            return {
+                "status": "skipped_empty_document",
+                "source_name": source_name,
+                "embedded_chunks": 0,
+                "skipped_chunks": 0,
+            }
+        decision = self.cache.inspect(text)
+        if not decision.should_embed:
+            return {
+                "status": decision.status,
+                "matched_doc_id": decision.matched_doc_id,
+                "reason": decision.reason,
+                "embedded_chunks": 0,
+                "skipped_chunks": 0,
+            }
+
+        doc_id = new_id("doc")
+        self.cache.save_document(doc_id, source_name, text, "embedded")
+
+        split_docs = self.splitter.split_documents(docs)
+        chunk_records = []
+        new_chunk_texts = []
+        skipped_chunks = 0
+
+        for index, split_doc in enumerate(split_docs):
+            chunk = split_doc.page_content.strip()
+            if not chunk:
+                continue
+            text_hash = sha256_text(normalize_text(chunk))
+            chunk_id = new_id("chunk")
+            chunk_metadata = {
+                **split_doc.metadata,
+                "doc_id": doc_id,
+                "chunk_id": chunk_id,
+                "chunk_index": index,
+                "source_name": source_name,
+                "text_hash": text_hash,
+            }
+            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)
+                continue
+            chunk_records.append((chunk_id, index, chunk, text_hash, chunk_metadata))
+            new_chunk_texts.append(chunk)
+
+        vectors = self.embeddings.embed_documents(new_chunk_texts)
+        points = []
+        for (chunk_id, index, chunk, text_hash, chunk_metadata), vector in zip(chunk_records, vectors):
+            self.cache.save_chunk(chunk_id, doc_id, index, chunk, text_hash, chunk_metadata, embedded=True)
+            points.append(
+                {
+                    "id": chunk_id,
+                    "vector": vector,
+                    "payload": {**chunk_metadata, "text": chunk},
+                }
+            )
+
+        self.qdrant.upsert_chunks(points)
+        BM25Index.from_db().save()
+
+        return {
+            "status": "embedded",
+            "doc_id": doc_id,
+            "embedded_chunks": len(points),
+            "skipped_chunks": skipped_chunks,
+        }

app/rag/qdrant_store.py

@@ -0,0 +1,110 @@
+import logging
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+from qdrant_client import QdrantClient
+from qdrant_client.http.models import Distance, FieldCondition, Filter, MatchValue, PointStruct, VectorParams
+
+from app.core.config import settings
+
+logger = logging.getLogger(__name__)
+
+
+@lru_cache
+def get_qdrant_client() -> QdrantClient:
+    if settings.qdrant_url.startswith("local:"):
+        path = settings.qdrant_url.removeprefix("local:")
+        if path == ":memory:":
+            return QdrantClient(":memory:")
+        Path(path).parent.mkdir(parents=True, exist_ok=True)
+        return QdrantClient(path=path)
+    return QdrantClient(
+        url=settings.qdrant_url,
+        api_key=settings.qdrant_api_key or None,
+        timeout=10,
+    )
+
+
+class QdrantVectorStore:
+    def __init__(self) -> None:
+        self.client = get_qdrant_client()
+    def ensure_collections(self) -> None:
+        for name in [settings.qdrant_collection, settings.qdrant_cache_collection]:
+            existing = [c.name for c in self.client.get_collections().collections]
+            if name not in existing:
+                self.client.create_collection(
+                    collection_name=name,
+                    vectors_config=VectorParams(size=settings.embedding_dim, distance=Distance.COSINE),
+                )
+                logger.info("Created Qdrant collection %s", name)
+
+    def upsert_chunks(self, points: list[dict[str, Any]]) -> None:
+        if not points:
+            return
+        self.client.upsert(
+            collection_name=settings.qdrant_collection,
+            points=[
+                PointStruct(id=p["id"], vector=p["vector"], payload=p["payload"])
+                for p in points
+            ],
+        )
+
+    def search_chunks(
+        self,
+        vector: list[float],
+        top_k: int,
+        metadata_filter: dict[str, Any] | None = None,
+    ) -> list[dict[str, Any]]:
+        query_filter = self._build_filter(metadata_filter)
+        hits = self.client.query_points(
+            collection_name=settings.qdrant_collection,
+            query=vector,
+            query_filter=query_filter,
+            limit=top_k,
+            with_payload=True,
+        ).points
+        return [
+            {
+                "id": str(hit.id),
+                "score": float(hit.score),
+                "text": hit.payload.get("text", ""),
+                "source_name": hit.payload.get("source_name", "unknown"),
+                "metadata": hit.payload,
+            }
+            for hit in hits
+        ]
+
+    def upsert_cache_answer(self, cache_id: str, vector: list[float], payload: dict[str, Any]) -> None:
+        self.client.upsert(
+            collection_name=settings.qdrant_cache_collection,
+            points=[PointStruct(id=cache_id, vector=vector, payload=payload)],
+        )
+
+    def search_cache(self, vector: list[float], top_k: int = 1) -> list[dict[str, Any]]:
+        hits = self.client.query_points(
+            collection_name=settings.qdrant_cache_collection,
+            query=vector,
+            limit=top_k,
+            with_payload=True,
+        ).points
+        return [
+            {
+                "id": str(hit.id),
+                "score": float(hit.score),
+                "payload": hit.payload or {},
+            }
+            for hit in hits
+        ]
+
+    def _build_filter(self, metadata_filter: dict[str, Any] | None) -> Filter | None:
+        if not metadata_filter:
+            return None
+        conditions = [
+            FieldCondition(key=key, match=MatchValue(value=value))
+            for key, value in metadata_filter.items()
+            if value is not None and value != ""
+        ]
+        if not conditions:
+            return None
+        return Filter(must=conditions)

app/rag/reranker.py

@@ -0,0 +1,82 @@
+import logging
+import re
+from functools import lru_cache
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+@lru_cache
+def get_flashrank_ranker():
+    from flashrank import Ranker
+
+    return Ranker(max_length=256)
+
+
+class FlashRankReranker:
+    def __init__(self) -> None:
+        self._ranker = None
+
+    @property
+    def ranker(self):
+        if self._ranker is None:
+            self._ranker = get_flashrank_ranker()
+        return self._ranker
+
+    def rerank(self, query: str, docs: list[dict[str, Any]], top_k: int) -> list[dict[str, Any]]:
+        if not docs:
+            return []
+        try:
+            from flashrank import RerankRequest
+
+            passages = [
+                {
+                    "id": doc["id"],
+                    "text": doc["text"],
+                    "meta": {**doc.get("metadata", {}), "source_name": doc.get("source_name")},
+                }
+                for doc in docs
+            ]
+            result = self.ranker.rerank(RerankRequest(query=query, passages=passages))
+            by_id = {doc["id"]: doc for doc in docs}
+            reranked = []
+            seen = set()
+            for item in result:
+                doc = by_id[str(item["id"])]
+                fingerprint = self._fingerprint(doc.get("text", ""))
+                if fingerprint in seen:
+                    continue
+                seen.add(fingerprint)
+                reranked.append(
+                    {
+                        **doc,
+                        "score": float(item.get("score", doc.get("score", 0.0))),
+                        "metadata": {**doc.get("metadata", {}), "reranker": "flashrank"},
+                    }
+                )
+                if len(reranked) >= top_k:
+                    break
+            return reranked
+        except Exception as exc:  # pragma: no cover - fallback for missing model cache
+            logger.warning("FlashRank fallback used: %s", exc)
+            return self._dedupe(
+                sorted(docs, key=lambda d: d.get("score", d.get("rrf_score", 0.0)), reverse=True),
+                top_k=top_k,
+            )
+
+    def _dedupe(self, docs: list[dict[str, Any]], top_k: int) -> list[dict[str, Any]]:
+        selected = []
+        seen = set()
+        for doc in docs:
+            fingerprint = self._fingerprint(doc.get("text", ""))
+            if fingerprint in seen:
+                continue
+            seen.add(fingerprint)
+            selected.append(doc)
+            if len(selected) >= top_k:
+                break
+        return selected
+
+    def _fingerprint(self, text: str) -> str:
+        normalized = re.sub(r"\s+", " ", text.lower()).strip()
+        return normalized[:500]

app/rag/rrf.py

@@ -0,0 +1,20 @@
+from collections import defaultdict
+from typing import Any
+
+
+def reciprocal_rank_fusion(
+    result_sets: list[list[dict[str, Any]]],
+    k: int = 60,
+    top_k: int = 10,
+) -> list[dict[str, Any]]:
+    scores: dict[str, float] = defaultdict(float)
+    docs: dict[str, dict[str, Any]] = {}
+
+    for results in result_sets:
+        for rank, doc in enumerate(results, start=1):
+            doc_id = doc["id"]
+            scores[doc_id] += 1.0 / (k + rank)
+            docs[doc_id] = {**doc, "rrf_score": scores[doc_id]}
+
+    ranked = sorted(docs.values(), key=lambda item: scores[item["id"]], reverse=True)
+    return ranked[:top_k]

app/rag/self_rag.py

@@ -0,0 +1,171 @@
+import json
+import re
+from typing import Any
+
+from app.core.config import settings
+from app.services.groq_llm import GroqLLM
+
+
+class SelfRAG:
+    def __init__(self) -> None:
+        self.llm = GroqLLM()
+
+    def grade_retrieval_need(self, query: str) -> dict[str, Any]:
+        normalized = re.sub(r"\s+", " ", query.lower()).strip(" ?!.")
+        normalized = re.sub(r"^\d+[\).\-\s]+", "", normalized).strip(" ?!.")
+        simple_markers = {"hello", "hi", "hey", "help", "what can you do"}
+        fallback = {
+            "should_retrieve": normalized not in simple_markers,
+            "retrieve": normalized not in simple_markers,
+            "intent": "smalltalk" if normalized in simple_markers else "claim_scenario",
+            "risk_level": "low" if normalized in simple_markers else "medium",
+        }
+        result = self.llm.invoke_json(
+            system=(
+                "You are the planner for an insurance RAG assistant. Classify the user's query and "
+                "decide whether retrieval from the insurance knowledge base is needed.\n\n"
+                "Return JSON with exactly these keys:\n"
+                "- should_retrieve: boolean\n"
+                "- retrieve: boolean, same value as should_retrieve\n"
+                "- intent: one of smalltalk, general_insurance_concept, claim_scenario\n"
+                "- risk_level: one of low, medium, high\n\n"
+                "Use general_insurance_concept for educational questions about insurance terms, "
+                "regulation, compliance, procedures, definitions, or how insurance works. These "
+                "questions should retrieve if they are insurance-related.\n"
+                "Use claim_scenario when the user describes an event, loss, damage, theft, injury, "
+                "death, bill, repair, approval, denial, coverage, or asks whether insurance will pay. "
+                "These questions should retrieve.\n"
+                "Use smalltalk only for greetings or capability questions. These usually do not retrieve.\n"
+                "High risk means coverage decisions, denial, settlement, legal, fraud, death, injury, "
+                "large loss, regulatory complaint, or money."
+            ),
+            user=f"Query: {query}",
+            fallback=fallback,
+        )
+        intent = str(result.get("intent", fallback["intent"]))
+        if intent not in {"smalltalk", "general_insurance_concept", "claim_scenario"}:
+            intent = fallback["intent"]
+        should_retrieve = bool(result.get("should_retrieve", fallback["should_retrieve"]))
+        if intent in {"general_insurance_concept", "claim_scenario"}:
+            should_retrieve = True
+        if intent == "smalltalk":
+            should_retrieve = False
+        risk_level = str(result.get("risk_level", fallback["risk_level"]))
+        if risk_level not in {"low", "medium", "high"}:
+            risk_level = fallback["risk_level"]
+        return {
+            "should_retrieve": should_retrieve,
+            "retrieve": should_retrieve,
+            "intent": intent,
+            "risk_level": risk_level,
+        }
+
+    def critique(
+        self,
+        query: str,
+        answer: str,
+        sources: list[dict[str, Any]],
+        iteration: int,
+    ) -> dict[str, Any]:
+        if not sources:
+            return {
+                "passed": False,
+                "retrieve": True,
+                "isrel": False,
+                "issup": False,
+                "isuse": False,
+                "confidence": 0.35,
+                "relevance_score": 0.0,
+                "faithfulness_score": 0.0,
+                "evidence_score": 0.0,
+                "needs_rewrite": iteration == 0,
+                "rewrite_query": query,
+                "issues": ["No retrieved evidence was available."],
+            }
+
+        if settings.low_latency_mode:
+            return self._heuristic_critique(answer, sources, iteration)
+
+        evidence = "\n\n".join(
+            f"[{i + 1}] {src.get('source_name', 'source')} :: {src.get('text', '')[:900]}"
+            for i, src in enumerate(sources)
+        )
+        fallback = self._heuristic_critique(answer, sources, iteration)
+        result = self.llm.invoke_json(
+            system=(
+                "You are a Self-RAG evaluator for an insurance claims assistant. Return JSON with "
+                "the classic Self-RAG labels: retrieve, isrel, issup, isuse. Definitions: retrieve "
+                "means external evidence was needed; ISREL means retrieved passages are relevant; "
+                "ISSUP means the generated answer is supported by those passages; ISUSE means the "
+                "overall response is useful for the user's claim scenario. Also return passed, "
+                "confidence, relevance_score, faithfulness_score, evidence_score, needs_rewrite, "
+                "rewrite_query, and issues."
+            ),
+            user=f"Query:\n{query}\n\nDraft answer:\n{answer}\n\nEvidence:\n{evidence}",
+            fallback=fallback,
+        )
+        return {
+            "passed": bool(result.get("passed", False)),
+            "retrieve": bool(result.get("retrieve", True)),
+            "isrel": bool(result.get("isrel", False)),
+            "issup": bool(result.get("issup", False)),
+            "isuse": bool(result.get("isuse", False)),
+            "confidence": float(result.get("confidence", 0.0)),
+            "relevance_score": float(result.get("relevance_score", 0.0)),
+            "faithfulness_score": float(result.get("faithfulness_score", 0.0)),
+            "evidence_score": float(result.get("evidence_score", 0.0)),
+            "needs_rewrite": bool(result.get("needs_rewrite", False)),
+            "rewrite_query": result.get("rewrite_query") or query,
+            "issues": result.get("issues", []),
+        }
+
+    def _heuristic_critique(self, answer: str, sources: list[dict[str, Any]], iteration: int) -> dict[str, Any]:
+        source_count = len(sources)
+        has_answer = len(answer.strip()) > 40
+        answer_lower = answer.lower()
+        has_decision = "decision:" in answer_lower
+        has_missing = "missing evidence:" in answer_lower
+        has_action = "recommended action" in answer_lower or "recommended tool" in answer_lower
+        has_citation = "[source" in answer_lower
+        query_terms = set()
+        source_terms = set()
+        for source in sources:
+            source_terms.update(re.findall(r"[a-zA-Z0-9_]+", source.get("text", "").lower()))
+        isrel = source_count > 0 and bool(source_terms)
+        issup = has_citation and source_count > 0
+        isuse = has_answer and has_decision and has_missing and has_action
+        confidence = min(
+            0.95,
+            0.35
+            + source_count * 0.07
+            + (0.15 if has_answer else 0)
+            + (0.15 if isrel else 0)
+            + (0.15 if issup else 0)
+            + (0.15 if isuse else 0),
+        )
+        passed = isrel and issup and isuse and (confidence >= 0.68 or iteration > 0)
+        issues = []
+        if not isrel:
+            issues.append("ISREL failed: retrieved passages appear weak or missing.")
+        if not issup:
+            issues.append("ISSUP failed: answer lacks clear support citation.")
+        if not isuse:
+            issues.append("ISUSE failed: answer is missing decision, missing evidence, or action structure.")
+        return {
+            "passed": passed,
+            "retrieve": True,
+            "isrel": isrel,
+            "issup": issup,
+            "isuse": isuse,
+            "confidence": confidence,
+            "relevance_score": 0.9 if isrel else 0.25,
+            "faithfulness_score": 0.85 if issup else 0.35,
+            "evidence_score": min(0.9, 0.35 + source_count * 0.1),
+            "needs_rewrite": not passed and iteration == 0,
+            "rewrite_query": None,
+            "issues": issues,
+        }
+
+
+def safe_json(data: Any) -> str:
+    return json.dumps(data, ensure_ascii=True)

app/rag/semantic_cache.py

@@ -0,0 +1,53 @@
+import json
+from typing import Any
+
+from app.core.config import settings
+from app.rag.embeddings import get_embedding_model
+from app.rag.qdrant_store import QdrantVectorStore
+from app.rag.text import new_id, normalize_text
+
+
+ANSWER_CACHE_VERSION = "rag-grounded-v4"
+
+
+class SemanticAnswerCache:
+    def __init__(self) -> None:
+        self.embeddings = get_embedding_model()
+        self.store = QdrantVectorStore()
+
+    def lookup(self, query: str) -> dict[str, Any] | None:
+        vector = self.embeddings.embed_query(normalize_text(query))
+        hits = self.store.search_cache(vector, top_k=1)
+        if not hits:
+            return None
+        best = hits[0]
+        if best["score"] < settings.semantic_cache_threshold:
+            return None
+        payload = best["payload"]
+        if payload.get("cache_version") != ANSWER_CACHE_VERSION:
+            return None
+        return {
+            "answer": payload.get("answer", ""),
+            "confidence": float(payload.get("confidence", 0.0)),
+            "sources": json.loads(payload.get("sources_json", "[]")),
+            "score": best["score"],
+        }
+
+    def save(self, query: str, answer: str, confidence: float, sources: list[dict[str, Any]]) -> None:
+        if confidence < 0.75:
+            return
+        normalized = normalize_text(query)
+        vector = self.embeddings.embed_query(normalized)
+        cache_id = new_id("cache")
+        self.store.upsert_cache_answer(
+            cache_id=cache_id,
+            vector=vector,
+            payload={
+                "query": query,
+                "normalized_query": normalized,
+                "answer": answer,
+                "confidence": confidence,
+                "cache_version": ANSWER_CACHE_VERSION,
+                "sources_json": json.dumps(sources, ensure_ascii=True),
+            },
+        )

app/rag/state.py

@@ -0,0 +1,47 @@
+from typing import Any, Literal, TypedDict
+
+
+class Source(TypedDict, total=False):
+    id: str
+    text: str
+    source_name: str
+    score: float
+    metadata: dict[str, Any]
+
+
+class Critique(TypedDict, total=False):
+    passed: bool
+    retrieve: bool
+    isrel: bool
+    issup: bool
+    isuse: bool
+    confidence: float
+    relevance_score: float
+    faithfulness_score: float
+    evidence_score: float
+    needs_rewrite: bool
+    rewrite_query: str | None
+    issues: list[str]
+
+
+class RagState(TypedDict, total=False):
+    request_id: str
+    query: str
+    sanitized_query: str
+    retrieval_query: str
+    normalized_query: str
+    intent: str
+    should_retrieve: bool
+    risk_level: Literal["low", "medium", "high"]
+    cache_hit: bool
+    use_cache: bool
+    answer: str
+    confidence: float
+    user_id: str
+    memory_context: str
+    metadata_filter: dict[str, Any]
+    sources: list[Source]
+    reranked_sources: list[Source]
+    self_rag: Critique
+    iteration: int
+    trace: list[dict[str, Any]]

app/rag/text.py

@@ -0,0 +1,56 @@
+import hashlib
+import re
+import uuid
+
+
+def normalize_text(text: str) -> str:
+    text = text.replace("\x00", " ")
+    text = re.sub(r"\s+", " ", text).strip().lower()
+    return text
+
+
+def sha256_text(text: str) -> str:
+    return hashlib.sha256(text.encode("utf-8")).hexdigest()
+
+
+def new_id(prefix: str) -> str:
+    return str(uuid.uuid4())
+
+
+def tokenize(text: str) -> list[str]:
+    return re.findall(r"[a-zA-Z0-9_]+", text.lower())
+
+
+def chunk_text(text: str, chunk_size: int = 900, overlap: int = 140) -> list[str]:
+    normalized = re.sub(r"\s+", " ", text).strip()
+    if not normalized:
+        return []
+    chunks: list[str] = []
+    start = 0
+    while start < len(normalized):
+        end = min(start + chunk_size, len(normalized))
+        chunk = normalized[start:end].strip()
+        if chunk:
+            chunks.append(chunk)
+        if end == len(normalized):
+            break
+        start = max(0, end - overlap)
+    return chunks
+
+
+def simhash(text: str, bits: int = 64) -> int:
+    tokens = tokenize(normalize_text(text))
+    vector = [0] * bits
+    for token in tokens:
+        digest = int(hashlib.md5(token.encode("utf-8")).hexdigest(), 16)
+        for i in range(bits):
+            vector[i] += 1 if digest & (1 << i) else -1
+    value = 0
+    for i, weight in enumerate(vector):
+        if weight > 0:
+            value |= 1 << i
+    return value
+
+
+def hamming_distance(a: int, b: int) -> int:
+    return (a ^ b).bit_count()

app/services/__init__.py

@@ -0,0 +1 @@
+"""External service clients."""

app/services/groq_llm.py

@@ -0,0 +1,73 @@
+import json
+import logging
+import re
+from typing import Any
+
+from app.core.config import settings
+
+logger = logging.getLogger(__name__)
+
+
+class GroqLLM:
+    def __init__(self) -> None:
+        self._llm = None
+
+    @property
+    def llm(self):
+        if self._llm is None:
+            if not settings.groq_api_key:
+                return None
+            from langchain_groq import ChatGroq
+
+            self._llm = ChatGroq(
+                model=settings.groq_model,
+                temperature=0,
+                max_retries=2,
+                api_key=settings.groq_api_key,
+            )
+        return self._llm
+
+    def invoke_text(self, system: str, user: str) -> str:
+        if self.llm is None:
+            return self._fallback_text(user)
+        response = self.llm.invoke(
+            [
+                ("system", system),
+                ("user", user),
+            ]
+        )
+        return str(response.content)
+
+    def invoke_json(self, system: str, user: str, fallback: dict[str, Any]) -> dict[str, Any]:
+        if self.llm is None:
+            return fallback
+        try:
+            response = self.llm.invoke(
+                [
+                    ("system", system + "\nReturn only valid JSON."),
+                    ("user", user),
+                ],
+                {"response_format": {"type": "json_object"}},
+            )
+            return self._parse_json(str(response.content), fallback)
+        except Exception as exc:
+            logger.warning("Groq JSON call failed, using fallback: %s", exc)
+            return fallback
+
+    def _parse_json(self, text: str, fallback: dict[str, Any]) -> dict[str, Any]:
+        try:
+            return json.loads(text)
+        except json.JSONDecodeError:
+            match = re.search(r"\{.*\}", text, flags=re.DOTALL)
+            if not match:
+                return fallback
+            try:
+                return json.loads(match.group(0))
+            except json.JSONDecodeError:
+                return fallback
+
+    def _fallback_text(self, user: str) -> str:
+        return (
+            "The Groq API key is not configured, so this local fallback cannot produce a full "
+            "LLM answer. Add GROQ_API_KEY to .env and rerun the service."
+        )

app/services/memory.py

@@ -0,0 +1,159 @@
+import json
+import re
+from typing import Any
+
+from app.db.sqlite import db
+from app.rag.text import new_id, normalize_text, tokenize
+
+
+class ClaimMemoryService:
+    """LangMem-ready memory facade with durable SQLite fallback.
+
+    The project initializes LangMem memory tools when the package is present, while
+    storing/retrieving operational memories locally so the app works without an
+    external LangGraph store.
+    """
+
+    def __init__(self) -> None:
+        self.langmem_available = False
+        self.store = None
+        self.manage_memory_tool = None
+        self.search_memory_tool = None
+        self._init_langmem()
+
+    def _init_langmem(self) -> None:
+        try:
+            from langmem import create_manage_memory_tool, create_search_memory_tool  # noqa: F401
+            from langgraph.store.memory import InMemoryStore
+
+            self.store = InMemoryStore()
+            namespace = ("claim_support_memories", "{user_id}")
+            self.manage_memory_tool = create_manage_memory_tool(namespace, store=self.store)
+            self.search_memory_tool = create_search_memory_tool(namespace, store=self.store)
+            self.langmem_available = True
+        except Exception:
+            self.langmem_available = False
+
+    def search(self, user_id: str, query: str, limit: int = 4) -> str:
+        langmem_lines = self._search_langmem_store(user_id, query, limit)
+        query_terms = set(tokenize(query))
+        with db() as conn:
+            rows = conn.execute(
+                """
+                SELECT kind, content, metadata_json, created_at
+                FROM memories
+                WHERE user_id = ?
+                ORDER BY created_at DESC
+                LIMIT 50
+                """,
+                (user_id,),
+            ).fetchall()
+
+        scored = []
+        for row in rows:
+            content = row["content"]
+            terms = set(tokenize(content))
+            score = len(query_terms & terms)
+            if score > 0:
+                scored.append((score, row))
+
+        scored.sort(key=lambda item: item[0], reverse=True)
+        selected = [row for _, row in scored[:limit]]
+        if not selected:
+            selected = rows[: min(limit, len(rows))]
+
+        if not selected and not langmem_lines:
+            return "No prior memory for this user."
+
+        lines = []
+        lines.extend(langmem_lines)
+        for row in selected:
+            lines.append(f"- {row['kind']}: {row['content']}")
+        return "\n".join(lines)
+
+    def save_interaction(
+        self,
+        user_id: str,
+        query: str,
+        answer: str,
+        critique: dict[str, Any],
+        sources: list[dict[str, Any]],
+    ) -> None:
+        decision = self._extract_decision(answer)
+        content = (
+            f"Scenario: {query} | Decision: {decision or 'unknown'} | "
+            f"Confidence: {critique.get('confidence', 0.0):.2f}"
+        )
+        metadata = {
+            "decision": decision,
+            "self_rag": {
+                "isrel": critique.get("isrel"),
+                "issup": critique.get("issup"),
+                "isuse": critique.get("isuse"),
+            },
+            "source_names": sorted({s.get("source_name", "unknown") for s in sources}),
+        }
+        self._insert_memory(user_id, "claim_interaction", content, metadata)
+        self._put_langmem_store(user_id, content, metadata)
+
+    def _search_langmem_store(self, user_id: str, query: str, limit: int) -> list[str]:
+        if not self.store:
+            return []
+        try:
+            items = self.store.search(("claim_support_memories", user_id), query=query, limit=limit)
+            lines = []
+            for item in items:
+                value = item.value or {}
+                content = value.get("content")
+                if content:
+                    lines.append(f"- langmem: {content}")
+            return lines
+        except Exception:
+            return []
+
+    def _put_langmem_store(self, user_id: str, content: str, metadata: dict[str, Any]) -> None:
+        if not self.store:
+            return
+        try:
+            self.store.put(
+                ("claim_support_memories", user_id),
+                key=new_id("memory"),
+                value={"content": content, "metadata": metadata},
+            )
+        except Exception:
+            return
+
+    def _insert_memory(self, user_id: str, kind: str, content: str, metadata: dict[str, Any]) -> None:
+        normalized = normalize_text(content)
+        with db() as conn:
+            existing = conn.execute(
+                """
+                SELECT memory_id FROM memories
+                WHERE user_id = ? AND kind = ? AND content = ?
+                """,
+                (user_id, kind, content),
+            ).fetchone()
+            if existing:
+                return
+            similar = conn.execute(
+                """
+                SELECT memory_id FROM memories
+                WHERE user_id = ? AND kind = ? AND lower(content) = ?
+                """,
+                (user_id, kind, normalized),
+            ).fetchone()
+            if similar:
+                return
+            conn.execute(
+                """
+                INSERT INTO memories(memory_id, user_id, kind, content, metadata_json)
+                VALUES (?, ?, ?, ?, ?)
+                """,
+                (new_id("memory"), user_id, kind, content, json.dumps(metadata, ensure_ascii=True)),
+            )
+
+    def _extract_decision(self, answer: str) -> str | None:
+        match = re.search(r"Decision:\s*(.+)", answer, flags=re.IGNORECASE)
+        if not match:
+            return None
+        return match.group(1).strip().rstrip(".")

data/eval/golden_claim_scenarios.jsonl

@@ -0,0 +1,14 @@
+{"id":"auto_deer_comprehensive","query":"I hit a deer and my car is damaged. Which part of my auto insurance covers this, and is it likely covered?","expected_decision":"Likely covered","must_have_sources":true,"reference":"Animal strikes are listed under comprehensive auto coverage in the coverage matrix, so this is likely covered if comprehensive coverage is active and the deductible applies."}
+{"id":"auto_stolen_car_comprehensive","query":"My car was stolen from my driveway. I have a police report. Is this likely covered by auto insurance?","expected_decision":"Likely covered","acceptable_decisions":["Likely covered","Needs human review"],"must_have_sources":true,"reference":"Vehicle theft is listed under comprehensive auto coverage. The claim still needs policy verification, deductible review, and theft documentation such as a police report."}
+{"id":"auto_hail_damage_comprehensive","query":"A hailstorm damaged my parked car. I have photos and a repair estimate. Will insurance likely pay?","expected_decision":"Likely covered","acceptable_decisions":["Likely covered","Needs human review"],"must_have_sources":true,"reference":"Hail damage to a car is listed under comprehensive auto coverage. Payment depends on active comprehensive coverage, deductible, photos, and repair estimate."}
+{"id":"auto_rental_after_accident","query":"My car is in the shop after an accident and repairs will take three weeks. Will insurance pay for my rental car?","expected_decision":"Needs human review","must_have_sources":true,"reference":"Rental reimbursement depends on whether the insured purchased rental coverage and who was at fault. The user should confirm daily and maximum rental limits and keep receipts."}
+{"id":"basement_flood_heavy_rain","query":"My basement flooded after heavy rain and water came through a foundation crack. Will homeowners insurance cover it?","expected_decision":"Likely not covered","acceptable_decisions":["Likely not covered","Needs human review"],"must_have_sources":true,"reference":"Rainwater entering through a foundation crack is likely flood or surface water, which standard homeowners policies exclude unless separate flood coverage applies."}
+{"id":"basement_sump_backup","query":"My basement flooded because the sump pump failed during a storm. I am not sure if I bought sewer or drain backup coverage. Is this covered?","expected_decision":"Needs human review","must_have_sources":true,"reference":"Sump pump or drain backup may require a sewer/drain backup endorsement. Coverage depends on whether that endorsement exists and the source of water."}
+{"id":"pipe_burst_homeowners","query":"A pipe burst in my home and caused water damage to the floor. I documented the damage and called a plumber. Is this likely covered?","expected_decision":"Likely covered","acceptable_decisions":["Likely covered","Needs human review"],"must_have_sources":true,"reference":"Water damage from a burst pipe is typically covered under the standard plumbing water damage peril, subject to policy terms, documentation, and deductible."}
+{"id":"earthquake_standard_policy","query":"Will earthquake damage be covered under a standard homeowners policy?","expected_decision":"Likely not covered","acceptable_decisions":["Likely not covered","Needs human review"],"must_have_sources":true,"reference":"Earthquake is listed as a major exclusion and normally requires a separate earthquake endorsement or policy."}
+{"id":"jewelry_stolen_hotel","query":"My diamond engagement ring was stolen from a hotel room. I have a police report, but the ring is worth more than my normal renters policy limit. Is it covered?","expected_decision":"Needs human review","must_have_sources":true,"reference":"Renters insurance may cover theft, but jewelry often has a sublimit such as $1,500 unless a scheduled personal property endorsement was purchased."}
+{"id":"life_contestability_period","query":"My husband died eight months after buying a life insurance policy and the insurer is investigating. Is that normal and will the claim be paid?","expected_decision":"Needs human review","must_have_sources":true,"reference":"Life policies commonly have a two-year contestability period. Investigation is normal during that period, and payment depends on whether there was material misrepresentation or other exclusion."}
+{"id":"bad_faith_long_investigation","query":"My insurer has been investigating my property claim for six months with no decision after I submitted proof of loss. What should happen?","expected_decision":"Needs human review","must_have_sources":true,"reference":"Most states require timely claim handling after complete proof of loss. A six-month delay may require escalation, written demand, DOI complaint, or legal review."}
+{"id":"inflated_repair_bill_red_flag","query":"The contractor invoice is much higher than the visible damage in my photos. Should this claim be approved now?","expected_decision":"Needs human review","must_have_sources":true,"reference":"An inflated repair bill or mismatch between claimed damage and photos is a fraud red flag. The file should be reviewed before approval and may require SIU or adjuster investigation."}
+{"id":"guaranty_property_casualty_limit","query":"My property casualty claim is $600,000, but my insurer became insolvent. How much may the state guaranty fund protect?","expected_decision":"Needs human review","must_have_sources":true,"reference":"The knowledge base lists typical property/casualty guaranty fund protection around $300,000 to $500,000 per claim, but exact limits vary by state and require review."}
+{"id":"total_loss_gap_insurance","query":"My insurer declared my car a total loss and offered $12,000, but I still owe $16,000 on the loan. Will insurance pay the full loan balance?","expected_decision":"Needs human review","must_have_sources":true,"reference":"A total loss settlement is usually based on actual cash value. Gap insurance may cover the difference between ACV and the remaining loan balance if purchased."}

docker-compose.yml

@@ -0,0 +1,38 @@
+services:
+  api:
+    build: .
+    container_name: insurance-copilot-api
+    ports:
+      - "8000:8000"
+    env_file:
+      - .env
+    environment:
+      QDRANT_URL: http://qdrant:6333
+      QDRANT_API_KEY: ""
+      APP_HOST: 0.0.0.0
+      APP_PORT: 8000
+      DOCUMENT_DIR: /app/data
+      UPLOAD_DIR: /app/data/uploads
+      SQLITE_PATH: /app/data/copilot.db
+      BM25_INDEX_PATH: /app/data/bm25_index.json
+      ENABLE_QUERY_REWRITE: "true"
+    depends_on:
+      - qdrant
+    volumes:
+      - ./data:/app/data
+      - huggingface_cache:/app/.cache/huggingface
+    restart: unless-stopped
+
+  qdrant:
+    image: qdrant/qdrant:latest
+    container_name: insurance-copilot-qdrant
+    ports:
+      - "6333:6333"
+      - "6334:6334"
+    volumes:
+      - qdrant_storage:/qdrant/storage
+    restart: unless-stopped
+
+volumes:
+  qdrant_storage:
+  huggingface_cache:

frontend/assets/app.js

@@ -0,0 +1,83 @@
+const messages = document.getElementById("messages");
+const form = document.getElementById("chatForm");
+const input = document.getElementById("queryInput");
+const sendBtn = document.getElementById("sendBtn");
+const sourcesPanel = document.getElementById("sourcesPanel");
+const sourcesBox = document.getElementById("sources");
+const latency = document.getElementById("latency");
+const sourceCount = document.getElementById("sourceCount");
+
+function appendMessage(role, text) {
+  const article = document.createElement("article");
+  article.className = `message ${role}`;
+  const bubble = document.createElement("div");
+  bubble.className = "bubble";
+  bubble.textContent = text;
+  article.appendChild(bubble);
+  messages.appendChild(article);
+  messages.scrollTop = messages.scrollHeight;
+  return bubble;
+}
+
+function renderSources(data) {
+  const sources = data.sources || [];
+  latency.textContent = `Latency: ${Math.round(data.latency_ms || 0)} ms`;
+  sourceCount.textContent = `Sources: ${sources.length}`;
+  sourcesPanel.hidden = false;
+
+  if (!sources.length) {
+    sourcesBox.innerHTML = "<p class=\"empty-source\">No source was returned for this answer.</p>";
+    return;
+  }
+
+  sourcesBox.innerHTML = sources.map((source, index) => {
+    const page = source.page === undefined || source.page === null ? "" : ` · page ${Number(source.page) + 1}`;
+    const score = source.score ? ` · score ${Number(source.score).toFixed(2)}` : "";
+    return `
+      <article class="source">
+        <strong>${index + 1}. ${escapeHtml(source.source_name || "unknown")}${page}${score}</strong>
+        <p>${escapeHtml((source.text || "").slice(0, 360))}</p>
+      </article>
+    `;
+  }).join("");
+}
+
+async function ask(query) {
+  appendMessage("user", query);
+  const pending = appendMessage("assistant", "Thinking...");
+  sendBtn.disabled = true;
+
+  try {
+    const response = await fetch("/api/query", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ query }),
+    });
+    const data = await response.json();
+    if (!response.ok) throw new Error(data.detail || "Request failed");
+    pending.textContent = data.answer || "No answer returned.";
+    renderSources(data);
+  } catch (error) {
+    pending.textContent = error.message;
+  } finally {
+    sendBtn.disabled = false;
+    input.focus();
+  }
+}
+
+function escapeHtml(value) {
+  return String(value)
+    .replaceAll("&", "&amp;")
+    .replaceAll("<", "&lt;")
+    .replaceAll(">", "&gt;")
+    .replaceAll('"', "&quot;")
+    .replaceAll("'", "&#039;");
+}
+
+form.addEventListener("submit", (event) => {
+  event.preventDefault();
+  const query = input.value.trim();
+  if (!query) return;
+  input.value = "";
+  ask(query);
+});

frontend/assets/styles.css

@@ -0,0 +1,224 @@
+:root {
+  --bg: #071015;
+  --panel: rgba(13, 27, 35, 0.88);
+  --panel-soft: rgba(18, 38, 48, 0.78);
+  --line: rgba(103, 232, 221, 0.22);
+  --text: #ecfbff;
+  --muted: #91aab2;
+  --cyan: #67e8dd;
+  --green: #a7f3c4;
+}
+
+* {
+  box-sizing: border-box;
+}
+
+body {
+  margin: 0;
+  min-height: 100vh;
+  color: var(--text);
+  font-family: Inter, ui-sans-serif, system-ui, -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif;
+  background:
+    linear-gradient(rgba(103, 232, 221, 0.035) 1px, transparent 1px),
+    linear-gradient(90deg, rgba(103, 232, 221, 0.035) 1px, transparent 1px),
+    radial-gradient(circle at 50% 0%, rgba(103, 232, 221, 0.14), transparent 34rem),
+    var(--bg);
+  background-size: 36px 36px, 36px 36px, auto, auto;
+}
+
+button,
+input {
+  font: inherit;
+}
+
+.chat-shell {
+  min-height: 100vh;
+  display: grid;
+  place-items: center;
+  padding: 28px;
+}
+
+.chat-card {
+  width: min(980px, 100%);
+  height: min(820px, calc(100vh - 56px));
+  display: grid;
+  grid-template-rows: auto 1fr auto auto;
+  border: 1px solid var(--line);
+  background: var(--panel);
+  box-shadow: 0 24px 90px rgba(0, 0, 0, 0.45);
+}
+
+.chat-header {
+  display: flex;
+  justify-content: space-between;
+  align-items: center;
+  gap: 16px;
+  padding: 22px;
+  border-bottom: 1px solid var(--line);
+}
+
+.chat-header p,
+.chat-header h1 {
+  margin: 0;
+}
+
+.chat-header p {
+  color: var(--cyan);
+  font-size: 12px;
+  font-weight: 800;
+  letter-spacing: 0.12em;
+  text-transform: uppercase;
+}
+
+.chat-header h1 {
+  margin-top: 4px;
+  font-size: clamp(24px, 4vw, 40px);
+  letter-spacing: 0;
+}
+
+.status {
+  border: 1px solid rgba(167, 243, 196, 0.34);
+  color: var(--green);
+  padding: 8px 10px;
+  white-space: nowrap;
+  background: rgba(167, 243, 196, 0.07);
+}
+
+.messages {
+  overflow: auto;
+  padding: 22px;
+}
+
+.message {
+  display: flex;
+  margin-bottom: 14px;
+}
+
+.message.user {
+  justify-content: flex-end;
+}
+
+.bubble {
+  max-width: min(720px, 88%);
+  padding: 14px 16px;
+  line-height: 1.6;
+  white-space: pre-wrap;
+  border: 1px solid rgba(145, 170, 178, 0.22);
+}
+
+.assistant .bubble {
+  background: rgba(9, 19, 25, 0.82);
+}
+
+.user .bubble {
+  background: rgba(103, 232, 221, 0.12);
+  border-color: rgba(103, 232, 221, 0.34);
+}
+
+.sources-panel {
+  border-top: 1px solid var(--line);
+  background: rgba(4, 12, 16, 0.48);
+  padding: 14px 22px;
+}
+
+.meta-row {
+  display: flex;
+  gap: 10px;
+  flex-wrap: wrap;
+  margin-bottom: 10px;
+}
+
+.meta-row span {
+  border: 1px solid var(--line);
+  color: #c9f7f4;
+  background: rgba(103, 232, 221, 0.07);
+  padding: 7px 9px;
+  font-size: 13px;
+}
+
+.sources {
+  display: grid;
+  gap: 8px;
+  max-height: 170px;
+  overflow: auto;
+}
+
+.source {
+  border: 1px solid rgba(145, 170, 178, 0.2);
+  background: var(--panel-soft);
+  padding: 10px;
+}
+
+.source strong {
+  color: var(--cyan);
+  font-size: 13px;
+}
+
+.source p,
+.empty-source {
+  margin: 7px 0 0;
+  color: #d5e8ec;
+  line-height: 1.45;
+  font-size: 14px;
+}
+
+.composer {
+  display: grid;
+  grid-template-columns: 1fr auto;
+  gap: 10px;
+  padding: 18px 22px 22px;
+  border-top: 1px solid var(--line);
+}
+
+.composer input {
+  min-height: 48px;
+  border: 1px solid rgba(145, 170, 178, 0.28);
+  background: #08161d;
+  color: var(--text);
+  outline: none;
+  padding: 0 14px;
+}
+
+.composer input:focus {
+  border-color: var(--cyan);
+  box-shadow: 0 0 0 3px rgba(103, 232, 221, 0.1);
+}
+
+.composer button {
+  min-width: 104px;
+  border: 1px solid rgba(103, 232, 221, 0.55);
+  color: var(--text);
+  background: linear-gradient(135deg, #0fb9b1, #176e84);
+  cursor: pointer;
+  font-weight: 800;
+}
+
+.composer button:disabled {
+  opacity: 0.65;
+  cursor: wait;
+}
+
+@media (max-width: 640px) {
+  .chat-shell {
+    padding: 0;
+  }
+
+  .chat-card {
+    min-height: 100vh;
+    height: 100vh;
+    border: 0;
+  }
+
+  .chat-header {
+    align-items: flex-start;
+    flex-direction: column;
+  }
+
+  .composer {
+    grid-template-columns: 1fr;
+  }
+
+  .composer button {
+    min-height: 46px;
+  }
+}

frontend/index.html

@@ -0,0 +1,50 @@
+<!doctype html>
+<html lang="en">
+  <head>
+    <meta charset="utf-8" />
+    <meta name="viewport" content="width=device-width, initial-scale=1" />
+    <title>Insurance Copilot</title>
+    <link rel="stylesheet" href="/assets/styles.css" />
+  </head>
+  <body>
+    <main class="chat-shell">
+      <section class="chat-card">
+        <header class="chat-header">
+          <div>
+            <p>Insurance Copilot</p>
+            <h1>Ask about insurance</h1>
+          </div>
+          <span class="status">RAG ready</span>
+        </header>
+
+        <div id="messages" class="messages">
+          <article class="message assistant">
+            <div class="bubble">
+              Describe your claim scenario and I will triage whether it looks likely covered, likely not covered, or needs human review.
+            </div>
+          </article>
+        </div>
+
+        <section id="sourcesPanel" class="sources-panel" hidden>
+          <div class="meta-row">
+            <span id="latency">Latency: --</span>
+            <span id="sourceCount">Sources: 0</span>
+          </div>
+          <div id="sources" class="sources"></div>
+        </section>
+
+        <form id="chatForm" class="composer">
+          <input
+            id="queryInput"
+            type="text"
+            autocomplete="off"
+            placeholder="Describe what happened in your claim..."
+          />
+          <button id="sendBtn" type="submit">Send</button>
+        </form>
+      </section>
+    </main>
+
+    <script src="/assets/app.js"></script>
+  </body>
+</html>

requirements-docker.txt

@@ -0,0 +1,22 @@
+fastapi>=0.115.0
+uvicorn[standard]>=0.30.0
+python-dotenv>=1.0.1
+pydantic>=2.8.0
+pydantic-settings>=2.4.0
+python-multipart>=0.0.9
+qdrant-client>=1.11.0
+rank-bm25>=0.2.2
+flashrank>=0.2.10
+langchain==1.2.18
+langchain-core==1.4.0
+langchain-community==0.4.1
+langchain-groq==1.1.2
+langchain-huggingface==1.2.2
+langchain-text-splitters==1.1.2
+langgraph==1.1.10
+langgraph-prebuilt==1.0.13
+langmem==0.0.30
+sentence-transformers>=3.0.0
+pypdf>=5.0.0
+numpy>=1.26.0
+orjson>=3.10.0

scripts/generate_sample_claim_pdf.py

@@ -0,0 +1,315 @@
+from pathlib import Path
+from textwrap import wrap
+
+
+PAGE_WIDTH = 612
+PAGE_HEIGHT = 792
+LEFT = 50
+TOP = 760
+LINE_HEIGHT = 14
+MAX_LINES_PER_PAGE = 46
+WRAP_WIDTH = 92
+PAGE_COUNT = 50
+
+
+AGENT_TOPICS = [
+    (
+        "Claim Support Agent Mission",
+        "The insurance claim support AI agent helps customers and support adjusters reason through "
+        "claim scenarios. The agent should not merely define insurance terms. It should ask what "
+        "happened, identify the likely claim type, retrieve relevant policy and procedure evidence, "
+        "consider prior user context from memory, and explain whether the claim appears likely covered, "
+        "likely not covered, or uncertain. The agent must avoid final binding coverage decisions unless "
+        "the policy and claim file clearly support the conclusion. When evidence is incomplete, the "
+        "agent should list missing information and recommend human review.",
+    ),
+    (
+        "Scenario Based Claim Reasoning",
+        "A scenario-based response begins by restating the facts that matter: cause of loss, date of "
+        "loss, property or vehicle involved, policy type, available evidence, mitigation steps, and any "
+        "red flags. The agent then maps the scenario to a claim category such as water damage, theft, "
+        "fire, auto collision, liability, flood, storm, or personal property. It should compare the "
+        "scenario with retrieved claim rules and explain the likely outcome as likely covered, likely "
+        "not covered, or needs review. The agent should include citations to retrieved sources and "
+        "should clearly separate evidence-based conclusions from assumptions.",
+    ),
+    (
+        "Memory Usage With LangMem",
+        "The agent should use memory to personalize support without exposing sensitive information. "
+        "Useful memory includes the customer's previous claim type, preferred contact method, recurring "
+        "missing documents, prior escalation outcomes, and approved resolution summaries. Memory should "
+        "not replace retrieval from policy documents. If memory says the customer previously had a water "
+        "claim with missing mitigation invoices, the agent may remind the user that mitigation evidence "
+        "was important before, but it must still retrieve current policy guidance before making a coverage "
+        "recommendation. Approved human resolutions are stronger memory than unreviewed draft answers.",
+    ),
+    (
+        "Tool Calling Policy",
+        "The agent can call tools when the answer depends on external operational data. A claim lookup "
+        "tool should be used to check claim status, date of loss, assigned adjuster, missing documents, "
+        "and previous notes. A plan lookup tool should be used to check policy limits, endorsements, "
+        "deductibles, covered property, and exclusions. An open ticket load tool should be used to decide "
+        "whether to route the matter to a human support queue. The agent should state which tool would be "
+        "useful and why when a tool result is needed but unavailable.",
+    ),
+    (
+        "Coverage Decision Labels",
+        "The agent should use cautious labels. 'Likely covered' means the retrieved evidence supports "
+        "coverage and no obvious exclusion appears in the provided scenario. 'Likely not covered' means "
+        "the retrieved evidence points to an exclusion or unmet condition. 'Needs human review' means "
+        "evidence is missing, policy language is ambiguous, the scenario is high risk, or a tool lookup is "
+        "required. These labels are support recommendations, not final legal or contractual decisions.",
+    ),
+    (
+        "Water Damage Scenario Rules",
+        "Water damage scenarios require attention to cause and timing. Sudden and accidental discharge "
+        "from a burst pipe may be treated more favorably than seepage, repeated leakage, mold, or poor "
+        "maintenance. Required evidence often includes notice of loss, photos, plumber report, repair "
+        "estimate, mitigation invoice, and proof that the policy was active. If the customer says water "
+        "leaked slowly for months, the agent should mark the claim as likely not covered or needs human "
+        "review because gradual leakage and maintenance issues may be excluded.",
+    ),
+    (
+        "Flood and Storm Scenario Rules",
+        "Flood scenarios should be separated from internal water damage. Heavy rain entering from surface "
+        "water, storm surge, overflowing bodies of water, or groundwater may require separate flood coverage. "
+        "Wind or hail damage may be handled differently from flood damage. If a customer says the basement "
+        "flooded after heavy rain, the agent should not promise coverage under a standard property policy. "
+        "It should recommend plan lookup for flood endorsement or separate flood policy and request photos, "
+        "weather date, water entry point, and mitigation records.",
+    ),
+    (
+        "Theft Scenario Rules",
+        "Theft scenarios require a police report, list of stolen items, proof of ownership, receipts, serial "
+        "numbers, and photos or security footage when available. If property was stolen from an unlocked car, "
+        "the agent should check whether the property policy or auto policy applies and whether limitations "
+        "or exclusions apply. High-value items such as jewelry, electronics, collectibles, firearms, and art "
+        "may have sublimits or scheduled property requirements. Missing police report or ownership proof "
+        "should trigger human review.",
+    ),
+    (
+        "Fire and Smoke Scenario Rules",
+        "Fire and smoke scenarios require fire department report, photos, repair estimate, damaged-property "
+        "inventory, proof of ownership for valuable items, and temporary housing receipts if additional living "
+        "expense is claimed. Suspected arson, inconsistent timelines, missing fire report, or unusually high "
+        "claimed values should trigger escalation. Smoke damage should be described separately from direct "
+        "fire damage because cleaning and odor remediation may require different documentation.",
+    ),
+    (
+        "Auto Collision Scenario Rules",
+        "Auto collision scenarios require accident date and location, driver details, vehicle photos, repair "
+        "estimate, registration, insurance information for involved parties, witness details, and police report "
+        "when available. If there is no police report, the claim may still proceed but needs stronger supporting "
+        "evidence. Liability depends on statements, traffic rules, point of impact, photos, and police report. "
+        "Total loss review requires actual cash value, title status, lienholder details, and state rules.",
+    ),
+    (
+        "Liability Scenario Rules",
+        "Liability scenarios involve allegations that the insured caused bodily injury or property damage to "
+        "another person. The agent should not admit fault. It should request incident description, claimant "
+        "contact details, photos, witness statements, medical bills for bodily injury, property repair invoices, "
+        "and any demand letter. Bodily injury, attorney involvement, policy limit demand, or legal threat should "
+        "trigger human review and possible specialist routing.",
+    ),
+    (
+        "Human Review Triggers",
+        "Human review is required when evidence is missing, documents appear altered, claim facts conflict, "
+        "policy language is unclear, the user asks for a denial or appeal decision, legal threats are present, "
+        "bodily injury is involved, fraud indicators appear, or high-value property is claimed without proof. "
+        "The agent should explain the reason for escalation in plain language and list the next best action.",
+    ),
+    (
+        "Fraud and Risk Signals",
+        "Risk signals include loss shortly after policy inception, duplicate receipts, altered invoices, refusal "
+        "to permit inspection, repair estimates that do not match photos, multiple similar claims, staged accident "
+        "concerns, missing ownership proof, inconsistent timelines, or pressure for immediate payment. Risk signals "
+        "do not prove fraud, but they justify additional documentation and senior review.",
+    ),
+    (
+        "Recommended Answer Format",
+        "For claim scenarios, the recommended answer format is: decision label, short reasoning, needed evidence, "
+        "tool or memory action, and source citation. Example labels are likely covered, likely not covered, and "
+        "needs human review. The agent should avoid long legal explanations unless requested. It should be concise, "
+        "helpful, and transparent about uncertainty.",
+    ),
+]
+
+
+SCENARIOS = [
+    (
+        "My basement flooded after heavy rain and water came through the floor drain. Will insurance pay?",
+        "Needs human review. This may involve flood, surface water, sewer backup, or storm water conditions. "
+        "The agent should call plan lookup to check flood or sewer backup endorsement and request photos, "
+        "water entry point, weather date, and mitigation records.",
+    ),
+    (
+        "A pipe suddenly burst in my kitchen while I was away for work. I have photos and a plumber report.",
+        "Likely covered if the policy covers sudden and accidental water discharge and no exclusion applies. "
+        "The agent should request mitigation invoices, repair estimates, date of loss, and policy verification.",
+    ),
+    (
+        "My bathroom leaked slowly for months and now there is mold behind the wall.",
+        "Likely not covered or needs human review because gradual leakage, mold, and maintenance issues may be "
+        "excluded. The agent should retrieve water damage exclusions and request contractor findings.",
+    ),
+    (
+        "My laptop and camera were stolen from my unlocked car.",
+        "Needs human review. The agent should check whether property or auto coverage applies, ask for a police "
+        "report, proof of ownership, receipts, serial numbers, and review sublimits for electronics.",
+    ),
+    (
+        "A small kitchen fire damaged cabinets and smoke damaged furniture.",
+        "Likely covered if fire is a covered peril and no exclusion applies. Required evidence includes fire "
+        "report, photos, repair estimate, smoke remediation estimate, inventory, and receipts.",
+    ),
+    (
+        "I hit another car but there is no police report. Can I still claim?",
+        "Needs review but may proceed with other evidence. The agent should request photos, driver information, "
+        "repair estimate, witness details, accident location, and statement of events.",
+    ),
+    (
+        "A guest slipped on my stairs and is asking me to pay medical bills.",
+        "Needs human review. Bodily injury liability matters should be escalated. The agent should request incident "
+        "description, photos, witness statements, medical bills, and any demand letter.",
+    ),
+    (
+        "My roof was damaged by hail during a storm.",
+        "Potentially covered depending on policy and evidence. The agent should request photos, contractor estimate, "
+        "weather date, inspection notes, and plan lookup for wind or hail coverage and deductible.",
+    ),
+    (
+        "My jewelry was stolen but I do not have receipts.",
+        "Needs human review. Jewelry may have sublimits or scheduled property requirements. The agent should request "
+        "police report, photos, appraisal, bank records, or other proof of ownership.",
+    ),
+    (
+        "The repair invoice looks higher than the visible damage in photos.",
+        "Needs human review because invoice and photo mismatch is a risk signal. The agent should request itemized "
+        "estimate, inspection, and senior adjuster review.",
+    ),
+]
+
+
+def escape_pdf_text(text: str) -> str:
+    return text.replace("\\", "\\\\").replace("(", "\\(").replace(")", "\\)")
+
+
+def paragraph_lines(text: str) -> list[str]:
+    lines: list[str] = []
+    for paragraph in text.split("\n"):
+        if not paragraph.strip():
+            lines.append("")
+            continue
+        lines.extend(wrap(paragraph, width=WRAP_WIDTH))
+    return lines
+
+
+def make_page(page_number: int) -> list[str]:
+    topic = AGENT_TOPICS[(page_number - 1) % len(AGENT_TOPICS)]
+    related = AGENT_TOPICS[page_number % len(AGENT_TOPICS)]
+    scenario = SCENARIOS[(page_number - 1) % len(SCENARIOS)]
+    second_scenario = SCENARIOS[page_number % len(SCENARIOS)]
+
+    body = (
+        f"Insurance Claim Support AI Agent with LangMem and RAG - Page {page_number:02d}\n\n"
+        f"{topic[0]}\n"
+        f"{topic[1]}\n\n"
+        f"RAG guidance: Retrieve policy rules, claim procedures, and prior approved resolutions before "
+        f"answering. If retrieved evidence is weak, say that evidence is insufficient. Cite retrieved "
+        f"sources. Do not invent policy terms, claim status, payment approval, or denial decisions.\n\n"
+        f"Memory guidance: Use LangMem-style memory for prior user interactions, repeated missing documents, "
+        f"preferred contact method, and approved claim resolutions. Memory may personalize the answer, but "
+        f"policy retrieval and tool results should control coverage reasoning.\n\n"
+        f"Tool guidance: Use claim lookup for claim status and missing documents. Use plan lookup for coverage, "
+        f"limits, deductibles, endorsements, and exclusions. Use ticket load or escalation tools when the case "
+        f"requires human review or specialist routing.\n\n"
+        f"Scenario example: {scenario[0]}\n"
+        f"Expected agent response: {scenario[1]}\n\n"
+        f"Additional scenario: {second_scenario[0]}\n"
+        f"Expected agent response: {second_scenario[1]}\n\n"
+        f"Related topic: {related[0]}. {related[1]}\n\n"
+        f"Recommended response structure: Decision label, explanation, missing evidence, recommended tool call, "
+        f"human review decision, and source citation."
+    )
+
+    lines = paragraph_lines(body)
+    if len(lines) > MAX_LINES_PER_PAGE:
+        return lines[:MAX_LINES_PER_PAGE]
+    return lines + [""] * (MAX_LINES_PER_PAGE - len(lines))
+
+
+def page_stream(lines: list[str]) -> bytes:
+    content_lines = ["BT", "/F1 10 Tf", f"{LEFT} {TOP} Td", f"{LINE_HEIGHT} TL"]
+    for index, line in enumerate(lines):
+        escaped = escape_pdf_text(line)
+        if index == 0:
+            content_lines.append(f"({escaped}) Tj")
+        else:
+            content_lines.append(f"T* ({escaped}) Tj")
+    content_lines.append("ET")
+    return "\n".join(content_lines).encode("latin-1", errors="replace")
+
+
+def build_pdf(pages: list[list[str]]) -> bytes:
+    objects: list[bytes] = []
+    pages_id = 2
+    font_id = 3
+    page_ids: list[int] = []
+    content_ids: list[int] = []
+
+    next_id = 4
+    for _ in pages:
+        page_ids.append(next_id)
+        next_id += 1
+        content_ids.append(next_id)
+        next_id += 1
+
+    kids = " ".join(f"{page_id} 0 R" for page_id in page_ids)
+    objects.append(f"<< /Type /Catalog /Pages {pages_id} 0 R >>".encode("ascii"))
+    objects.append(f"<< /Type /Pages /Kids [{kids}] /Count {len(page_ids)} >>".encode("ascii"))
+    objects.append(b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>")
+
+    for page_id, content_id, page_lines in zip(page_ids, content_ids, pages):
+        objects.append(
+            (
+                f"<< /Type /Page /Parent {pages_id} 0 R /MediaBox [0 0 {PAGE_WIDTH} {PAGE_HEIGHT}] "
+                f"/Resources << /Font << /F1 {font_id} 0 R >> >> /Contents {content_id} 0 R >>"
+            ).encode("ascii")
+        )
+        content = page_stream(page_lines)
+        objects.append(
+            b"<< /Length " + str(len(content)).encode("ascii") + b" >>\nstream\n" + content + b"\nendstream"
+        )
+
+    pdf = bytearray(b"%PDF-1.4\n")
+    offsets = [0]
+    for obj_id, body in enumerate(objects, start=1):
+        offsets.append(len(pdf))
+        pdf.extend(f"{obj_id} 0 obj\n".encode("ascii"))
+        pdf.extend(body)
+        pdf.extend(b"\nendobj\n")
+
+    xref_start = len(pdf)
+    pdf.extend(f"xref\n0 {len(objects) + 1}\n".encode("ascii"))
+    pdf.extend(b"0000000000 65535 f \n")
+    for offset in offsets[1:]:
+        pdf.extend(f"{offset:010d} 00000 n \n".encode("ascii"))
+    pdf.extend(
+        (
+            f"trailer\n<< /Size {len(objects) + 1} /Root 1 0 R >>\n"
+            f"startxref\n{xref_start}\n%%EOF\n"
+        ).encode("ascii")
+    )
+    return bytes(pdf)
+
+
+def main() -> None:
+    pages = [make_page(page_number) for page_number in range(1, PAGE_COUNT + 1)]
+    output = Path("data") / "sample_insurance_claim_guide.pdf"
+    output.parent.mkdir(parents=True, exist_ok=True)
+    output.write_bytes(build_pdf(pages))
+    print(output)
+
+
+if __name__ == "__main__":
+    main()

scripts/run_eval.py

@@ -0,0 +1,108 @@
+import argparse
+import json
+import re
+import sys
+import time
+from pathlib import Path
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from app.db.sqlite import init_db
+from app.rag.graph import ClaimsRAGGraph
+from app.rag.ingestion import DocumentIngestionService
+from app.rag.qdrant_store import QdrantVectorStore
+
+
+DECISION_RE = re.compile(r"Decision:\s*(Likely covered|Likely not covered|Needs human review)", re.I)
+
+
+def load_jsonl(path: Path) -> list[dict[str, Any]]:
+    rows = []
+    for line in path.read_text(encoding="utf-8").splitlines():
+        if line.strip():
+            rows.append(json.loads(line))
+    return rows
+
+
+def extract_decision(answer: str) -> str:
+    match = DECISION_RE.search(answer)
+    if not match:
+        return "Unknown"
+    return match.group(1).capitalize().replace("Not", "not")
+
+
+def accepted_decisions(case: dict[str, Any]) -> set[str]:
+    values = case.get("acceptable_decisions") or [case["expected_decision"]]
+    return {str(v).lower() for v in values}
+
+
+def evaluate(dataset_path: Path, user_id: str, limit: int | None = None) -> dict[str, Any]:
+    init_db()
+    QdrantVectorStore().ensure_collections()
+    DocumentIngestionService().ingest_pdf_directory()
+    graph = ClaimsRAGGraph()
+
+    cases = load_jsonl(dataset_path)
+    if limit:
+        cases = cases[:limit]
+    results = []
+    for case in cases:
+        started = time.perf_counter()
+        state = graph.run(case["query"], user_id=user_id, use_cache=False)
+        latency_ms = round((time.perf_counter() - started) * 1000, 2)
+        answer = state.get("answer", "")
+        decision = extract_decision(answer)
+        sources = state.get("reranked_sources") or state.get("sources", [])
+        critique = state.get("self_rag", {})
+        decision_ok = decision.lower() in accepted_decisions(case)
+        sources_ok = bool(sources) if case.get("must_have_sources", True) else True
+        self_rag_ok = bool(critique.get("isrel")) and bool(critique.get("issup")) and bool(critique.get("isuse"))
+        passed = decision_ok and sources_ok and self_rag_ok
+        results.append(
+            {
+                "id": case["id"],
+                "expected": case["expected_decision"],
+                "decision": decision,
+                "decision_ok": decision_ok,
+                "sources": len(sources),
+                "sources_ok": sources_ok,
+                "self_rag": {
+                    "ISREL": critique.get("isrel"),
+                    "ISSUP": critique.get("issup"),
+                    "ISUSE": critique.get("isuse"),
+                },
+                "self_rag_ok": self_rag_ok,
+                "latency_ms": latency_ms,
+                "passed": passed,
+            }
+        )
+
+    total = len(results)
+    summary = {
+        "total": total,
+        "passed": sum(1 for r in results if r["passed"]),
+        "decision_accuracy": round(sum(1 for r in results if r["decision_ok"]) / total, 3) if total else 0,
+        "source_rate": round(sum(1 for r in results if r["sources_ok"]) / total, 3) if total else 0,
+        "self_rag_pass_rate": round(sum(1 for r in results if r["self_rag_ok"]) / total, 3) if total else 0,
+        "avg_latency_ms": round(sum(r["latency_ms"] for r in results) / total, 2) if total else 0,
+        "results": results,
+    }
+    return summary
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--dataset", default="data/eval/golden_claim_scenarios.jsonl")
+    parser.add_argument("--user-id", default="eval_user")
+    parser.add_argument("--limit", type=int, default=None)
+    args = parser.parse_args()
+
+    summary = evaluate(Path(args.dataset), args.user_id, args.limit)
+    print(json.dumps(summary, indent=2))
+
+
+if __name__ == "__main__":
+    main()

scripts/run_ragas_eval.py

@@ -0,0 +1,154 @@
+import argparse
+import asyncio
+import json
+import sys
+import time
+from pathlib import Path
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from datasets import Dataset
+from langchain_groq import ChatGroq
+from ragas import RunConfig, evaluate
+from ragas.embeddings import LangchainEmbeddingsWrapper
+from ragas.llms import LangchainLLMWrapper
+from ragas.metrics import (
+    answer_relevancy,
+    context_precision,
+    context_recall,
+    faithfulness,
+)
+
+from app.core.config import settings
+from app.db.sqlite import init_db
+from app.rag.embeddings import get_embedding_model
+from app.rag.graph import ClaimsRAGGraph
+from app.rag.ingestion import DocumentIngestionService
+from app.rag.qdrant_store import QdrantVectorStore
+
+
+def load_jsonl(path: Path) -> list[dict[str, Any]]:
+    rows = []
+    for line in path.read_text(encoding="utf-8").splitlines():
+        if line.strip():
+            rows.append(json.loads(line))
+    return rows
+
+
+def default_reference(case: dict[str, Any]) -> str:
+    if case.get("reference"):
+        return str(case["reference"])
+    decision = case["expected_decision"]
+    return (
+        f"Decision: {decision}. The answer should use the retrieved insurance claim "
+        "guidance to explain the coverage triage, identify missing evidence, and "
+        "recommend the next action without inventing unsupported policy terms."
+    )
+
+
+def build_eval_rows(dataset_path: Path, user_id: str, limit: int | None) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+    init_db()
+    QdrantVectorStore().ensure_collections()
+    DocumentIngestionService().ingest_pdf_directory()
+    graph = ClaimsRAGGraph()
+
+    cases = load_jsonl(dataset_path)
+    if limit:
+        cases = cases[:limit]
+
+    ragas_rows = []
+    run_rows = []
+    for case in cases:
+        started = time.perf_counter()
+        state = graph.run(case["query"], user_id=user_id, use_cache=False)
+        latency_ms = round((time.perf_counter() - started) * 1000, 2)
+        sources = state.get("reranked_sources") or state.get("sources", [])
+        contexts = [str(source.get("text", "")) for source in sources if source.get("text")]
+
+        ragas_rows.append(
+            {
+                "user_input": case["query"],
+                "response": state.get("answer", ""),
+                "retrieved_contexts": contexts,
+                "reference": default_reference(case),
+            }
+        )
+        run_rows.append(
+            {
+                "id": case["id"],
+                "expected_decision": case["expected_decision"],
+                "sources": len(contexts),
+                "latency_ms": latency_ms,
+            }
+        )
+    return ragas_rows, run_rows
+
+
+async def run_ragas(dataset_path: Path, user_id: str, limit: int | None) -> dict[str, Any]:
+    if not settings.groq_api_key:
+        raise RuntimeError("GROQ_API_KEY is required for RAGAS LLM-judge metrics.")
+
+    ragas_rows, run_rows = build_eval_rows(dataset_path, user_id, limit)
+    dataset = Dataset.from_list(ragas_rows)
+
+    judge_llm = ChatGroq(
+        model=settings.groq_model,
+        temperature=0,
+        max_retries=2,
+        api_key=settings.groq_api_key,
+    )
+    ragas_llm = LangchainLLMWrapper(judge_llm)
+    ragas_embeddings = LangchainEmbeddingsWrapper(get_embedding_model().model)
+
+    answer_relevancy.strictness = 1
+    metrics = [faithfulness, answer_relevancy, context_precision, context_recall]
+
+    result = evaluate(
+        dataset=dataset,
+        metrics=metrics,
+        llm=ragas_llm,
+        embeddings=ragas_embeddings,
+        run_config=RunConfig(timeout=180, max_workers=2, max_retries=2),
+    )
+
+    scores = result.to_pandas().to_dict(orient="records")
+    rows = []
+    for run_row, score_row in zip(run_rows, scores, strict=False):
+        rows.append({**run_row, "ragas": score_row})
+
+    metric_names = ["faithfulness", "answer_relevancy", "context_precision", "context_recall"]
+    summary = {
+        "total": len(rows),
+        "metrics": {},
+        "results": rows,
+    }
+    for metric in metric_names:
+        values = [
+            float(row["ragas"][metric])
+            for row in rows
+            if row["ragas"].get(metric) is not None and str(row["ragas"][metric]).lower() != "nan"
+        ]
+        summary["metrics"][metric] = round(sum(values) / len(values), 3) if values else None
+    return summary
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--dataset", default="data/eval/golden_claim_scenarios.jsonl")
+    parser.add_argument("--user-id", default="ragas_eval_user")
+    parser.add_argument("--limit", type=int, default=None)
+    parser.add_argument("--output", default=None)
+    args = parser.parse_args()
+
+    summary = asyncio.run(run_ragas(Path(args.dataset), args.user_id, args.limit))
+    text = json.dumps(summary, indent=2)
+    print(text)
+    if args.output:
+        Path(args.output).write_text(text + "\n", encoding="utf-8")
+
+
+if __name__ == "__main__":
+    main()