AHS-Core — Conflict Resolution & Forensic Traceability for Agentic RAG

When documents disagree, AHS catches it — and proves it.

AHS-Core is a Python library that sits on top of your RAG / agent stack (LangGraph, CrewAI, raw OpenAI — anything) and adds three things generic agent frameworks don't give you out of the box:

1. Conflict detection between retrieved premises — direct contradictions, numeric disagreements, and version drift (old SOP vs. new regulation).
2. Cascading-hallucination awareness — every conflict is tracked across reasoning stages so an error from hop 1 doesn't silently become the answer at hop 5.
3. A forensic audit log — append-only, hash-chained JSONL recording every query, retrieved document, conflict, prompt, and answer. Tamper-evident and auditor-ready.

AHS is not a replacement for LangGraph or CrewAI. It's the layer you add when your RAG has to be correct and provable — compliance, legal, clinical, insurance, finance, policy.

Install

pip install ahs-agentic                # core (offline, deterministic LLM)
pip install ahs-agentic[openai]        # + OpenAI embeddings / LLM
pip install ahs-agentic[dev]           # + pytest, build, twine

From source:

git clone https://github.com/sachinagenticai/AHS_Agentic.git
cd AHS_Agentic
pip install -e ".[dev]"
pytest

60-second demo (no API key)

ahs demo

…or, in Python:

import asyncio
from ahs_agentic import (
    Evidence, HashEmbedder, InMemoryRetriever, SpeculativeRetriever,
    SkepticSubroutine, Reconciler, DeterministicLLM,
)

async def main():
    embedder = HashEmbedder(dim=512)
    backend  = InMemoryRetriever(embedder=embedder)
    retr     = SpeculativeRetriever(backend=backend, embedder=embedder)
    skeptic  = SkepticSubroutine(embedder=embedder, sensitivity_threshold=0.4)

    result = await Reconciler(retr, skeptic, DeterministicLLM()).reconcile(
        question="How long must customer tickets be retained?",
        corpus=[
            Evidence(id="v2022", source="policy.pdf", version="2022",
                     text="Tickets must be retained for 24 months. Encryption is optional."),
            Evidence(id="v2024", source="policy.pdf", version="2024",
                     text="Tickets must be retained for 60 months. Encryption at rest is mandatory."),
        ],
        audit_path="audit.jsonl",
    )

    print(result.answer)
    print("conflicts:", [c.conflict_type for c in result.conflicts])
    print("audit_id :", result.audit_id)

asyncio.run(main())

Sample output:

Question: How long must customer tickets be retained?

Answer based on the evidence:
- Tickets must be retained for 24 months. Encryption is optional. [v2022]
- Tickets must be retained for 60 months. Encryption at rest is mandatory. [v2024]

Source(s): v2022, v2024

Conflicts detected (human review recommended):
- [NUMERIC_DISAGREEMENT] between 'v2022' and 'v2024': Numbers disagree: ['24'] vs ['60'] ...

Architecture

User question
     │
     ▼
┌──────────────────┐
│   Reconciler     │   one-call façade
└──────────────────┘
     │
     ├──► SpeculativeRetriever ──► RetrieverBackend (in-mem / Chroma / pgvector)
     │       (parallel fan-out, bounded concurrency, metrics)
     │
     ├──► SkepticSubroutine  ──► Embedder (HashEmbedder / OpenAI / local)
     │       cosine delta + negation/version/numeric heuristics
     │
     ├──► ForensicLogger  ──► audit.jsonl  (append-only, hash-chained)
     │
     └──► LLM (DeterministicLLM / OpenAI / any provider)
             grounded, citation-only answer
     │
     ▼
ReconciliationResult { answer, evidence[], conflicts[], audit_id, cited_ids[] }

Why "conflict-aware" matters

Naive RAG retrieves chunks and asks the LLM to answer. When two chunks disagree, the model either silently picks one or hallucinates a compromise. Production pipelines in 2026 show the damage:

• Cascading hallucinations propagate across multi-step pipelines; stage-level conflict detection interrupts that propagation before it reaches the final answer.
• GraphRAG with factuality gates cuts hallucinations by ~62% vs. naive chunk-and-retrieve (MLOps Community benchmark, May 2026).
• Self-RAG / CRAG achieve 5.8–10.5% hallucination rates vs. 14%+ for static RAG.

AHS packages the detection and audit parts of those architectures into a library you can add to an existing pipeline in an afternoon.

Command-line usage

# Reconcile a question against a folder of .txt/.md files
ahs reconcile \
  --question "How long are customer tickets retained?" \
  --corpus  ./docs/policies/ \
  --out     report.json \
  --audit   audit.jsonl

# Or feed a JSONL corpus
ahs reconcile --question "..." --corpus corpus.jsonl --out report.json

Use it with LangGraph / CrewAI

AHS is just a Python object — call it from any node / agent:

# LangGraph node
def ahs_node(state):
    result = asyncio.run(reconciler.reconcile(
        question=state["question"], corpus=state["corpus"]))
    return {"answer": result.answer,
            "conflicts": [c.to_dict() for c in result.conflicts],
            "audit_id": result.audit_id}

# CrewAI custom agent
class ComplianceReviewer(Agent):
    def run(self, question, corpus):
        return asyncio.run(reconciler.reconcile(question, corpus)).to_dict()

Project status — v0.1 MVP

What's implemented

• ✅ Installable PyPI package (ahs-agentic) with a stable public API
• ✅ Pluggable Embedder, RetrieverBackend, LLMClient interfaces
• ✅ HashEmbedder (offline) + OpenAIEmbedder
• ✅ InMemoryRetriever + SpeculativeRetriever (parallel fan-out, batching, concurrency limits, metrics)
• ✅ SkepticSubroutine — cosine delta plus direct-contradiction, numeric-disagreement, and version-drift detection
• ✅ Reconciler façade — plan → retrieve → conflict-check → grounded LLM answer → audit log
• ✅ OrchestratorManager + TaskStateMachine + BaseAgent — audited routing with keyword/specialty scoring and escalation
• ✅ ForensicLogger — hash-chained JSONL, tamper-evident
• ✅ DeterministicLLM for offline tests/CI + OpenAILLM adapter
• ✅ Resilience layer (tenacity retries, tiktoken prompt trimming)
• ✅ CLI: ahs demo, ahs reconcile
• ✅ Full pytest suite + GitHub Actions CI

What's v0.2+ (not in MVP)

• 🔜 Chroma / FAISS / pgvector retriever backends
• 🔜 LangGraph / CrewAI integration packages
• 🔜 Cross-stage cascade detector (full pipeline)
• 🔜 Anthropic / local-LLM clients
• 🔜 Orchestrator with TaskStateMachine (see docs/ORCHESTRATOR.md for the design)
• 🔜 Streamlit explorer UI
• 🔜 Public conflict-reconciliation benchmark

See ROADMAP.md.

Smoke evaluation

examples/eval.py is a tiny, reproducible, offline eval on 20 hand-curated premise pairs (10 conflicts, 10 aligned paraphrases). It is not a research benchmark — it's a guard rail so changes to the Skeptic don't silently break obvious cases.

$ python examples/eval.py
Accuracy : 1.00
Precision: 1.00
Recall   : 1.00
F1       : 1.00

For real numbers against your corpus, pair SkepticSubroutine with OpenAIEmbedder (or any semantic embedder) and run it over a labelled conflict set.

Development

pip install -e ".[dev]"
pytest                       # all tests, offline
python examples/reconcile_demo.py
python examples/orchestrator_demo.py
python examples/resilience_demo.py
python examples/eval.py
ahs demo

License

MIT — see LICENSE.