RAG Harness: Extracting a Trusted Foundation from Evidence-Bound

┌──────────────────────────────────────────────────────────────────────┐
│                        RAG HARNESS CORE                              │
│                                                                      │
│  ┌──────────┐   ┌──────────┐   ┌───────────┐   ┌────────────────┐  │
│  │ Ingest   │──>│  Store   │──>│ Retrieve  │──>│   Generate     │  │
│  │          │   │          │   │           │   │                │  │
│  │ Parse    │   │ Chunks   │   │ Hybrid    │   │ LLM + Verify   │  │
│  │ Chunk    │   │ Vectors  │   │ Search    │   │ Grade + Cite   │  │
│  │ Embed    │   │ Metadata │   │ Rerank    │   │ Validate       │  │
│  └──────────┘   └──────────┘   └───────────┘   └────────────────┘  │
│       │              │              │                  │             │
│  ┌────┴────┐   ┌─────┴────┐  ┌─────┴─────┐    ┌──────┴──────┐     │
│  │ Parser  │   │ Search   │  │ Embedding │    │ LLM Client  │     │
│  │ Client  │   │ Client   │  │ Client    │    │ (4 provdrs) │     │
│  │ (3 imp) │   │ (2 imp)  │  │ (2 imp)   │    │ + Reranker  │     │
│  └─────────┘   └──────────┘  └───────────┘    └─────────────┘     │
│                                                                      │
│  CROSS-CUTTING:                                                      │
│  ┌────────────┐ ┌────────────┐ ┌──────────┐ ┌───────────────────┐  │
│  │ Security   │ │ Observ-    │ │ Caching  │ │ Cost Tracking     │  │
│  │ (Injection │ │ ability    │ │ (Embed + │ │ (per-component,   │  │
│  │  + Blocklst│ │ (OTEL +   │ │  Query   │ │  per-request)     │  │
│  │  + Homglyph│ │  Langfuse) │ │  LRU+TTL)│ │                   │  │
│  └────────────┘ └────────────┘ └──────────┘ └───────────────────┘  │
│                                                                      │
│  EXTENSION POINTS (domain-specific):                                 │
│  ┌────────────┐ ┌────────────┐ ┌──────────┐ ┌───────────────────┐  │
│  │ Data       │ │ Evidence   │ │ Verifier │ │ Response          │  │
│  │ Isolation  │ │ Grader     │ │ Prompt   │ │ Schema            │  │
│  │ (tenant/   │ │ (A/B/C or  │ │ (legal,  │ │ (citations,       │  │
│  │  workspace)│ │  custom)   │ │  medical)│ │  refusals, etc)   │  │
│  └────────────┘ └────────────┘ └──────────┘ └───────────────────┘  │
└──────────────────────────────────────────────────────────────────────┘

class DocumentQualityRouter:
    """Route documents to appropriate processing pipeline."""
 
    def classify(self, file_path: str, metadata: dict) -> DocumentProfile:
        """Return quality profile: digital/scanned/mixed, structure level, table density."""
 
    def route(self, profile: DocumentProfile) -> ProcessingPipeline:
        """Select parser, chunking strategy, and embedding approach based on profile."""

class TableProcessor:
    """Extract and dual-embed tables from documents."""
 
    def detect_tables(self, page_content: str) -> list[TableRegion]:
        """Identify table boundaries in page text."""
 
    def extract_structured(self, table: TableRegion) -> StructuredTable:
        """Convert to CSV/markdown preserving headers and values."""
 
    def dual_embed(self, table: StructuredTable) -> tuple[list[float], list[float]]:
        """Return (structural_embedding, summary_embedding)."""

class MetadataFilter:
    """Build search filters from query analysis and document metadata."""
 
    def extract_query_filters(self, question: str) -> dict[str, Any]:
        """Parse date ranges, doc types, entity names from question."""
 
    def apply_to_search(self, base_query: SearchQuery, filters: dict) -> SearchQuery:
        """Add metadata filters to search request."""

class DocumentTree:
    """Hierarchical document representation."""
 
    def build_tree(self, parsed_doc: ParseResult) -> TreeNode:
        """Build section/subsection tree from parsed document."""
 
    def multi_level_embed(self, tree: TreeNode) -> list[TreeChunk]:
        """Embed at multiple granularity levels: section summary + leaf chunks."""
 
    def route_query(self, question: str, tree: TreeNode) -> list[TreeChunk]:
        """Broad question -> high-level nodes. Precise question -> leaf nodes."""

class AgenticRetriever:
    """Multi-turn retrieval with query refinement."""
 
    def retrieve_with_refinement(
        self,
        question: str,
        max_iterations: int = 3,
    ) -> AgenticResult:
        """
        Loop:
        1. Analyze question -> decompose if complex
        2. Retrieve candidates
        3. Verify relevance
        4. If confidence < threshold: reformulate query, try again
        5. If max iterations: return best result or refuse
        """

rag-harness/
├── pyproject.toml                # Package config
├── README.md
│
├── rag_harness/
│   ├── __init__.py
│   ├── pipeline.py               # RAGPipeline orchestrator
│   ├── config.py                 # Pydantic Settings (not flat env vars)
│   │
│   ├── ingest/
│   │   ├── __init__.py
│   │   ├── chunker.py            # Chunking with offsets (from ingestion.py)
│   │   ├── quality_router.py     # NEW: Document quality classification
│   │   └── table_processor.py    # NEW: Table detection + dual embedding
│   │
│   ├── providers/
│   │   ├── __init__.py
│   │   ├── search/
│   │   │   ├── base.py           # SearchClient ABC (from search/base.py)
│   │   │   ├── azure.py          # Azure AI Search (from search/azure.py)
│   │   │   └── local.py          # BM25 + Vector + RRF (from search/local.py)
│   │   ├── llm/
│   │   │   ├── base.py           # LLMClient ABC (from llm/base.py)
│   │   │   ├── azure_openai.py
│   │   │   ├── anthropic.py
│   │   │   ├── gemini.py
│   │   │   └── ollama.py
│   │   ├── embedding/
│   │   │   ├── base.py           # EmbeddingClient ABC
│   │   │   ├── azure_openai.py
│   │   │   └── local.py
│   │   ├── parser/
│   │   │   ├── base.py           # ParserClient ABC (from parsers/base.py)
│   │   │   ├── pypdf.py
│   │   │   ├── marker.py
│   │   │   └── llamaparse.py
│   │   └── reranker/
│   │       ├── base.py           # RerankerClient ABC
│   │       └── local.py
│   │
│   ├── retrieval/
│   │   ├── __init__.py
│   │   ├── bm25.py               # BM25 engine (from retrieval.py)
│   │   ├── fusion.py             # RRF fusion (from retrieval.py)
│   │   ├── hybrid.py             # Hybrid search orchestration
│   │   └── agentic.py            # NEW: Multi-turn retrieval loop
│   │
│   ├── verification/
│   │   ├── __init__.py
│   │   ├── verifier.py           # LLM verification (from verification.py)
│   │   ├── parallel.py           # Parallel verification (from ask_service.py)
│   │   └── auto_verify.py        # Fast path for high-confidence (from ask_service.py)
│   │
│   ├── citation/
│   │   ├── __init__.py
│   │   ├── validator.py          # Citation validation (from evidence.py)
│   │   ├── grader.py             # Evidence grading (from evidence.py, configurable)
│   │   └── negation.py           # Adversarial negation detection (from evidence.py)
│   │
│   ├── security/
│   │   ├── __init__.py
│   │   ├── injection.py          # Prompt injection detection (from policy.py)
│   │   ├── blocklist.py          # Span content blocklist (from verification.py)
│   │   └── isolation.py          # Data isolation interface (NEW)
│   │
│   ├── observability/
│   │   ├── __init__.py
│   │   ├── otel.py               # OpenTelemetry setup (from otel.py)
│   │   ├── langfuse.py           # Langfuse integration (from otel.py)
│   │   ├── metrics.py            # Custom metrics (from otel.py)
│   │   └── cost.py               # Cost tracking (from cost.py)
│   │
│   ├── cache/
│   │   ├── __init__.py
│   │   ├── embedding_cache.py    # LRU embedding cache (from cache.py)
│   │   └── query_cache.py        # TTL query cache (from cache.py)
│   │
│   └── http/
│       ├── __init__.py
│       └── client.py             # httpx pool manager (from http_client.py)
│
├── tests/
│   ├── test_bm25.py
│   ├── test_fusion.py
│   ├── test_injection.py
│   ├── test_citation.py
│   ├── test_verification.py
│   ├── test_cache.py
│   ├── test_cost.py
│   └── test_pipeline.py          # End-to-end pipeline test
│
└── examples/
    ├── legal/                     # Evidence-Bound domain config
    │   ├── grader.py              # Legal evidence grading
    │   ├── verifier_prompt.txt    # Legal verification prompt
    │   ├── roles.py               # Attorney/Paralegal/Viewer
    │   └── schemas.py             # Citation + EvidenceSupport
    ├── medical/                   # Example: clinical RAG
    │   ├── grader.py
    │   ├── verifier_prompt.txt
    │   └── schemas.py
    └── quickstart.py              # Minimal working example

from rag_harness import RAGPipeline
from rag_harness.providers.llm import AzureOpenAIClient
from rag_harness.providers.embedding import AzureOpenAIEmbeddingClient
from rag_harness.providers.search import LocalSearchClient
from rag_harness.providers.parser import MarkerParser
from rag_harness.citation import EvidenceGrader
from rag_harness.security import InjectionDetector
 
# Configure the pipeline
pipeline = RAGPipeline(
    llm=AzureOpenAIClient(endpoint="...", api_key="...", model="gpt-5-mini"),
    embedding=AzureOpenAIEmbeddingClient(endpoint="...", deployment="text-embedding-3-large"),
    search=LocalSearchClient(top_k=5, rrf_k=60),
    parser=MarkerParser(force_ocr=False),
 
    # Domain-specific extension points
    grader=EvidenceGrader(
        thresholds={"A": 2.5, "B": 1.5, "C": 0.3},
        require_verification=True,
    ),
    security=InjectionDetector(patterns="default"),
 
    # Infrastructure
    cache_embeddings=True,
    cache_queries=True,
    enable_langfuse=True,
    enable_otel=True,
)
 
# Ingest a document
pipeline.ingest("contract.pdf", metadata={"doc_type": "contract", "date": "2024-01-15"})
 
# Ask a question
result = pipeline.ask(
    question="What is the indemnification cap?",
    filters={"tenant_id": "acme-corp", "workspace_id": "case-42"},
)
 
# Result includes: answer, citations, evidence grade, cost, latency breakdown
print(result.answer)          # "According to Section 8.2 (page 14)..."
print(result.citations)       # [Citation(doc="contract.pdf", page=14, span="...")]
print(result.evidence.grade)  # "A"
print(result.cost.total_usd)  # 0.0034
print(result.latency.total_ms)# 2100