File size: 4,194 Bytes
181a630 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 | 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)))
|