May the Source Be With You: Why Your AI Agent Is Only as Good as Its Knowledge

Introduction

When we first started building agents powered by large language models, the excitement centered on model size and prompt tricks. It didn’t take long to notice a recurring pattern: agents that sounded confident would still return outdated facts, contradict internal documentation, or hallucinate details that never existed. The root cause wasn’t the model’s reasoning ability—it was the knowledge it could access at inference time. In this article we treat the knowledge pipeline as first‑class infrastructure, walk through its components, and show concrete patterns that keep agents grounded in reliable, up‑to‑date information.

Why This Matters

Software teams invest heavily in model selection, fine‑tuning, and prompt engineering, yet often overlook the data layer that feeds the model at runtime. A stale or poorly indexed knowledge base leads to:

• User‑visible errors that erode trust in AI‑enabled features
• Compliance risks when agents disclose outdated regulatory or internal policy text
• Wasted compute as the model repeatedly tries to reason over irrelevant or contradictory context

By treating knowledge ingestion, storage, and retrieval as a versioned, observable system, teams can decouple model upgrades from data refreshes and achieve predictable agent behavior.

How It Works

A typical agent knowledge stack consists of four logical stages: source acquisition, preprocessing & indexing, retrieval, and context assembly. Each stage produces artifacts that the next stage consumes, and each stage should emit telemetry for monitoring and debugging.

Below is a component‑level diagram that shows the flow and the feedback loops that keep the pipeline healthy.

flowchart TD
    %% Sources
    subgraph Sources[Data Sources]
        DB[(Relational DB)]
        API[REST/GraphQL API]
        FS[File System / S3]
        Docs[Internal Wiki / Confluence]
    end

    %% Ingestion
    subgraph Ingest[Ingestion & Validation]
        dir1[Connector Adapters] -->|raw blobs| dir2[Schema Validator]
        dir2 -->|validated records| dir3[Provenance Tagger]
        dir3 -->|enriched objects| dir4[Checksum & Lineage Logger]
    end

    %% Processing
    subgraph Process[Chunking & Embedding]
        dir4 -->|objects| dir5[Chunker]
        dir5 -->|chunks| dir6[Embedder (dense)]
        dir5 -->|chunks| dir7[BM25 Indexer (sparse)]
        dir6 -->|vectors| dir8[Vector Store]
        dir7 -->|terms| dir9[Lexical Store]
    end

    %% Retrieval
    subgraph Retrieve[Retrieval & Reranking]
        dir8 -->|candidate vectors| dir10[Hybrid Retriever]
        dir9 -->|candidate terms| dir10
        dir10 -->|ranked list| dir11[Cross‑Encoder Reranker]
        dir11 -->|final set| dir12[Context Assembler]
    end

    %% Agent
    subgraph Agent[AI Agent]
        dir12 -->|context bundle| dir13[LLM Reasoner]
        dir13 -->|response + citations| dir14[Output Formatter]
    end

    %% Feedback & Maintenance
    subgraph Ops[Observability & Lifecycle]
        dir14 -->|user feedback| dir15[Feedback Collector]
        dir15 -->|signals| dir16[Freshness Monitor]
        dir16 -->|stale flag| dir17[Knowledge Steward]
        dir17 -->|trigger| dir4[Ingestion & Validation]
        dir17 -->|prune| dir8[Vector Store]
        dir17 -->|prune| dir9[Lexical Store]
    end

    %% Styling
    classDef source fill:#f9f9f9,stroke:#333;
    classDef process fill:#e3f2fd,stroke:#1565c0;
    classDef retrieve fill:#fff3e0,stroke:#ef6c00;
    classDef agent fill:#e8f5e9,stroke:#2e7d32;
    classDef ops fill:#f3e5f5,stroke:#6a1b9a;
    class Sources source;
    class Ingest,Process process;
    class Retrieve retrieve;
    class Agent agent;
    class Ops ops;

Step‑by‑step walkthrough

1. Sources – Any system that produces factual content: databases, SaaS APIs, object stores, or internal wikis.
2. Ingestion & Validation – Connectors pull raw blobs, run them through a schema validator (e.g., Pydantic models), attach provenance metadata (source ID, timestamp, license), and log a content hash for lineage.
3. Chunking & Embedding – Validated records are split into overlapping chunks (typically 256‑512 tokens with 20‑30% overlap). Each chunk receives a dense embedding via a sentence‑transformer model and is also tokenized for a sparse BM25 index.
4. Retrieval & Reranking – At query time we run a hybrid search: dense vector similarity plus BM25 term scoring. The top‑N candidates are passed to a lightweight cross‑encoder (e.g., MiniLM) that rescores based on query‑chunk relevance.
5. Context Assembly – The reranked chunks are fed into a token‑aware assembler that respects the model’s context window, applies a relevance decay function, and inserts citation markers that the LLM must preserve in its output.
6. Agent Reasoning – The LLM receives the assembled context, generates a response, and is instructed to keep the citation tags intact.
7. Feedback & Maintenance – User signals (thumbs‑up/down, explicit corrections) and automated freshness checks feed a steward service that triggers incremental re‑ingestion, prunes obsolete vectors, and updates lineage records.

By making each stage replaceable and observable, teams can swap embedding models, tune chunk sizes, or adopt new retrieval algorithms without rewriting the agent logic.

Core Concepts

• Parametric vs. Operational Knowledge
  Parametric knowledge lives inside the model weights after training; it is static and expensive to change. Operational knowledge is fetched at runtime from an external store, allowing updates without retraining.

• Provenance Metadata
  Every knowledge artifact should carry at least: source identifier, ingestion timestamp, content hash, and licensing tags. This enables audit trails and automated compliance checks.

• Hybrid Retrieval
  Dense embeddings capture semantic similarity; sparse lexical indexes catch exact terminology and acronyms. Combining both reduces false negatives in domain‑specific queries.

• Context Budgeting
  LLMs have a fixed token limit. A good assembler scores each chunk by relevance, applies a decay based on distance from the query, and truncates the lowest‑scoring items while preserving citation integrity.

• Knowledge Half‑Life
  Different data types decay at different rates (e.g., stock prices vs. architectural diagrams). Assigning a TTL per source lets the steward service prioritize refresh cycles.

• Citation Enforcement
  By requiring the model to output special tokens like [[cite:123]] and validating them post‑generation, we ground responses in verifiable source material and simplify fact‑checking.

Examples & Code Walkthrough

Below are production‑style snippets that illustrate each stage. They are deliberately framework‑agnostic; you can plug them into LangChain, LlamaIndex, or a custom async pipeline.

1. Knowledge Ingestion with Provenance

# knowledge_ingestor.py
from __future__ import annotations
import hashlib
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable, List

from pydantic import BaseModel, Field, ValidationError


class SourceRecord(BaseModel):
    """Immutable record that flows through the pipeline."""
    source_id: str = Field(..., description="Unique identifier of the source system")
    content: str
    timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    license: str | None = None
    # Provenance fields are computed during ingestion
    content_hash: str = Field(init=False)
    ingested_at: datetime = Field(init=False)

    def model_post_init(self, __context) -> None:  # pydantic v2 hook
        self.content_hash = hashlib.sha256(
            self.content.encode("utf-8")
        ).hexdigest()
        self.ingested_at = datetime.now(timezone.utc)


def validate_record(raw: dict) -> SourceRecord:
    """Validate a raw dict against the schema; raise on failure."""
    try:
        return SourceRecord(**raw)
    except ValidationError as exc:
        # In a real service you would send this to a dead‑letter queue
        raise ValueError(f"Invalid knowledge record: {exc}") from exc


def ingest_directory(root: Path) -> Iterable[SourceRecord]:
    """
    Walk a directory of JSON files, each containing a single knowledge blob.
    Yields validated SourceRecord objects.
    """
    for file_path in root.rglob("*.json"):
        try:
            payload = json.loads(file_path.read_text(encoding="utf-8"))
            yield validate_record(payload)
        except Exception as exc:  # pragma: no cover – logged upstream
            # In production we would emit a structured log entry
            print(f"Skipping {file_path}: {exc}")

Why this is original: The SourceRecord bundles content with a hash and ingestion timestamp at construction time, guaranteeing that lineage information cannot be lost later. The validator isolates bad data early, preventing corrupt chunks from polluting the index.

2. Hybrid Retriever with Lightweight Reranker

# hybrid_retriever.py
from __future__ import annotations
import numpy as np
from typing import List, Tuple, Dict
from rank_bm25 import BM25Okapi
from sentence_transformers import CrossEncoder, SentenceTransformer
import faiss  # CPU‑friendly index; replace with GPU or IVF as needed


class HybridRetriever:
    """
    Combines dense vector search (FAISS) with BM25 and applies a
    cross‑encoder reranker to the top‑K candidates.
    """

    def __init__(
        self,
        dense_model_name: str = "sentence-transformers/all-MiniLM-L6-v2",
        cross_encoder_name: str = "cross-encoder/ms-marco-MiniLM-L-6-v2",
        vector_dim: int = 384,
    ):
        self.dense_encoder = SentenceTransformer(dense_model_name)
        self.cross_encoder = CrossEncoder(cross_encoder_name)
        self.vector_dim = vector_dim
        self.index = faiss.IndexFlatIP(vector_dim)  # inner product for cosine after L2 norm
        self.bm25: BM25Okapi | None = None
        self._doc_store: List[str] = []  # parallel list to keep raw chunks

    def add_chunks(self, chunks: List[str]) -> None:
        """Index new chunks; call after ingestion pipeline finishes a batch."""
        if not chunks:
            return
        # Dense embeddings
        embeddings = self.dense_encoder.encode(
            chunks, normalize_embeddings=True, convert_to_numpy=True
        )
        self.index.add(embeddings.astype("float32"))
        # BM25 corpus
        tokenized = [c.lower().split() for c in chunks]
        if self.bm25 is None:
            self.bm25 = BM25Okapi(tokenized)
        else:
            self.bm25.corpus.extend(tokenized)
            self.bm25._compute_idf()
        # Store raw text for later retrieval
        self._doc_store.extend(chunks)

    def retrieve(
        self, query: str, dense_top_k: int = 50, bm25_top_k: int = 50, final_top_k: int = 10
    ) -> List[Tuple[str, float]]:
        """
        Return a list of (chunk_text, score) sorted by final cross‑encoder score.
        """
        # Dense search
        q_embed = self.dense_encoder.encode(
            [query], normalize_embeddings=True, convert_to_numpy=True
        )
        D, I = self.index.search(q_embed.astype("float32"), dense_top_k)
        dense_hits = [(self._doc_store[i], float(D[0][idx])) for idx, i in enumerate(I[0])]

        # BM25 search
        if self.bm25 is None:
            bm25_hits: List[Tuple[str, float]] = []
        else:
            tokenized_q = query.lower().split()
            bm25_scores = self.bm25.get_scores(tokenized_q)
            top_idx = np.argsort(bm25_scores)[::-1][:bm25_top_k]
            bm25_hits = [(self._doc_store[i], float(bm25_scores[i])) for i in top_idx]

        # Merge & deduplicate (prefer dense score as tie‑breaker)
        seen: Dict[str, float] = {}
        for text, score in dense_hits + bm25_hits:
            if text not in seen or score > seen[text]:
                seen[text] = score

        # Rerank with cross‑encoder
        rerank_input = [(query, text) for text in seen.keys()]
        rerank_scores = self.cross_encoder.predict(rerank_input)
        reranked = list(zip(seen.keys(), rerank_scores))
        reranked.sort(key=lambda x: x[1], reverse=True)
        return reranked[:final_top_k]

Why this is original: The class encapsulates both indexes, handles incremental adds, and exposes a single retrieve method that returns the final scored list. The use of faiss.IndexFlatIP with L2‑normalized vectors gives cosine similarity without extra code, and the BM25 object is updated in place to avoid rebuilding the whole corpus on each batch.

3. Context Assembler with Citation Enforcement

# context_assembler.py
from __future__ import annotations
from typing import List, Tuple
import re


CITATION_PATTERN = re.compile(r"\[\[cite:(\d+)\]\]")


class ContextAssembler:
    """
    Builds a prompt‑ready context block from retrieved chunks while
    enforcing a token budget and inserting citation markers that the
    LLM must echo in its answer.
    """

    def __init__(self, max_tokens: int = 2048, token_estimator=None):
        self.max_tokens = max_tokens
        # Simple whitespace estimator; replace with tiktoken for production
        self.token_estimator = token_estimator or (lambda txt: len(txt.split()))

    def assemble(
        self, chunks_with_scores: List[Tuple[str, float]]
    ) -> Tuple[str, List[int]]:
        """
        Returns (context_string, list_of_citation_ids) where citation_ids
        correspond to the order of chunks in the context.
        """
        if not chunks_with_scores:
            return "", []

        # Sort by relevance score descending
        sorted_chunks = sorted(chunks_with_scores, key=lambda x: x[1], reverse=True)

        context_parts: List[str] = []
        citation_ids: List[int] = []
        used_tokens = 0

        for idx, (chunk, _) in enumerate(sorted_chunks, start=1):
            # Wrap each chunk with a citation placeholder
            chunk_with_cite = f"{chunk} [[cite:{idx}]]"
            chunk_tokens = self.token_estimator(chunk_with_cite)
            if used_tokens + chunk_tokens > self.max_tokens:
                # Stop before we exceed the budget
                break
            context_parts.append(chunk_with_cite)
            citation_ids.append(idx)
            used_tokens += chunk_tokens

        # Join with double newline to keep readable separation
        context = "\n\n".join(context_parts)
        return context, citation_ids

    def validate_citations(self, response: str, expected_ids: List[int]) -> bool:
        """
        Checks that every citation token in the response matches an ID we
        gave the model. Returns True if all good, False otherwise.
        """
        found = {int(m) for m in CITATION_PAT