self._entries: List[ClaimRecord] = []

def register(self, claim_text: str, source: Optional[SourceReference], needs_proof: bool = True) -> None:
    self._entries.append(ClaimRecord(
        text=claim_text,
        source=source,
        requires_attribution=needs_proof
    ))

**Architecture Rationale:** Using dataclasses ensures immutability and clear serialization boundaries. The `requires_attribution` flag acknowledges that not all output claims originate from external data. Computational results, structural statements, and agent metadata should be tracked but exempt from source validation.

### Step 2: Capture Metadata at Tool Boundaries
Attribution must be recorded immediately after tool execution, before the language model synthesizes the response. This preserves the ground-truth relationship between fetched data and its origin.

```python
def execute_search_pipeline(query: str, ledger: AttributionLedger) -> str:
    raw_result = external_search_api.fetch(query)
    
    source_ref = SourceReference(
        identifier=f"search-{raw_result.run_id}",
        category="web_search",
        endpoint=raw_result.url,
        metadata={"rank": raw_result.position, "domain": raw_result.domain}
    )
    
    ledger.register(
        claim_text=f"Market analysis indicates a {raw_result.metric_value}% shift in Q3.",
        source=source_ref,
        needs_proof=True
    )
    
    return raw_result.summary

Architecture Rationale: Capturing at the tool boundary eliminates ambiguity. The ledger stores exactly what the code retrieved, not what the model interprets. This prevents drift when models rephrase, summarize, or hallucinate source relationships.

Step 3: Enforce Pre-Flight Validation

Before any response reaches the user, the ledger must verify that every claim requiring attribution has a valid source reference. This acts as a deterministic gate.

@dataclass
class CoverageReport:
    total_claims: int
    cited_claims: int
    uncited_claims: List[str]
    is_compliant: bool

    def summary(self) -> str:
        return f"Coverage: {self.cited_claims}/{self.total_claims} | Compliant: {self.is_compliant}"

def audit_coverage(ledger: AttributionLedger) -> CoverageReport:
    uncited = [
        entry.text for entry in ledger._entries 
        if entry.requires_attribution and entry.source is None
    ]
    
    return CoverageReport(
        total_claims=len(ledger._entries),
        cited_claims=len(ledger._entries) - len(uncited),
        uncited_claims=uncited,
        is_compliant=len(uncited) == 0
    )

Architecture Rationale: Validation is separated from registration to allow flexible enforcement strategies. Teams can choose to raise exceptions, log warnings, or apply partial coverage thresholds based on environment (e.g., strict in production, lenient in staging).

Step 4: Serialize for Downstream Consumption

The ledger must export to formats compatible with API responses, audit storage, and monitoring systems. JSONL is preferred for append-only audit trails, while dictionary serialization suits real-time API payloads.

import json

def export_for_api(ledger: AttributionLedger) -> Dict[str, Any]:
    return {
        "output_summary": " ".join(e.text for e in ledger._entries),
        "attribution_map": [
            {
                "claim": e.text,
                "source_id": e.source.identifier if e.source else None,
                "source_category": e.source.category if e.source else None,
                "timestamp": e.source.fetched_at.isoformat() if e.source else None
            }
            for e in ledger._entries
        ]
    }

def export_audit_log(ledger: AttributionLedger, filepath: str) -> None:
    with open(filepath, "a", encoding="utf-8") as stream:
        for entry in ledger._entries:
            record = {
                "claim": entry.text,
                "source": vars(entry.source) if entry.source else None,
                "requires_proof": entry.requires_attribution,
                "logged_at": datetime.now(timezone.utc).isoformat()
            }
            stream.write(json.dumps(record) + "\n")

Architecture Rationale: JSONL enables incremental writes without loading entire files into memory, making it suitable for long-running agent sessions. Dictionary export provides immediate compatibility with REST/GraphQL schemas. Both formats preserve the claim-source relationship for downstream querying.

Pitfall Guide

1. Trusting Model-Generated Inline Citations

Explanation: Prompting the LLM to append footnotes or markdown links to its output creates unstructured, unverifiable attribution. Models frequently fabricate URLs, mismatch IDs, or cite sources that don't contain the claimed information. Fix: Never parse citations from model output. Capture source metadata at the tool boundary and attach it programmatically before generation occurs.

2. Skipping the Pre-Flight Validation Gate

Explanation: Teams often register citations but forget to run coverage checks before returning responses. This allows uncited claims to reach production, breaking audit trails and compliance requirements. Fix: Integrate audit_coverage() into the response pipeline. Treat non-compliance as a hard failure in production environments, and log warnings in development.

3. Misclassifying Computational or Structural Claims

Explanation: Not every sentence in an agent's output originates from external data. Statements like "The pipeline processed 4 sources" or "Results are sorted by confidence" are derived from code logic, not retrieved documents. Marking these as requiring attribution causes false validation failures. Fix: Use the requires_attribution=False flag for computational, structural, or agent-metadata claims. Keep them in the ledger for completeness but exclude them from coverage enforcement.

4. Premature Source Deduplication

Explanation: Aggressively deduplicating sources before registration can break claim-level traceability. If two different queries return the same URL but different data slices, collapsing them into a single reference obscures which claim maps to which retrieval context. Fix: Preserve unique source identifiers per tool call. Deduplicate only during post-processing or reporting, never during registration. Maintain a 1:1 relationship between claim and source reference.

5. Coupling Attribution with Claim Extraction

Explanation: Attribution and claim extraction are orthogonal concerns. Attempting to split unstructured LLM output into claims while simultaneously attaching citations creates fragile parsing logic and couples two independent failure modes. Fix: Structure claims before attribution. Use schema-constrained generation (e.g., Pydantic models) to force the model to output discrete claim objects. Attach citations to those objects using the ledger, not through regex or NLP extraction.

6. Ignoring Temporal Metadata

Explanation: Sources decay. A URL that was valid at generation time may return 404 or serve updated content hours later. Without capturing fetch timestamps and snapshot metadata, audit trails become unverifiable. Fix: Always record fetched_at timestamps, query hashes, and response snapshots. Store versioned source references to enable historical replay and compliance verification.

7. Treating Attribution as Factual Verification

Explanation: A structured citation proves where a claim originated, not whether the claim is true. Teams often confuse attribution with fact-checking, expecting the ledger to validate source accuracy. Fix: Treat attribution as a provenance system. Pair it with separate RAG evaluation pipelines, cross-source consensus checks, or human-in-the-loop review for factual verification. The ledger answers "where," not "is it correct."

Production Bundle

Action Checklist

• Initialize a fresh attribution ledger at the start of each agent execution cycle
• Capture source metadata immediately after tool execution, before LLM synthesis
• Register every output claim with explicit attribution flags
• Run coverage validation as a hard gate before serializing responses
• Export audit trails to JSONL with immutable timestamps and query hashes
• Implement coverage thresholds that vary by environment (strict prod, lenient staging)
• Decouple claim structuring from attribution using schema-constrained generation
• Monitor uncited claim rates and source drift in observability dashboards

Decision Matrix

Scenario	Recommended Approach	Why	Cost Impact
Regulated Financial/RAG Output	Strict pre-flight gate + JSONL audit	Compliance requires deterministic traceability and immutable logs	High initial engineering, low compliance risk
Internal Analytics Dashboard	Coverage threshold (90%+) + dictionary export	Teams need visibility but can tolerate minor gaps in exploratory queries	Medium engineering, fast iteration
Creative/Generative Agent	No attribution ledger required	No external tool calls or factual claims to verify	Zero overhead
Multi-Agent Orchestration	Per-agent ledger + centralized audit aggregator	Isolation prevents cross-agent contamination; aggregation enables system-wide traceability	High architecture complexity, scalable compliance

Configuration Template

# production_attribution.py
import os
import json
from datetime import datetime, timezone
from typing import Dict, Any, Optional

class ProductionAttributionPipeline:
    def __init__(self, env: str = "production") -> None:
        self.env = env
        self.strict_mode = env == "production"
        self.coverage_threshold = 1.0 if self.strict_mode else 0.90
        self.audit_path = os.getenv("AUDIT_LOG_PATH", "/var/log/agent-attribution.jsonl")
        self._ledger: Dict[str, Any] = {"claims": [], "metadata": {}}

    def record_claim(self, text: str, source_id: Optional[str], source_type: Optional[str], needs_proof: bool = True) -> None:
        self._ledger["claims"].append({
            "text": text,
            "source_id": source_id,
            "source_type": source_type,
            "requires_proof": needs_proof,
            "recorded_at": datetime.now(timezone.utc).isoformat()
        })

    def validate_and_finalize(self) -> Dict[str, Any]:
        required_claims = [c for c in self._ledger["claims"] if c["requires_proof"]]
        cited_claims = [c for c in required_claims if c["source_id"] is not None]
        
        coverage = len(cited_claims) / len(required_claims) if required_claims else 1.0
        is_compliant = coverage >= self.coverage_threshold

        if not is_compliant and self.strict_mode:
            raise RuntimeError(f"Attribution coverage {coverage:.2%} below threshold {self.coverage_threshold:.0%}")

        self._ledger["metadata"] = {
            "coverage_score": coverage,
            "is_compliant": is_compliant,
            "total_claims": len(self._ledger["claims"]),
            "validated_at": datetime.now(timezone.utc).isoformat()
        }

        self._persist_audit()
        return self._ledger

    def _persist_audit(self) -> None:
        with open(self.audit_path, "a", encoding="utf-8") as f:
            for claim in self._ledger["claims"]:
                f.write(json.dumps(claim) + "\n")

Quick Start Guide

1. Install and import: Add the attribution module to your agent project. Initialize a fresh ledger instance at the beginning of each request lifecycle.
2. Hook tool outputs: Modify your tool execution layer to instantiate SourceReference objects immediately after fetches. Call register() with the claim text and source metadata before passing data to the LLM.
3. Add validation gate: Insert audit_coverage() into your response pipeline. Configure environment-specific thresholds and failure behaviors (raise vs log).
4. Serialize and monitor: Export to JSONL for audit storage and dictionary format for API responses. Track coverage scores and uncited claim rates in your observability stack.
5. Iterate with schema constraints: Transition from free-text claims to structured Pydantic models. This eliminates extraction fragility and enables deterministic attribution mapping across model upgrades.