AID-H-020.002: Source Reputation Weighting

Version 1.20261015 · Actionable sub-technique

Assign and store per-chunk reputation scores based on source trust. Re-rank retrievals by combining similarity and reputation to bias toward trusted sources.

Pillars: data

Lifecycle: building, operation

Scope boundary

Owns calibrated, provenance-bound source reputation scores and their policy-approved use in retrieval ranking to prefer more trusted sources. Reputation weighting does not grant read entitlement or prove semantic truth.

AID-H-020.003: AID-H-020.002 changes ranking weight from source trust; AID-H-020.003 enforces whether the current principal may retrieve each document or chunk. High reputation never overrides access control.

AID-H-020.004: AID-H-020.002 supplies source-trust ranking evidence; AID-H-020.004 enforces whether the source, release, and content are approved and fresh for current context assembly. A reputable source can still be stale or absent from the active context policy.

Implementation guidance

AID-H-020.002-G001: Apply a versioned, provenance-bound source reputation policy after score calibration and validate its effect on a labeled retrieval suite.

AID-H-020.002-G001

Apply a versioned, provenance-bound source reputation policy after score calibration and validate its effect on a labeled retrieval suite.

Do not trust document-supplied reputation metadata

Resolve a chunk's source ID from the authenticated connector and ingestion lineage record. Load reputation, similarity normalization bounds, unknown-source behavior, and weighting from a signed/versioned policy. Do not mix raw distances from different retrievers with reputation values; first convert the deployed retriever's score to the policy's common 0..1 scale.

# File: rag_pipeline/reputation_reranker.py
from __future__ import annotations

import math
from dataclasses import dataclass
from typing import Iterable


@dataclass(frozen=True)
class ReputationPolicy:
    version: str
    source_scores: dict[str, float]
    reputation_weight: float
    similarity_min: float
    similarity_max: float
    unknown_source_score: float = 0.0

    def validate(self) -> None:
        values = [
            self.reputation_weight,
            self.unknown_source_score,
            self.similarity_min,
            self.similarity_max,
            *self.source_scores.values(),
        ]
        if (
            not self.version
            or any(isinstance(value, bool) or not isinstance(value, (int, float)) for value in values)
            or not all(math.isfinite(float(value)) for value in values)
            or any(value < 0.0 or value > 1.0 for value in [
                self.reputation_weight, self.unknown_source_score, *self.source_scores.values()
            ])
        ):
            raise ValueError("invalid reputation policy")
        if self.similarity_max <= self.similarity_min:
            raise ValueError("invalid similarity calibration range")


def normalize_similarity(raw_score: float, policy: ReputationPolicy) -> float:
    scaled = (raw_score - policy.similarity_min) / (
        policy.similarity_max - policy.similarity_min
    )
    return max(0.0, min(1.0, scaled))


def rerank_results(
    candidates: Iterable[tuple[object, float]],
    trusted_source_ids: dict[str, str],
    policy: ReputationPolicy,
) -> tuple[list[object], list[dict]]:
    policy.validate()
    ranked = []
    evidence = []
    seen_chunk_ids: set[str] = set()
    for document, raw_similarity in candidates:
        chunk_id = document.metadata["chunk_id"]
        if not isinstance(chunk_id, str) or not chunk_id or chunk_id in seen_chunk_ids:
            raise ValueError("candidate chunk IDs must be non-empty and unique")
        seen_chunk_ids.add(chunk_id)
        if chunk_id not in trusted_source_ids:
            raise LookupError("authenticated ingestion lineage is incomplete")
        source_id = trusted_source_ids[chunk_id]
        if not isinstance(source_id, str) or not source_id:
            raise LookupError("authenticated source identity is unavailable")
        if isinstance(raw_similarity, bool) or not math.isfinite(float(raw_similarity)):
            raise ValueError("retriever returned a non-finite similarity score")
        reputation = policy.source_scores.get(source_id, policy.unknown_source_score)
        similarity = normalize_similarity(float(raw_similarity), policy)
        final_score = (
            (1.0 - policy.reputation_weight) * similarity
            + policy.reputation_weight * reputation
        )
        ranked.append((document, final_score))
        evidence.append({
            "chunk_id": chunk_id,
            "source_id": source_id,
            "raw_similarity": raw_similarity,
            "normalized_similarity": similarity,
            "reputation": reputation,
            "final_score": final_score,
            "policy_version": policy.version,
        })
    ranked.sort(key=lambda item: (-item[1], item[0].metadata["chunk_id"]))
    return [document for document, _ in ranked], evidence

Calibrate and gate the policy

Fit the similarity bounds and reputation weight on a versioned, labeled query set for the deployed embedding model, retriever, corpus, and prompt bundle. Test compromised high-similarity sources, previously unseen sources, trusted low-similarity sources, and ordinary benign queries. Promotion must meet policy-owned retrieval-quality and poison-suppression thresholds; missing lineage or an incompatible retriever version is INSUFFICIENT_DATA, not a trusted default.

Evidence

Persist the signed policy digest, connector/lineage snapshot, embedding and retriever versions, labeled-suite version, per-candidate score decomposition, ranking before and after weighting, and explicit PASS/FAIL/INSUFFICIENT_DATA/ERROR verdict. Reputation is a ranking signal, not authorization: enforce chunk entitlements separately through AID-H-020.003.

Open-source tools

LangChain (custom source-reputation fusion required)
LlamaIndex (custom source-reputation fusion required)
Vespa
OpenSearch (source-reputation field storage and function-score fusion)

Commercial / hosted tools

Cohere Rerank (similarity signal only; custom source-reputation fusion required)
Alation Data Intelligence Platform
Collibra Platform

Threat mappings

MITRE ATLAS

AML.T0099 AI Agent Tool Data Poisoning
AML.T0070 RAG Poisoning
AML.T0066 Retrieval Content Crafting
AML.T0071 False RAG Entry Injection
AML.T0130 AI Agent Response Biasing (retrieval reputation comes from authenticated lineage and signed policy rather than document-supplied trust claims)

MAESTRO

Compromised RAG Pipelines (L2)

OWASP LLM Top 10 2026

LLM05:2026 Data and Model Poisoning (source reputation lowers poisoned content retrieval influence but does not authenticate or sanitize the content)
LLM07:2026 Misinformation
LLM09:2026 Vector and Embedding Weaknesses

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

ASI06:2026 Memory & Context Poisoning (RAG context poisoning from untrusted sources gets deprioritized)