首页 / 文章 / 实用指南：如何修复不断检索到错误上下文的RAG系统

machine-learning python llm rag agents architecture

实用指南：如何修复不断检索到错误上下文的RAG系统

《实用笔记》操作指南：如何修复不断检索错误上下文的RAG系统——为采用该模式的团队提供的合同、检查清单及即用代码模板。

2332 词

可将此内容作为《如何修复不断检索错误上下文的RAG系统》一文中理念的面向操作员的简化版本：清晰的阶段划分、有序的代码模块，以及便于交接时使用的恢复说明。在“概览”阶段，若能将其视为可量化的界面来处理效果最佳。在扩大范围之前，先记录一份理想的操作日志、一个故障案例以及回滚说明。相较于庞大的脚本，应优先使用小型且可测试的单元。当某一步骤出现故障时，故障原因应能明确指向某个特定责任模块，而非复杂的流程链。

你需要一个可复现的故障案例

若需要失败阶段，应在修改代码之前明确输入参数、该步骤的负责人以及退出标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。将此阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。需引用实际作为答案依据的段落。没有引用的话，操作人员就无法区分是幻觉还是索引缺失所致。

chunks = [
    {
        "id": "audit_03",
        "source": "audit-logs",
        "text": (
            "Enterprise audit logs are retained for 365 days "
            "before automatic deletion."
        ),
    },
    {
        "id": "errors_07",
        "source": "api-errors",
        "text": (
            "NX-204 means the requested resource exists but is not "
            "available in the caller's current region."
        ),
    },
    {
        "id": "exports_01",
        "source": "csv-exports",
        "text": (
            "CSV exports run asynchronously and appear in the exports "
            "panel when processing completes."
        ),
    },
    {
        "id": "exports_04",
        "source": "csv-exports",
        "text": (
            "A completed CSV download link remains active for seven days."
        ),
    },
]

eval_cases = [
    {
        "query": "How long are enterprise audit logs kept?",
        "relevant": {"audit_03"},
    },
    {
        "query": "What does error NX-204 mean?",
        "relevant": {"errors_07"},
    },
    {
        "query": "How long is a CSV export link usable?",
        "relevant": {"exports_04"},
    },
]

你最初使用的是一个刻意设计得较为简单的检索器

对于从某个阶段开始的任务，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解这些成本信息，可避免在任务从演示环境转移到共享环境时出现意外费用。必须注明实际作为答案依据的段落；没有引用的话，操作人员就无法区分是幻觉内容还是索引缺失导致的错误。

import numpy as np

from sklearn.decomposition import TruncatedSVD
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
from sklearn.preprocessing import normalize


class LsaRetriever:
    def __init__(self, chunks, dims=16):
        self.chunks = chunks

        self.tfidf = TfidfVectorizer(
            stop_words="english",
            ngram_range=(1, 2),
            sublinear_tf=True,
        )

        term_matrix = self.tfidf.fit_transform(
            chunk["text"] for chunk in chunks
        )

        # The corpus is tiny. SVD doesn't need dimensions it cannot use.
        dims = min(
            dims,
            term_matrix.shape[0] - 1,
            term_matrix.shape[1] - 1,
        )

        if dims < 1:
            raise ValueError("Need more text to build the LSA index.")

        self.svd = TruncatedSVD(
            n_components=dims,
            random_state=0,
        )

        self.index = normalize(
            self.svd.fit_transform(term_matrix)
        )

    def search(self, query, limit=None):
        query_vec = self.tfidf.transform([query])
        query_vec = normalize(self.svd.transform(query_vec))

        similarity = cosine_similarity(
            query_vec,
            self.index,
        )[0]

        ranked = np.argsort(similarity)[::-1]

        if limit is not None:
            ranked = ranked[:limit]

        return [
            (self.chunks[i], float(similarity[i]))
            for i in ranked
        ]

1. 0.879  exports_01
   CSV exports run asynchronously and appear in the exports panel...

2. 0.843  exports_02
   Large exports are split into multiple compressed files.

3. 0.830  exports_03
   Users can cancel an export while it is still queued...

4. 0.772  exports_04
   A completed CSV download link remains active for seven days.

最简单的调试工具反而最为实用

在最基础的调试阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审计。需引用实际作为答案依据的段落。如果没有引用，操作人员就无法区分是虚假信息还是索引缺失导致的错误。在最基础的调试阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相较于庞大的脚本，应优先使用小型且可测试的单元。当某个步骤失败时，故障原因应能指向单一的责任主体而非多个部分。

总比混乱的管道要好。

def show_hits(retriever, query, limit=5):
    print(f"\n{query}\n")

    for position, (chunk, score) in enumerate(
        retriever.search(query, limit),
        start=1,
    ):
        print(
            f"{position:>2}. {score:.3f}  "
            f"{chunk['id']} ({chunk['source']})"
        )
        print(f"    {chunk['text']}\n")

你不想让评估依赖于精确的措辞

在处理该阶段时，首先写下契约：所需的输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。将这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功检测标准，并拒绝默许的部分完成。在调整提示词之前，先在固定的问题集上测试召回率。仅仅更换提示词很难改善较差的检索效果。

if answer_hint in chunk["text"]:
    ...

def evaluate_retriever(retriever, cases, k=3):
    recall_scores = []
    reciprocal_ranks = []

    for case in cases:
        hits = retriever.search(case["query"])
        relevant = case["relevant"]

        relevant_positions = [
            position
            for position, (chunk, _) in enumerate(hits, start=1)
            if chunk["id"] in relevant
        ]

        found_in_top_k = sum(
            position <= k
            for position in relevant_positions
        )

        recall_scores.append(
            found_in_top_k / len(relevant)
        )

        reciprocal_ranks.append(
            1 / relevant_positions[0]
            if relevant_positions
            else 0.0
        )

    return {
        f"recall@{k}": float(np.mean(recall_scores)),
        "mrr": float(np.mean(reciprocal_ranks)),
    }

然后你就把问题归咎于分块处理

在处理“Then you blamed chunking”阶段时，首先写下相关约定：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免从演示环境过渡到共享环境时出现意外费用。在调整提示词之前，先用固定的问题集测试召回率。仅仅更换提示词很难改善较差的检索效果。

chunk_size = 500
chunk_overlap = 50

BM25让实验结果略显尴尬

在实现 BM25 实验阶段时，首先需明确相关规范：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持透明可追溯。应将配置信息与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中管理，以便操作人员无需查看全部代码即可进行审计。在调整提示词之前，先使用固定的问题集来衡量召回率。仅仅更换提示词往往无法解决检索效果不佳的问题。在实现 BM25 实验阶段时，首先需明确相关规范：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持透明可追溯。相比庞大的脚本，更应优先使用小型且易于测试的单元。当某个步骤出现故障时，故障点应能明确指向单一责任模块，而非复杂的流程链。

你仍然需要这两种信号

将这两类阶段工作视为可度量的对象处理时效果最佳。在扩大范围之前，记录一个成功的案例、一个失败案例以及回滚说明。把这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，绝不允许默许部分完成的情况。将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应强制要求重写另一项。

from collections import defaultdict


def fuse_rankings(vector_hits, bm25_hits, rrf_k=60):
    fused = defaultdict(float)
    chunks_by_id = {}

    for hits in (vector_hits, bm25_hits):
        for rank, (chunk, _) in enumerate(hits, start=1):
            chunk_id = chunk["id"]
            chunks_by_id[chunk_id] = chunk
            fused[chunk_id] += 1 / (rrf_k + rank)

    ranked_ids = sorted(
        fused,
        key=fused.get,
        reverse=True,
    )

    return [
        (chunks_by_id[chunk_id], fused[chunk_id])
        for chunk_id in ranked_ids
    ]

def hybrid_search(query, lsa, bm25, candidate_k=20):
    vector_hits = lsa.search(query, limit=candidate_k)
    bm25_hits = bm25.search(query, limit=candidate_k)

    return fuse_rankings(vector_hits, bm25_hits)

重新排序是您最后测试的功能

重新排序是最后阶段的处理方式，若将其视为可测量的指标则效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。在功能结果旁还需记录处理时间以及令牌或查询成本。提前了解这些成本信息，就能避免在系统从演示环境转向共享环境时出现意外费用。应将分块策略与检索策略分开，当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。

def cheap_local_rerank(query, candidates, limit=5):
    """
    Good enough for this experiment.
    I'd use a learned reranker for a real deployment.
    """
    candidate_text = [
        chunk["text"]
        for chunk, _ in candidates
    ]

    tfidf = TfidfVectorizer(
        analyzer="char_wb",
        ngram_range=(3, 5),
        min_df=1,
    )

    matrix = tfidf.fit_transform(
        [query, *candidate_text]
    )

    relevance = cosine_similarity(
        matrix[0],
        matrix[1:],
    )[0]

    reranked = sorted(
        zip(candidates, relevance),
        key=lambda row: row[1],
        reverse=True,
    )

    return [
        (chunk, float(score))
        for ((chunk, _), score) in reranked[:limit]
    ]

最终的数值并不如你预期的那么有趣

最终数值在作为可度量的指标时最适用于阶段测试。在扩大范围之前，需记录一份最佳案例、一个故障实例以及回滚说明。将配置置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放，以便操作人员无需查看整个系统结构即可进行审计。将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。最终数值在作为可度量的指标时最适用于阶段测试。在扩大范围之前，需记录一份最佳案例、一个故障实例以及回滚说明。相比庞大的脚本，应优先使用小型且可测试的单元。当某个步骤出现故障时，故障点应指向单一责任模块，而非复杂的处理流程。

返回 CSV 查询

在“返回 CSV”阶段，应在修改代码之前明确输入内容、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。将此阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。需引用实际作为答案依据的段落。没有引用的话，操作人员就无法区分是幻觉还是索引缺失所致。

How long is a CSV export link usable?

1. CSV exports run asynchronously...
2. Large exports are split...
3. Users can cancel an export...
4. A completed CSV download link remains active for seven days.

1. A completed CSV download link remains active for seven days.
2. Users can cancel an export while it is still queued...
3. CSV exports run asynchronously...

现在的调试流程简单多了

在调试时，应先明确各阶段的输入内容、负责该步骤的人员以及退出标准，然后再修改代码。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解这些成本信息，可避免在从演示环境过渡到共享环境时出现意外费用。必须注明支撑答案的具体内容，否则操作人员就无法区分是虚假信息还是索引缺失导致的错误。

最终思考与结论

在最终思考与总结阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放于一个操作人员可以审核的位置，无需阅读整个系统结构。需引用实际作为答案依据的段落。没有引用的话，操作人员就无法区分是幻觉内容还是索引缺失导致的错误。在最终思考与总结阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相较于庞大的脚本，更应优先使用小型且可测试的单元。当某个步骤失败时，故障原因应指向单一责任模块而非多个部分。

复杂的流程链。

操作检查清单

在制定操作检查清单时，首先明确合同要求：所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。

同时记录正常流程和故障恢复流程。重试机制、人工审核环节以及错误处理方式都是产品本身的组成部分，而非后续的优化内容。

在调整提示词之前，先使用固定的问题集来衡量检索效果。仅仅更换提示词很难解决检索能力不足的问题。

锁定依赖项的版本，并记录用于演示的图像摘要。可重复性比经验知识更为重要。

优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤出现故障时，故障原因应能明确指向某个具体的功能模块，而非整个复杂的流程链。

在调整提示词之前，先使用固定的问题集测试召回率。仅仅更换提示词很难改善较差的检索效果。

在升级整个系统之前，先冻结现有版本，为关键流程记录标准文本样本，并明确回滚步骤。共享环境需要设置访问频率限制、租户验证机制，以及明确的密钥轮换负责人。与其追求华丽的临时演示，不如注重扎实的稳定性。

关于4527c294eba8的批注：请将服务提供商密钥存放在仓库之外，设定单次会话的令牌上限，并将文本样本与评估用文件放在一起，以便后续更换模型时仍能保持对比一致性。

在处理强化措施的第0阶段时，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。应将配置信息与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中于一个位置，这样操作人员无需查看整个系统结构即可进行审计。

强化措施细节0/820：需测量该措施的执行时间、错误类型以及令牌消耗情况，然后根据固定的评估标准而非个人经验来决定是否保留该修改。

全部文章发送简报