Upload 11 files

API_SPEC.md

@@ -0,0 +1,30 @@
+# Fuzzy LangChain API
+
+## GET /health
+Returns service status and number of ingested documents.
+
+## POST /ingest
+```json
+{
+  "replace": true,
+  "documents": [
+    {
+      "id": "doc-1",
+      "text": "The current access code is BRAVO-9.",
+      "title": "Current notice",
+      "source": "official",
+      "authority": 0.98,
+      "freshness": 1.0,
+      "reliability": 0.95,
+      "metadata": {}
+    }
+  ]
+}
+```
+
+## POST /query
+```json
+{"question":"What is the current access code?","k":4}
+```
+
+The response contains `answer`, `citations`, `confidence`, `margin`, `decision`, `reason`, and ranked `evidence`.

Dockerfile

@@ -0,0 +1,9 @@
+FROM python:3.12-slim
+WORKDIR /app
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+COPY src ./src
+COPY app.py .
+ENV PYTHONPATH=/app/src
+EXPOSE 7860
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Angshul Majumdar
+
+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.

README.md

@@ -1,3 +1,9 @@
----
-license: mit
----
+# Fuzzy LangChain — Hugging Face API
+
+Minimal deployable API for the Fuzzy LangChain RAG runtime. This package intentionally contains **only the serving surface and the minimum runtime needed by it**. Benchmarks, experiments, tests, and research documentation belong in the GitHub RAG repository.
+
+Endpoints: `GET /health`, `POST /ingest`, `POST /query`.
+
+The query path runs through the actual LangChain `BaseRetriever` implementation, fuzzy evidence assessment, optional retrieval expansion, grounded answer generation, citation return, and abstention under insufficient or ambiguous evidence.
+
+See `API_SPEC.md` for request/response examples.

app.py

@@ -0,0 +1 @@
+from fuzzy_langchain.api import app

requirements.txt

@@ -0,0 +1,5 @@
+fastapi>=0.110
+uvicorn>=0.27
+langchain-core>=1.4.7,<2
+numpy>=1.24
+scikit-learn>=1.3

src/fuzzy_langchain/__init__.py

@@ -0,0 +1,17 @@
+from .core import (
+    DocumentRecord,
+    FuzzyEvidence,
+    FuzzyEvidenceController,
+    FuzzyIndex,
+    FuzzyRAGEngine,
+    FuzzyRetrievalConfig,
+)
+
+__all__ = [
+    "DocumentRecord",
+    "FuzzyEvidence",
+    "FuzzyEvidenceController",
+    "FuzzyIndex",
+    "FuzzyRAGEngine",
+    "FuzzyRetrievalConfig",
+]

src/fuzzy_langchain/api.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from typing import Any
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel, Field
+
+from .core import DocumentRecord, FuzzyIndex, FuzzyRetrievalConfig
+from .rag import LangChainFuzzyRAG
+
+
+class IngestDocument(BaseModel):
+    id: str
+    text: str
+    title: str = ""
+    source: str = ""
+    authority: float = Field(0.5, ge=0.0, le=1.0)
+    freshness: float = Field(0.5, ge=0.0, le=1.0)
+    reliability: float = Field(0.5, ge=0.0, le=1.0)
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+
+class IngestRequest(BaseModel):
+    documents: list[IngestDocument]
+    replace: bool = True
+
+
+class QueryRequest(BaseModel):
+    question: str
+    k: int = Field(4, ge=1, le=20)
+
+
+def create_app() -> FastAPI:
+    app = FastAPI(title="Fuzzy LangChain RAG API", version="1.0.0")
+    holder: dict[str, Any] = {"index": FuzzyIndex(config=FuzzyRetrievalConfig())}
+
+    @app.get("/health")
+    def health():
+        return {"status": "ok", "documents": len(holder["index"].documents)}
+
+    @app.post("/ingest")
+    def ingest(req: IngestRequest):
+        docs: list[DocumentRecord] = []
+        for d in req.documents:
+            md = dict(d.metadata)
+            md.update({"authority": d.authority, "freshness": d.freshness, "reliability": d.reliability})
+            docs.append(DocumentRecord(d.id, d.text, d.title, d.source, md))
+        if req.replace:
+            holder["index"].fit(docs)
+        else:
+            holder["index"].add(docs)
+        return {"documents": len(holder["index"].documents)}
+
+    @app.post("/query")
+    def query(req: QueryRequest):
+        if not holder["index"].documents:
+            raise HTTPException(status_code=400, detail="No documents ingested")
+        rag = LangChainFuzzyRAG(holder["index"], k=req.k)
+        return rag.invoke(req.question)
+
+    return app
+
+
+app = create_app()

src/fuzzy_langchain/core.py

@@ -0,0 +1,307 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Iterable, Sequence
+import math
+import re
+
+import numpy as np
+from sklearn.feature_extraction.text import TfidfVectorizer
+from sklearn.metrics.pairwise import cosine_similarity
+
+
+@dataclass(frozen=True)
+class DocumentRecord:
+    id: str
+    text: str
+    title: str = ""
+    source: str = ""
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class FuzzyRetrievalConfig:
+    candidate_pool: int = 10
+    answer_threshold: float = 0.60
+    ambiguity_margin: float = 0.015
+    min_raw_relevance: float = 0.14
+    expand_threshold: float = 0.48
+    extended_pool: int = 16
+
+
+@dataclass(frozen=True)
+class FuzzyEvidence:
+    document: DocumentRecord
+    raw_relevance: float
+    fuzzy_score: float
+    authority: float
+    freshness: float
+    reliability: float
+    memberships: dict[str, float]
+
+
+class FuzzyEvidenceController:
+    """Small dependency-free Sugeno fuzzy controller for RAG evidence quality.
+
+    Inputs are normalized to [0, 1]: lexical/semantic relevance, source authority,
+    freshness, and source reliability. The rules intentionally make relevance a
+    gate: highly trusted but irrelevant evidence must not become high confidence.
+    """
+
+    @staticmethod
+    def _clip(x: float) -> float:
+        return max(0.0, min(1.0, float(x)))
+
+    @classmethod
+    def _left(cls, x: float, full_until: float, zero_at: float) -> float:
+        x = cls._clip(x)
+        if x <= full_until:
+            return 1.0
+        if x >= zero_at:
+            return 0.0
+        return (zero_at - x) / (zero_at - full_until)
+
+    @classmethod
+    def _right(cls, x: float, zero_until: float, full_at: float) -> float:
+        x = cls._clip(x)
+        if x <= zero_until:
+            return 0.0
+        if x >= full_at:
+            return 1.0
+        return (x - zero_until) / (full_at - zero_until)
+
+    @classmethod
+    def _triangle(cls, x: float, left: float, peak: float, right: float) -> float:
+        x = cls._clip(x)
+        if x <= left or x >= right:
+            return 0.0
+        if x == peak:
+            return 1.0
+        if x < peak:
+            return (x - left) / (peak - left)
+        return (right - x) / (right - peak)
+
+    def memberships(self, relevance: float, authority: float, freshness: float, reliability: float) -> dict[str, float]:
+        r = self._clip(relevance)
+        a = self._clip(authority)
+        f = self._clip(freshness)
+        q = self._clip(reliability)
+        return {
+            "rel_low": self._left(r, 0.05, 0.30),
+            "rel_med": self._triangle(r, 0.12, 0.42, 0.70),
+            "rel_high": self._right(r, 0.42, 0.72),
+            "auth_low": self._left(a, 0.0, 0.55),
+            "auth_med": self._triangle(a, 0.35, 0.62, 0.84),
+            "auth_high": self._right(a, 0.68, 0.90),
+            "fresh_low": self._left(f, 0.0, 0.42),
+            "fresh_med": self._triangle(f, 0.28, 0.58, 0.82),
+            "fresh_high": self._right(f, 0.68, 0.90),
+            "reliability_low": self._left(q, 0.0, 0.50),
+            "reliability_high": self._right(q, 0.58, 0.88),
+        }
+
+    def score(self, relevance: float, authority: float, freshness: float, reliability: float) -> tuple[float, dict[str, float]]:
+        m = self.memberships(relevance, authority, freshness, reliability)
+        # First-order Sugeno rules. Consequents are confidence-like evidence values.
+        rules: list[tuple[float, float]] = [
+            (min(m["rel_high"], m["auth_high"], m["fresh_high"], m["reliability_high"]), 0.98),
+            (min(m["rel_med"], m["auth_high"], m["fresh_high"], m["reliability_high"]), 0.90),
+            (min(m["rel_high"], m["auth_high"]), 0.92),
+            (min(m["rel_med"], m["auth_high"]), 0.84),
+            (min(m["rel_high"], m["auth_med"], m["fresh_high"]), 0.74),
+            (min(m["rel_med"], m["auth_med"]), 0.54),
+            (min(m["rel_high"], m["fresh_low"]), 0.28),
+            (min(m["rel_high"], m["auth_low"]), 0.24),
+            (min(m["rel_med"], m["auth_low"]), 0.20),
+            (min(m["rel_high"], m["reliability_low"]), 0.26),
+            (m["rel_low"], 0.06),
+        ]
+        denom = sum(w for w, _ in rules)
+        if denom <= 1e-12:
+            # No fuzzy rule fired strongly; relevance still gates the fallback.
+            base = 0.10 + 0.55 * self._clip(relevance)
+        else:
+            base = sum(w * z for w, z in rules) / denom
+        # Relevance is already a fuzzy antecedent and is also enforced by the
+        # hard minimum-relevance guard in the final decision.
+        value = self._clip(base)
+        return value, m
+
+
+class FuzzyIndex:
+    def __init__(self, documents: Sequence[DocumentRecord] | None = None, config: FuzzyRetrievalConfig | None = None):
+        self.config = config or FuzzyRetrievalConfig()
+        self.controller = FuzzyEvidenceController()
+        self.documents: list[DocumentRecord] = []
+        self._vectorizer: TfidfVectorizer | None = None
+        self._matrix = None
+        if documents:
+            self.fit(documents)
+
+    @staticmethod
+    def _trust(doc: DocumentRecord) -> tuple[float, float, float]:
+        md = doc.metadata or {}
+        authority = float(md.get("authority", 0.50))
+        freshness = float(md.get("freshness", 0.50))
+        if "reliability" in md:
+            reliability = float(md["reliability"])
+        elif "reliability_alpha" in md or "reliability_beta" in md:
+            a = float(md.get("reliability_alpha", 1.0))
+            b = float(md.get("reliability_beta", 1.0))
+            reliability = a / max(a + b, 1e-12)
+        else:
+            reliability = 0.50
+        return tuple(max(0.0, min(1.0, x)) for x in (authority, freshness, reliability))
+
+    def fit(self, documents: Sequence[DocumentRecord]) -> "FuzzyIndex":
+        self.documents = list(documents)
+        corpus = [f"{d.title}. {d.text}".strip() for d in self.documents]
+        self._vectorizer = TfidfVectorizer(ngram_range=(1, 2), stop_words="english", sublinear_tf=True)
+        self._matrix = self._vectorizer.fit_transform(corpus) if corpus else None
+        return self
+
+    def add(self, documents: Iterable[DocumentRecord]) -> "FuzzyIndex":
+        return self.fit([*self.documents, *list(documents)])
+
+    def search(self, query: str, k: int = 4, candidate_pool: int | None = None) -> list[FuzzyEvidence]:
+        if not self.documents or self._vectorizer is None or self._matrix is None:
+            return []
+        q = self._vectorizer.transform([query])
+        sims = cosine_similarity(q, self._matrix)[0]
+        pool = min(candidate_pool or self.config.candidate_pool, len(self.documents))
+        candidate_ids = np.argsort(-sims, kind="stable")[:pool]
+        hits: list[FuzzyEvidence] = []
+        for i in candidate_ids:
+            doc = self.documents[int(i)]
+            relevance = float(sims[int(i)])
+            authority, freshness, reliability = self._trust(doc)
+            fuzzy, memberships = self.controller.score(relevance, authority, freshness, reliability)
+            hits.append(FuzzyEvidence(doc, relevance, fuzzy, authority, freshness, reliability, memberships))
+        hits.sort(key=lambda h: (h.fuzzy_score, h.raw_relevance, h.authority, h.freshness), reverse=True)
+        return hits[: max(1, k)]
+
+    def decision(self, hits: Sequence[FuzzyEvidence]) -> dict[str, Any]:
+        if not hits:
+            return {"route": "abstain", "confidence": 0.0, "margin": 0.0, "reason": "no_evidence"}
+        top = hits[0]
+        second = hits[1] if len(hits) > 1 else None
+        if second is None:
+            margin = top.fuzzy_score
+        else:
+            # Fuzzy confidence can intentionally plateau for equally trustworthy
+            # sources. A small relevance component breaks benign ties while exact
+            # equal-authority/equal-relevance contradictions remain ambiguous.
+            margin = 0.80 * (top.fuzzy_score - second.fuzzy_score) + 0.20 * (top.raw_relevance - second.raw_relevance)
+        if top.raw_relevance < self.config.min_raw_relevance:
+            return {"route": "abstain", "confidence": top.fuzzy_score, "margin": margin, "reason": "low_relevance"}
+        trusted_enough = top.authority >= 0.60 or top.reliability >= 0.75
+        if top.fuzzy_score >= self.config.answer_threshold and margin >= self.config.ambiguity_margin and trusted_enough:
+            return {"route": "generate", "confidence": top.fuzzy_score, "margin": margin, "reason": "strong_evidence"}
+        if top.fuzzy_score >= self.config.expand_threshold:
+            return {"route": "expand", "confidence": top.fuzzy_score, "margin": margin, "reason": "borderline_or_ambiguous"}
+        return {"route": "abstain", "confidence": top.fuzzy_score, "margin": margin, "reason": "weak_evidence"}
+
+    def should_abstain(self, hits: Sequence[FuzzyEvidence]) -> bool:
+        return self.decision(hits)["route"] != "generate"
+
+
+class PlainTfidfIndex:
+    """Baseline retriever intentionally unaware of source quality or uncertainty."""
+
+    def __init__(self, documents: Sequence[DocumentRecord]):
+        self.documents = list(documents)
+        corpus = [f"{d.title}. {d.text}".strip() for d in self.documents]
+        self.vectorizer = TfidfVectorizer(ngram_range=(1, 2), stop_words="english", sublinear_tf=True)
+        self.matrix = self.vectorizer.fit_transform(corpus)
+
+    def search(self, query: str, k: int = 4) -> list[FuzzyEvidence]:
+        q = self.vectorizer.transform([query])
+        sims = cosine_similarity(q, self.matrix)[0]
+        ids = np.argsort(-sims, kind="stable")[: min(k, len(self.documents))]
+        out = []
+        for i in ids:
+            d = self.documents[int(i)]
+            s = float(sims[int(i)])
+            out.append(FuzzyEvidence(d, s, s, 0.5, 0.5, 0.5, {}))
+        return out
+
+
+class ExtractiveGroundedGenerator:
+    """Offline answer synthesizer used for reproducible RAG evaluation.
+
+    It selects the query-most-relevant sentence from the accepted evidence and
+    returns it with an explicit citation. Production users can replace this with
+    any LangChain Runnable/LLM without changing fuzzy retrieval or routing.
+    """
+
+    abstain_text = "Insufficient reliable evidence to answer."
+
+    @staticmethod
+    def _sentences(text: str) -> list[str]:
+        parts = re.split(r"(?<=[.!?])\s+", text.strip())
+        return [p.strip() for p in parts if p.strip()]
+
+    def generate(self, question: str, hits: Sequence[FuzzyEvidence]) -> tuple[str, list[str]]:
+        if not hits:
+            return self.abstain_text, []
+        top = hits[0]
+        sentences = self._sentences(top.document.text) or [top.document.text]
+        if len(sentences) == 1:
+            best = sentences[0]
+        else:
+            vec = TfidfVectorizer(ngram_range=(1, 2), stop_words="english")
+            X = vec.fit_transform([question, *sentences])
+            scores = cosine_similarity(X[0:1], X[1:])[0]
+            best = sentences[int(np.argmax(scores))]
+        return f"{best} [{top.document.id}]", [top.document.id]
+
+
+class FuzzyRAGEngine:
+    def __init__(self, documents: Sequence[DocumentRecord] | None = None, config: FuzzyRetrievalConfig | None = None):
+        self.index = FuzzyIndex(documents or [], config)
+        self.generator = ExtractiveGroundedGenerator()
+
+    def ingest(self, documents: Sequence[DocumentRecord], replace: bool = True) -> int:
+        if replace:
+            self.index.fit(documents)
+        else:
+            self.index.add(documents)
+        return len(self.index.documents)
+
+    def query(self, question: str, k: int = 4) -> dict[str, Any]:
+        hits = self.index.search(question, k=k)
+        decision = self.index.decision(hits)
+        expanded = False
+        if decision["route"] == "expand":
+            expanded = True
+            hits = self.index.search(question, k=max(k * 2, 6), candidate_pool=self.index.config.extended_pool)
+            decision = self.index.decision(hits)
+            if decision["route"] == "expand":
+                decision = {**decision, "route": "abstain", "reason": "ambiguous_after_expansion"}
+        if decision["route"] == "generate":
+            answer, citations = self.generator.generate(question, hits)
+        else:
+            answer, citations = self.generator.abstain_text, []
+        return {
+            "question": question,
+            "answer": answer,
+            "citations": citations,
+            "confidence": round(float(decision["confidence"]), 6),
+            "margin": round(float(decision["margin"]), 6),
+            "decision": decision["route"],
+            "reason": decision["reason"],
+            "expanded": expanded,
+            "evidence": [
+                {
+                    "id": h.document.id,
+                    "title": h.document.title,
+                    "source": h.document.source,
+                    "raw_relevance": round(h.raw_relevance, 6),
+                    "fuzzy_score": round(h.fuzzy_score, 6),
+                    "authority": round(h.authority, 6),
+                    "freshness": round(h.freshness, 6),
+                    "reliability": round(h.reliability, 6),
+                }
+                for h in hits
+            ],
+        }

src/fuzzy_langchain/rag.py

@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+from typing import Any, Callable
+
+from .core import ExtractiveGroundedGenerator, FuzzyIndex, FuzzyRetrievalConfig
+from .retriever import FuzzyRetriever
+
+
+class LangChainFuzzyRAG:
+    """End-to-end LangChain RAG using fuzzy evidence control.
+
+    The optional ``generator`` receives ``(question, documents)`` and may be any
+    callable backed by an LLM. Without one, a deterministic grounded sentence
+    synthesizer is used for offline evaluation and zero-key deployments.
+    """
+
+    def __init__(self, index: FuzzyIndex, k: int = 4, generator: Callable[[str, list[Any]], str] | None = None):
+        self.index = index
+        self.retriever = FuzzyRetriever(index=index, k=k)
+        self.k = k
+        self.generator = generator
+        self.extractive = ExtractiveGroundedGenerator()
+
+    def invoke(self, question: str) -> dict[str, Any]:
+        docs = self.retriever.invoke(question)
+        # Crucially, decision logic uses the LangChain retriever result rather than
+        # silently issuing a second retrieval outside the framework path.
+        from .core import DocumentRecord, FuzzyEvidence
+
+        hits: list[FuzzyEvidence] = []
+        for d in docs:
+            md = dict(d.metadata or {})
+            record = DocumentRecord(
+                id=str(md.get("id", "unknown")),
+                text=d.page_content,
+                title=str(md.get("title", "")),
+                source=str(md.get("source", "")),
+                metadata=md,
+            )
+            hits.append(
+                FuzzyEvidence(
+                    document=record,
+                    raw_relevance=float(md.get("raw_relevance", 0.0)),
+                    fuzzy_score=float(md.get("fuzzy_score", 0.0)),
+                    authority=float(md.get("authority", 0.5)),
+                    freshness=float(md.get("freshness", 0.5)),
+                    reliability=float(md.get("reliability", 0.5)),
+                    memberships={},
+                )
+            )
+        decision = self.index.decision(hits)
+        if decision["route"] == "expand":
+            expanded_retriever = FuzzyRetriever(index=self.index, k=max(self.k * 2, 6), candidate_pool=self.index.config.extended_pool)
+            docs = expanded_retriever.invoke(question)
+            hits = []
+            for d in docs:
+                md = dict(d.metadata or {})
+                record = DocumentRecord(
+                    id=str(md.get("id", "unknown")), text=d.page_content,
+                    title=str(md.get("title", "")), source=str(md.get("source", "")), metadata=md
+                )
+                hits.append(FuzzyEvidence(
+                    record, float(md.get("raw_relevance", 0.0)), float(md.get("fuzzy_score", 0.0)),
+                    float(md.get("authority", 0.5)), float(md.get("freshness", 0.5)),
+                    float(md.get("reliability", 0.5)), {}
+                ))
+            decision = self.index.decision(hits)
+            if decision["route"] == "expand":
+                decision = {**decision, "route": "abstain", "reason": "ambiguous_after_expansion"}
+        if decision["route"] != "generate":
+            answer, citations = self.extractive.abstain_text, []
+        elif self.generator is not None:
+            answer = str(self.generator(question, docs))
+            citations = [str(d.metadata.get("id")) for d in docs[:1]]
+        else:
+            answer, citations = self.extractive.generate(question, hits)
+        return {
+            "question": question,
+            "answer": answer,
+            "citations": citations,
+            "confidence": float(decision["confidence"]),
+            "margin": float(decision["margin"]),
+            "decision": decision["route"],
+            "reason": decision["reason"],
+            "evidence": [dict(d.metadata) for d in docs],
+        }
+
+    def as_runnable(self):
+        try:
+            from langchain_core.runnables import RunnableLambda
+        except Exception as exc:  # pragma: no cover
+            raise RuntimeError("langchain-core is required for as_runnable()") from exc
+        return RunnableLambda(lambda x: self.invoke(x["question"] if isinstance(x, dict) else str(x)))

src/fuzzy_langchain/retriever.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from typing import Any
+
+from .core import DocumentRecord, FuzzyIndex
+
+try:
+    from langchain_core.documents import Document
+    from langchain_core.retrievers import BaseRetriever
+    from pydantic import ConfigDict
+except Exception:  # pragma: no cover - package remains importable without LangChain
+    Document = None
+    BaseRetriever = object
+    ConfigDict = dict
+
+
+class FuzzyRetriever(BaseRetriever):
+    """LangChain BaseRetriever adapter around :class:`FuzzyIndex`."""
+
+    if BaseRetriever is not object:
+        model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    index: Any
+    k: int = 4
+    candidate_pool: int | None = None
+
+    def _get_relevant_documents(self, query: str, *, run_manager: Any = None):
+        if Document is None:
+            raise RuntimeError("langchain-core is required for FuzzyRetriever")
+        hits = self.index.search(query, k=self.k, candidate_pool=self.candidate_pool)
+        docs = []
+        for h in hits:
+            md = dict(h.document.metadata)
+            md.update(
+                {
+                    "id": h.document.id,
+                    "title": h.document.title,
+                    "source": h.document.source,
+                    "raw_relevance": h.raw_relevance,
+                    "fuzzy_score": h.fuzzy_score,
+                    "authority": h.authority,
+                    "freshness": h.freshness,
+                    "reliability": h.reliability,
+                }
+            )
+            docs.append(Document(page_content=h.document.text, metadata=md))
+        return docs
+
+
+def to_records(documents: list[Any]) -> list[DocumentRecord]:
+    records: list[DocumentRecord] = []
+    for i, d in enumerate(documents):
+        md = dict(getattr(d, "metadata", {}) or {})
+        records.append(
+            DocumentRecord(
+                id=str(md.get("id", f"doc-{i}")),
+                text=str(getattr(d, "page_content", "")),
+                title=str(md.get("title", "")),
+                source=str(md.get("source", "")),
+                metadata=md,
+            )
+        )
+    return records