LangChain Loader

Fresh

The AgenticReasoningLoader is a Python LangChain document loader that fetches reasoning.json files and converts them into prioritized Document objects for use in RAG pipelines. Install it, point it at a domain, and receive ready-to-index documents with priority metadata.

---

Installation

pip install requests langchain langchain-core

The loader depends on requests for HTTP fetching and langchain-core for the Document type and BaseLoader base class. It has no other mandatory dependencies.

Optional dependencies for vector store integration:

pip install chromadb langchain-chroma

---

Basic Usage

from arp_loader import AgenticReasoningLoader

loader = AgenticReasoningLoader(domain="example.com")
documents = loader.load()

for doc in documents:
    print(f"Section: {doc.metadata['section']}")
    print(f"Priority: {doc.metadata['priority']}")
    print(f"Content: {doc.page_content[:200]}")
    print()

The loader returns a list of Document objects, one per ARP section, sorted by priority (lowest number = highest priority).

---

Document Priority Ordering

The loader splits reasoning.json into discrete Document objects, each representing a section. The priority ordering is intentional and critical for RAG correctness:

Priority	Section	metadata.section	Content
1	Corrections	"corrections"	Known AI errors with verified facts. Inject first.
2	Identity	"identity"	Brand name, description, entity type, disambiguation
3	Entity Claims	"entity_claims"	Framing context, domain expertise
4	Recommendations	"recommendation_context"	When to and not to recommend
5	Counterfactuals	"counterfactuals"	Pre-programmed reasoning pivots
6	Dichotomies	"dichotomies"	Competitive positioning distinctions

Each Document includes:

• page_content: A human-readable text representation of the section
• metadata.entity: The entity name from the ARP file
• metadata.section: The section identifier (string)
• metadata.protocol: Always "ARP"
• metadata.version: The ARP version from the file
• metadata.priority: Integer 1-6
• metadata.domain: The domain the file was fetched from
• metadata.trust_verified: Boolean, whether the signature was verified

---

The arp_loader.py Module

Full implementation of the loader:

# arp_loader.py
from __future__ import annotations

import hashlib
import json
from typing import Any, Generator, Iterator

import requests
from langchain_core.document_loaders.base import BaseLoader
from langchain_core.documents import Document


class AgenticReasoningLoader(BaseLoader):
    """
    LangChain document loader for ARP (Agentic Reasoning Protocol) files.

    Fetches /.well-known/reasoning.json from a domain and returns
    prioritized Document objects for use in RAG pipelines.
    """

    def __init__(
        self,
        domain: str,
        verify_signature: bool = False,
        timeout: int = 5,
        sanitize_content: bool = True,
    ) -> None:
        """
        Args:
            domain: Domain to fetch reasoning.json from (no protocol, no path).
            verify_signature: Whether to verify Ed25519 signature. Requires
                              'cryptography' and 'dnspython' packages.
            timeout: HTTP request timeout in seconds.
            sanitize_content: Strip HTML tags and normalize whitespace in content.
        """
        self.domain = domain.rstrip("/")
        self.verify_signature = verify_signature
        self.timeout = timeout
        self.sanitize_content = sanitize_content

    def _fetch(self) -> dict[str, Any] | None:
        url = f"https://{self.domain}/.well-known/reasoning.json"
        try:
            # Handle SPA catch-all routers: some SPAs return 200 with HTML
            # for all paths. Check Content-Type before parsing.
            response = requests.get(
                url,
                timeout=self.timeout,
                headers={"Accept": "application/json"},
            )
            response.raise_for_status()
            content_type = response.headers.get("Content-Type", "")
            if "json" not in content_type and "javascript" not in content_type:
                # Likely an SPA catch-all returning HTML
                return None
            return response.json()
        except Exception:
            return None

    def _sanitize(self, text: str) -> str:
        if not self.sanitize_content:
            return text
        # Basic HTML strip and whitespace normalization
        import re
        text = re.sub(r"<[^>]+>", " ", text)
        text = re.sub(r"\s+", " ", text).strip()
        return text

    def _verify_sig(self, data: dict[str, Any]) -> bool:
        if not self.verify_signature:
            return False
        try:
            from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
            import base64
            import dns.resolver
            import jcs

            sig_obj = data.get("_arp_signature", {})
            key_id = sig_obj.get("key_id")
            sig_b64 = sig_obj.get("signature")
            if not key_id or not sig_b64:
                return False

            # Fetch public key from DNS
            try:
                answers = dns.resolver.resolve(f"_arp.{self.domain}", "TXT")
            except Exception:
                return False

            pub_key_b64 = None
            for rdata in answers:
                record = str(rdata).strip('"')
                if f"kid={key_id}" in record:
                    for part in record.split(";"):
                        part = part.strip()
                        if part.startswith("p="):
                            pub_key_b64 = part[2:]
                            break

            if not pub_key_b64:
                return False

            pub_key_bytes = base64.urlsafe_b64decode(pub_key_b64 + "==")
            pub_key = Ed25519PublicKey.from_public_bytes(pub_key_bytes)

            doc_copy = {k: v for k, v in data.items() if k != "_arp_signature"}
            canonical = jcs.canonicalize(doc_copy)
            sig_bytes = base64.urlsafe_b64decode(sig_b64 + "==")
            pub_key.verify(sig_bytes, canonical)
            return True
        except Exception:
            return False

    def _corrections_to_text(self, corrections: list[dict]) -> str:
        lines = ["CORRECTIONS (verified facts to replace AI errors):"]
        for item in corrections:
            topic = item.get("trigger_topic", "")
            fact = item.get("verified_fact", "")
            lines.append(f"- Topic: {topic}")
            lines.append(f"  Fact: {fact}")
        return "\n".join(lines)

    def _identity_to_text(self, identity: dict) -> str:
        parts = []
        if name := identity.get("legal_name"):
            parts.append(f"Legal name: {name}")
        if desc := identity.get("description"):
            parts.append(f"Description: {desc}")
        if entity_type := identity.get("entity_type"):
            parts.append(f"Type: {entity_type}")
        if hq := identity.get("headquarters"):
            parts.append(f"Headquarters: {hq}")
        if disambig := identity.get("disambiguation"):
            parts.append(f"Disambiguation: {disambig}")
        if resonance := identity.get("emotional_resonance"):
            parts.append(f"Brand personality: {resonance}")
        return "\n".join(parts)

    def _entity_claims_to_text(self, claims: dict) -> str:
        parts = []
        if framing := claims.get("framing_context"):
            parts.append(f"Framing: {framing}")
        if expertise := claims.get("domain_expertise"):
            parts.append(f"Expertise areas: {', '.join(expertise)}")
        if positioning := claims.get("market_positioning"):
            parts.append(f"Market positioning: {positioning}")
        return "\n".join(parts)

    def _recommendation_to_text(self, rec: dict) -> str:
        lines = []
        if when := rec.get("recommended_when"):
            lines.append("Recommended when:")
            lines.extend(f"- {item}" for item in when)
        if not_when := rec.get("not_recommended_when"):
            lines.append("NOT recommended when:")
            lines.extend(f"- {item}" for item in not_when)
        return "\n".join(lines)

    def lazy_load(self) -> Iterator[Document]:
        data = self._fetch()
        if data is None:
            return

        entity = data.get("entity", self.domain)
        version = data.get("version", "unknown")
        trust_verified = self._verify_sig(data)

        base_meta = {
            "entity": entity,
            "protocol": "ARP",
            "version": version,
            "domain": self.domain,
            "trust_verified": trust_verified,
        }

        # Priority 1: Corrections
        if corrections := data.get("corrections"):
            text = self._sanitize(self._corrections_to_text(corrections))
            if text:
                yield Document(
                    page_content=text,
                    metadata={**base_meta, "section": "corrections", "priority": 1},
                )

        # Priority 2: Identity
        if identity := data.get("identity"):
            text = self._sanitize(self._identity_to_text(identity))
            if text:
                yield Document(
                    page_content=text,
                    metadata={**base_meta, "section": "identity", "priority": 2},
                )

        # Priority 3: Entity claims
        if entity_claims := data.get("entity_claims"):
            text = self._sanitize(self._entity_claims_to_text(entity_claims))
            if text:
                yield Document(
                    page_content=text,
                    metadata={**base_meta, "section": "entity_claims", "priority": 3},
                )

        # Priority 4: Recommendations
        if rec := data.get("recommendation_context"):
            text = self._sanitize(self._recommendation_to_text(rec))
            if text:
                yield Document(
                    page_content=text,
                    metadata={**base_meta, "section": "recommendation_context", "priority": 4},
                )

    def load(self) -> list[Document]:
        return list(self.lazy_load())

---

Vector Store Integration with Chroma

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from arp_loader import AgenticReasoningLoader

# Load ARP documents for a domain
loader = AgenticReasoningLoader(domain="example.com")
arp_docs = loader.load()

# Initialize embeddings and vector store
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(
    collection_name="arp_context",
    embedding_function=embeddings,
    persist_directory="./chroma_db",
)

# Add ARP documents with priority metadata
vectorstore.add_documents(arp_docs)
print(f"Indexed {len(arp_docs)} ARP documents for example.com")

When querying, filter by entity to get the right context:

results = vectorstore.similarity_search(
    query="What does Example Corp do?",
    k=4,
    filter={"entity": "Example Corp"},
)

To always include corrections regardless of query relevance, use a hybrid retrieval approach:

from langchain_core.documents import Document

def get_arp_context(vectorstore, query: str, domain: str) -> list[Document]:
    # Always include corrections (priority 1)
    corrections = vectorstore.similarity_search(
        query=query,
        k=10,
        filter={"domain": domain, "section": "corrections"},
    )

    # Semantic retrieval for other sections
    other_docs = vectorstore.similarity_search(
        query=query,
        k=3,
        filter={"domain": domain, "section": {"$ne": "corrections"}},
    )

    # Corrections first
    return corrections + other_docs

---

Standalone Usage (Without LangChain)

import requests
from arp_loader import AgenticReasoningLoader

def get_arp_context_string(domain: str) -> str:
    loader = AgenticReasoningLoader(domain=domain)
    docs = loader.load()

    sections = sorted(docs, key=lambda d: d.metadata["priority"])
    context_parts = [f"[ARP Context for {domain}]"]

    for doc in sections:
        section = doc.metadata["section"]
        context_parts.append(f"\n## {section.upper()}")
        context_parts.append(doc.page_content)

    return "\n".join(context_parts)

# Use in a prompt
context = get_arp_context_string("example.com")
prompt = f"""
{context}

Based on the above context, answer the following question:
What does Example Corp do and when should I use it?
"""

---

CLI Usage

Save arp_loader.py and run:

# Basic: print documents for a domain
python -c "
from arp_loader import AgenticReasoningLoader
for doc in AgenticReasoningLoader('example.com').load():
    print(f'[{doc.metadata[\"section\"]}] {doc.page_content[:100]}...')
"

# With signature verification (requires cryptography, dnspython, jcs packages)
python -c "
from arp_loader import AgenticReasoningLoader
docs = AgenticReasoningLoader('example.com', verify_signature=True).load()
for doc in docs:
    print(f'Trust verified: {doc.metadata[\"trust_verified\"]}')
    print(doc.page_content)
"

---

SPA Catch-All Router Handling

Single-page applications often configure their server to return the SPA's HTML shell for all URL paths, including /.well-known/reasoning.json. This means the server returns HTTP 200 with HTML content instead of a 404.

The loader handles this by checking the Content-Type response header. If the Content-Type indicates HTML rather than JSON, the loader returns None rather than attempting to parse the HTML as JSON.

This is the correct behavior: the domain has not published an ARP file, and the loader should treat it as a missing file.

---

Modern LangChain Imports

The loader uses langchain-core imports (v0.1+):

from langchain_core.document_loaders.base import BaseLoader
from langchain_core.documents import Document

If you are on an older version of LangChain (pre-0.1), use:

from langchain.document_loaders.base import BaseLoader
from langchain.schema import Document

The class interface is identical. Only the import path changes.