跳到正文
DEV Community· Muhammad Ikram UL Mustafa·· 3 小時前AI 評分64

Context Rot:RAG 管線為何會在沒有部署變更時逐漸退化

Context Rot: Why Your RAG Pipeline Degrades Without Anyone Deploying Anything

AI 導讀

RAG 管線即使沒有模型或提示詞變更,準確度仍可能因索引與來源文件不同步而下降。文章建議為來源內容及嵌入模型版本留存雜湊與版本資料,定期檢查過期或孤立的文本區塊,並將檢索評估與生成評估分開。對低相關或沒有結果的檢索,文章建議記錄未命中情況,以監察語料未涵蓋的用戶需求。

正文

Nobody changed the model. Nobody touched the prompt. Accuracy dropped anyway.

This is the most common production RAG failure I see, and it is the hardest to notice, because there is no deploy to correlate it with. The system that passed every test in March is quietly wrong in September.

The cause is almost always the same: your index drifted away from your source of truth, and nothing told you.

The failure nobody instruments for

A RAG pipeline has two copies of your knowledge. The source documents, and the embedded chunks derived from them. Those are updated at different times, by different code paths, and often by different people.

Every gap between them is a window where retrieval returns something that is no longer true.

Here is what actually happens over six months:

Documents get updated, chunks do not. Someone edits a policy document. The text store updates. The embedding job is queued behind something else, or failed silently three weeks ago, or was never wired up for edits because it was built for initial ingestion. Retrieval now returns the old content, confidently.

Documents get deleted, chunks survive. A product is discontinued and its documentation is removed. Nobody deletes the chunks, because deletion was never part of the ingestion path. The model is still happily answering questions about it.

New documents change the neighbourhood. You add two hundred new documents. Nothing about the old ones changed, but they are no longer in the top five for the queries they used to answer, because something more semantically similar now outranks them. Your accuracy on existing questions drops and nobody deployed anything.

The embedding model changes under you. Provider updates the model behind the same API name. New queries are embedded with the new version, your stored chunks with the old one. Now you are comparing vectors from two different spaces. This one is particularly nasty because similarity scores still come back, they are just meaningless.

Query distribution shifts. Your users start asking about things the corpus never covered well. Retrieval still returns the closest match, because that is what it does. The closest match to a question you have no answer for is simply the wrong answer with high confidence.

None of these throw an error.

Why your tests did not catch it

Most teams test retrieval once, at build time, with a set of questions written by the person who built it. That tells you the pipeline works. It tells you nothing about whether it still works.

The second problem is what gets measured. Teams evaluate the final answer, because that is what the user sees. But the answer is downstream of retrieval, so a retrieval failure and a generation failure look identical from the output. You cannot tell whether the model reasoned badly or whether it reasoned correctly over the wrong chunks.

Instrument the gap, not the output

1. Track whether chunks match their source

The cheapest useful thing you can do is hash the source content at ingestion and store it with the chunk.

import hashlib

def content_hash(text: str) -> str:
    return hashlib.sha256(text.encode()).hexdigest()

async def ingest_document(doc: Document, tenant_id: UUID):
    doc_hash = content_hash(doc.content)

    existing = await db.fetchval(
        "SELECT source_hash FROM doc_chunks "
        "WHERE document_id = $1 LIMIT 1",
        doc.id
    )

    if existing == doc_hash:
        return  # nothing changed, skip re-embedding

    chunks = chunk_text(doc.content)
    embeddings = await embed_batch(chunks)

    async with db.transaction():
        await db.execute(
            "DELETE FROM doc_chunks WHERE document_id = $1", doc.id
        )
        for chunk, embedding in zip(chunks, embeddings):
            await db.execute(
                "INSERT INTO doc_chunks "
                "(document_id, tenant_id, content, embedding, "
                " source_hash, embedding_model, indexed_at) "
                "VALUES ($1, $2, $3, $4, $5, $6, now())",
                doc.id, tenant_id, chunk, embedding,
                doc_hash, EMBEDDING_MODEL
            )

Three things worth noting. The hash check means re-ingestion is idempotent, so you can run it on a schedule without re-embedding everything. The delete and insert are in one transaction, so a query cannot land in a window where the old chunks are gone and the new ones are not written yet. And embedding_model is stored per row, which matters for the next part.

2. Find the drift

Once you have the hash and the model version stored, staleness becomes a query rather than a mystery.

-- chunks whose source document has since changed
SELECT d.id, d.title, c.indexed_at
FROM documents d
JOIN doc_chunks c ON c.document_id = d.id
WHERE c.source_hash != encode(sha256(d.content::bytea), 'hex')
GROUP BY d.id, d.title, c.indexed_at;

-- orphaned chunks: source document is gone
SELECT DISTINCT c.document_id
FROM doc_chunks c
LEFT JOIN documents d ON d.id = c.document_id
WHERE d.id IS NULL;

-- chunks embedded with a different model than you are querying with
SELECT embedding_model, count(*)
FROM doc_chunks
GROUP BY embedding_model;

That last one should return exactly one row. If it returns two, you have vectors from two different embedding spaces in the same index and your similarity scores are not comparable.

Run these on a schedule and alert on them. They are cheap and they catch the majority of real drift.

3. Evaluate retrieval separately from generation

This is the change that matters most, and it is the one teams skip.

Build a set of questions where you know which document should be retrieved. Not which answer is correct, which source is correct. Then measure whether it still comes back.

RETRIEVAL_CASES = [
    {
        "query": "what is the refund window for annual plans",
        "must_retrieve": "billing-policy-v4",
    },
    {
        "query": "how do I rotate an API key",
        "must_retrieve": "api-authentication",
    },
]

async def check_retrieval(tenant_id: UUID) -> list[dict]:
    failures = []
    for case in RETRIEVAL_CASES:
        results = await search(case["query"], tenant_id=tenant_id, limit=5)
        retrieved_docs = {r["document_slug"] for r in results}

        if case["must_retrieve"] not in retrieved_docs:
            failures.append({
                "query": case["query"],
                "expected": case["must_retrieve"],
                "got": list(retrieved_docs),
            })
    return failures

Run this nightly. When it starts failing, you know the problem is retrieval and not the model, which is half the debugging done before you have opened anything.

The useful property of this test is that it fails for reasons that have nothing to do with your code. New documents outranking old ones, an embedding model change, a chunking tweak that seemed harmless. Those are exactly the changes that have no deploy attached.

4. Treat an empty or weak retrieval as a real state

If nothing crosses your similarity threshold, that is information. Most pipelines throw it away and send whatever came back anyway.

MIN_SIMILARITY = 0.35

async def answer(query: str, tenant_id: UUID):
    results = await search(query, tenant_id=tenant_id, limit=6)
    relevant = [r for r in results if r["similarity"] >= MIN_SIMILARITY]

    if not relevant:
        await log_retrieval_miss(query, tenant_id)
        return "I don't have information about that."

    return await generate(query, context=relevant)

Those logged misses are the most valuable dataset you will build. They are a direct record of what your users want and your corpus does not cover, and the rate of them climbing over time is drift showing up in your metrics before it shows up in complaints.

The shape of the problem

A RAG system is not a feature you ship. It is two stores that have to agree, updated on different schedules by different code, with a model in front making whatever they return sound reasonable.

Everything above is the same idea in different places: make the disagreement visible. Hash so you can detect it, version so you can detect it, test retrieval so you can detect it, log misses so you can detect it.

Because the default behaviour of this failure is a well-written answer, delivered confidently, that happens to be six months out of date. Nothing in your stack is going to tell you about it.


I write about production AI systems and what breaks in them. More at ikramulmustafa.com, and a RAG agent with documented failure handling at github.com/ikramulmustafa/n8n-rag-lead-agent.

來源:DEV Community · dev.to