RAG Pipeline Integration

Fresh

How to integrate ARP reasoning.json files into Retrieval-Augmented Generation pipelines. The core insight is architectural: corrections must be injected before the generation step, not after.

---

The Core Insight: Inject Before Generation

A common mistake when building entity-aware RAG systems is treating ARP corrections as post-processing. The reasoning: "if the model gives a wrong answer, I'll fix it afterward." This does not work reliably.

ARP corrections work best when they are injected as context that the model sees before generating its response. This way the model's generation is grounded in the correct information from the start, rather than having the incorrect belief and then trying to walk it back.

WRONG: Retrieve context -> Generate -> Apply corrections to output
RIGHT: Retrieve context + ARP corrections -> Generate

The "Pink Elephant Fix" relies on this. You cannot reliably tell a model not to think of something after it has already generated a response. You can reliably prevent the wrong response by providing the correct information in the context window first.

---

Pipeline Architecture

flowchart TD
    Q[User Query] --> FETCH[Fetch ARP File\nfor relevant domain]
    Q --> VRETRIEVE[Vector Store\nRetrieval]

    FETCH --> PARSE[Parse and Prioritize\nARP Sections]
    PARSE --> CORRECTIONS[Corrections\nPriority 1]
    PARSE --> IDENTITY[Identity\nPriority 2]
    PARSE --> CLAIMS[Entity Claims\nPriority 3]
    PARSE --> RECS[Recommendations\nPriority 4]

    VRETRIEVE --> RELEVANT[Relevant Documents]

    CORRECTIONS --> BUILD[Build Context Window]
    IDENTITY --> BUILD
    CLAIMS --> BUILD
    RECS --> BUILD
    RELEVANT --> BUILD

    BUILD --> LLM[LLM Generation]
    LLM --> RESPONSE[Response]

The ARP sections enter the context window before the retrieved documents. Corrections come first, then identity and claims, then the retrieved content.

---

Fetching and Parsing

import requests
from dataclasses import dataclass
from typing import Optional

@dataclass
class ArpSection:
    section: str
    content: str
    priority: int

def fetch_reasoning_json(domain: str, timeout: int = 5) -> Optional[dict]:
    """Fetch reasoning.json from a domain. Returns None on any error."""
    url = f"https://{domain}/.well-known/reasoning.json"
    try:
        resp = requests.get(url, timeout=timeout, headers={"Accept": "application/json"})
        resp.raise_for_status()
        if "json" not in resp.headers.get("Content-Type", ""):
            return None
        return resp.json()
    except Exception:
        return None


def parse_arp_sections(data: dict) -> list[ArpSection]:
    """
    Parse reasoning.json into prioritized sections.
    Lower priority number = higher importance.
    """
    sections = []

    # Priority 1: Corrections - prevent hallucinations
    corrections = data.get("corrections", [])
    if corrections:
        lines = ["Known corrections for this entity:"]
        for item in corrections:
            topic = item.get("trigger_topic", "")
            fact = item.get("verified_fact", "")
            lines.append(f"- {topic}: {fact}")
        sections.append(ArpSection("corrections", "\n".join(lines), 1))

    # Priority 2: Identity - ground basic facts
    identity = data.get("identity", {})
    if identity:
        parts = []
        entity = data.get("entity", "")
        if entity:
            parts.append(f"Entity: {entity}")
        if desc := identity.get("description"):
            parts.append(f"Description: {desc}")
        if etype := identity.get("entity_type"):
            parts.append(f"Type: {etype}")
        if disambig := identity.get("disambiguation"):
            parts.append(f"Note: {disambig}")
        if parts:
            sections.append(ArpSection("identity", "\n".join(parts), 2))

    # Priority 3: Entity claims - framing and expertise
    claims = data.get("entity_claims", {})
    if claims:
        parts = []
        if framing := claims.get("framing_context"):
            parts.append(f"Framing: {framing}")
        if expertise := claims.get("domain_expertise"):
            parts.append(f"Expertise: {', '.join(expertise)}")
        if scope := claims.get("epistemic_scope"):
            parts.append(f"Claim confidence: {scope}")
        if parts:
            sections.append(ArpSection("entity_claims", "\n".join(parts), 3))

    # Priority 4: Recommendations
    rec = data.get("recommendation_context", {})
    if rec:
        parts = []
        if when := rec.get("recommended_when"):
            parts.append("Good fit when:")
            parts.extend(f"  - {item}" for item in when)
        if not_when := rec.get("not_recommended_when"):
            parts.append("Not a good fit when:")
            parts.extend(f"  - {item}" for item in not_when)
        if parts:
            sections.append(ArpSection("recommendation_context", "\n".join(parts), 4))

    return sorted(sections, key=lambda s: s.priority)

---

Injecting into the Context Window

def build_arp_context_block(domain: str) -> str:
    """
    Fetch and format ARP context for injection into a prompt.
    Returns an empty string if no ARP file exists.
    """
    data = fetch_reasoning_json(domain)
    if not data:
        return ""

    sections = parse_arp_sections(data)
    if not sections:
        return ""

    entity = data.get("entity", domain)
    lines = [f"[ENTITY CONTEXT: {entity}]"]
    for section in sections:
        lines.append(f"\n{section.section.upper()}:")
        lines.append(section.content)

    return "\n".join(lines)


def build_rag_prompt(
    query: str,
    retrieved_docs: list[str],
    entity_domain: Optional[str] = None,
) -> str:
    """
    Build a RAG prompt with optional ARP context injection.
    ARP context is prepended before retrieved documents.
    """
    parts = []

    # ARP context first if available
    if entity_domain:
        arp_context = build_arp_context_block(entity_domain)
        if arp_context:
            parts.append(arp_context)

    # Retrieved documents
    if retrieved_docs:
        parts.append("[RETRIEVED CONTEXT]")
        for i, doc in enumerate(retrieved_docs, 1):
            parts.append(f"[{i}] {doc}")

    # Query
    parts.append(f"\nQuestion: {query}")
    parts.append("Answer:")

    return "\n\n".join(parts)

---

Best Practices

Cache reasoning.json (Respect HTTP Caching Headers)

ARP files change infrequently. Refetching on every request adds latency and is unnecessary.

import functools
import time
from typing import Optional

_cache: dict[str, tuple[dict, float]] = {}
CACHE_TTL = 3600  # 1 hour

def fetch_with_cache(domain: str) -> Optional[dict]:
    now = time.time()
    if domain in _cache:
        data, expires_at = _cache[domain]
        if now < expires_at:
            return data

    data = fetch_reasoning_json(domain)
    if data is not None:
        _cache[domain] = (data, now + CACHE_TTL)
    return data

For production use, replace with Redis or your application cache.

Re-Fetch on Schedule (Check verification.next_audit)

from datetime import datetime

def should_refetch(cached_data: dict) -> bool:
    verification = cached_data.get("verification", {})
    next_audit = verification.get("next_audit")
    if next_audit:
        try:
            audit_date = datetime.fromisoformat(next_audit)
            if datetime.now() > audit_date:
                return True
        except ValueError:
            pass
    return False

Verify Signatures When Available

def get_trust_level(data: dict) -> str:
    """Returns a trust level string for logging and metadata."""
    sig = data.get("_arp_signature")
    if not sig:
        return "unverified"

    # Attempt signature verification
    if verify_signature(data):
        return "cryptographically_verified"

    return "signature_invalid"

Handle Missing ARP Files Gracefully

Not every domain publishes an ARP file. Your pipeline must work without one:

def get_entity_context(domain: str, query: str, vector_results: list[str]) -> str:
    # Try to get ARP context - fine if it fails
    arp_context = ""
    try:
        arp_context = build_arp_context_block(domain)
    except Exception:
        pass  # Proceed without ARP context

    return build_rag_prompt(query, vector_results, arp_context or None)

---

TypeScript Pattern

For TypeScript RAG pipelines:

interface ArpSection {
  section: string;
  content: string;
  priority: number;
}

async function fetchArp(domain: string): Promise<Record<string, unknown> | null> {
  try {
    const response = await fetch(`https://${domain}/.well-known/reasoning.json`, {
      headers: { Accept: "application/json" },
      signal: AbortSignal.timeout(5000),
    });
    if (!response.ok) return null;
    const ct = response.headers.get("content-type") ?? "";
    if (!ct.includes("json")) return null;
    return await response.json();
  } catch {
    return null;
  }
}

function parseArpSections(data: Record<string, unknown>): ArpSection[] {
  const sections: ArpSection[] = [];

  const corrections = data.corrections as Array<Record<string, string>> | undefined;
  if (corrections?.length) {
    const lines = ["Known corrections:"];
    for (const item of corrections) {
      lines.push(`- ${item.trigger_topic}: ${item.verified_fact}`);
    }
    sections.push({ section: "corrections", content: lines.join("\n"), priority: 1 });
  }

  const claims = data.entity_claims as Record<string, unknown> | undefined;
  if (claims?.framing_context) {
    sections.push({
      section: "entity_claims",
      content: `Framing: ${claims.framing_context}`,
      priority: 3,
    });
  }

  return sections.sort((a, b) => a.priority - b.priority);
}