我要投稿

爆改RAG！Relevant Segment Extraction（RSE）让你的AI检索“有头有尾”，不再碎片化

发布日期：2025-07-16 06:44:31 浏览次数： 1710

作者：许泽宇的技术分享

微信搜一搜，关注“许泽宇的技术分享”

你还在用传统RAG（Retrieval-Augmented Generation）检索一堆“东一榔头西一棒槌”的文本碎片，然后让大模型自己拼拼凑凑？别闹了，AI都快被你玩成拼图大师了！今天，咱们聊聊如何用Relevant Segment Extraction（RSE）让你的RAG系统“上下文连贯”，让大模型如虎添翼，输出更靠谱、更有逻辑的答案！

1. 传统RAG的“碎片化”困境

先来个灵魂拷问：你见过大模型被一堆无头无尾的文本块折磨得语无伦次吗？

传统RAG的套路是这样的：

把文档切成一堆小块（chunk），每块几百字。
用户提问后，检索最相关的Top-K块，拼成上下文，扔给大模型。
大模型努力“脑补”上下文，输出答案。

看似合理，实则暗藏杀机——相关内容往往是连续的，但你检索出来的却是“东一块西一块”，上下文断裂，信息丢失，模型理解难度飙升。

举个栗子：你问“什么是可解释AI？为什么重要？”
传统RAG可能给你拼出：

第5块：XAI定义
第12块：XAI在医疗的应用
第27块：XAI的挑战
第33块：XAI的未来展望

模型一看：这都啥啊？我得自己把这些碎片拼成一篇论文，累不累！

2. RSE：让检索“成段成片”，上下文更连贯

Relevant Segment Extraction（RSE）来了！它的核心思想很简单：

❝
相关内容往往在文档中是连续的，咱们就应该“整段整段”地检索出来，别再让模型拼拼凑凑了！

RSE的流程如下：

文档切块：还是老套路，先切成小块。
相关性打分：每块和用户问题算相关性分数。
连续片段提取：用“最大子段和”算法，找出一段段连续的高分块，拼成完整的上下文片段。
上下文重构：把这些片段拼起来，作为大模型的输入。

这样，模型拿到的上下文是“有头有尾”的连续内容，理解起来轻松多了，输出自然更靠谱！

3. RSE全流程实战拆解

3.1 PDF文本提取

先用PyMuPDF（fitz）把PDF里的内容全都抽出来：

import fitz

def extract_text_from_pdf(pdf_path):
    mypdf = fitz.open(pdf_path)
    all_text = ""
    for page_num in range(mypdf.page_count):
        page = mypdf[page_num]
        text = page.get_text("text")
        all_text += text
    return all_text

3.2 文本切块（Chunking）

切块的艺术：每块800字，不重叠，方便后续连续片段重构。

def chunk_text(text, chunk_size=800, overlap=0):
    chunks = []
    for i in range(0, len(text), chunk_size - overlap):
        chunk = text[i:i + chunk_size]
        if chunk:
            chunks.append(chunk)
    return chunks

3.3 向量化与向量库

用OpenAI API（或其他embedding模型）把每个chunk变成向量，存进自建的SimpleVectorStore：

class SimpleVectorStore:
    def __init__(self, dimension=1536):
        self.dimension = dimension
        self.vectors = []
        self.documents = []
        self.metadata = []

    def add_documents(self, documents, vectors=None, metadata=None):
        if vectors is None:
            vectors = [None] * len(documents)
        if metadata is None:
            metadata = [{} for _ in range(len(documents))]
        for doc, vec, meta in zip(documents, vectors, metadata):
            self.documents.append(doc)
            self.vectors.append(vec)
            self.metadata.append(meta)

    def search(self, query_vector, top_k=5):
        import numpy as np
        query_array = np.array(query_vector)
        similarities = []
        for i, vector in enumerate(self.vectors):
            if vector is not None:
                similarity = np.dot(query_array, vector) / (
                    np.linalg.norm(query_array) * np.linalg.norm(vector)
                )
                similarities.append((i, similarity))
        similarities.sort(key=lambda x: x[1], reverse=True)
        results = []
        for i, score in similarities[:top_k]:
            results.append({
                "document": self.documents[i],
                "score": float(score),
                "metadata": self.metadata[i]
            })
        return results

3.4 相关性打分

每个chunk和query算相关性分数，减去不相关惩罚，得到“chunk value”：

def calculate_chunk_values(query, chunks, vector_store, irrelevant_chunk_penalty=0.2):
    query_embedding = create_embeddings([query])[0]
    num_chunks = len(chunks)
    results = vector_store.search(query_embedding, top_k=num_chunks)
    relevance_scores = {result["metadata"]["chunk_index"]: result["score"] for result in results}
    chunk_values = []
    for i in range(num_chunks):
        score = relevance_scores.get(i, 0.0)
        value = score - irrelevant_chunk_penalty
        chunk_values.append(value)
    return chunk_values

3.5 连续片段提取（RSE核心算法）

用“最大子段和”思想，找出一段段连续的高分块，保证每段不超过20块，总共不超过30块：

def find_best_segments(chunk_values, max_segment_length=20, total_max_length=30, min_segment_value=0.2):
    best_segments = []
    segment_scores = []
    total_included_chunks = 0
    while total_included_chunks < total_max_length:
        best_score = min_segment_value
        best_segment = None
        for start in range(len(chunk_values)):
            if any(start >= s[0] and start < s[1] for s in best_segments):
                continue
            for length in range(1, min(max_segment_length, len(chunk_values) - start) + 1):
                end = start + length
                if any(end > s[0] and end <= s[1] for s in best_segments):
                    continue
                segment_value = sum(chunk_values[start:end])
                if segment_value > best_score:
                    best_score = segment_value
                    best_segment = (start, end)
        if best_segment:
            best_segments.append(best_segment)
            segment_scores.append(best_score)
            total_included_chunks += best_segment[1] - best_segment[0]
        else:
            break
    best_segments = sorted(best_segments, key=lambda x: x[0])
    return best_segments, segment_scores

3.6 片段重构与格式化

把连续的chunk拼成完整的上下文片段，格式化给大模型：

def reconstruct_segments(chunks, best_segments):
    reconstructed_segments = []
    for start, end in best_segments:
        segment_text = " ".join(chunks[start:end])
        reconstructed_segments.append({
            "text": segment_text,
            "segment_range": (start, end),
        })
    return reconstructed_segments

def format_segments_for_context(segments):
    context = []
    for i, segment in enumerate(segments):
        segment_header = f"SEGMENT {i+1} (Chunks {segment['segment_range'][0]}-{segment['segment_range'][1]-1}):"
        context.append(segment_header)
        context.append(segment['text'])
        context.append("-" * 80)
    return "\n\n".join(context)

3.7 大模型生成答案

把格式化好的上下文和问题一起扔给大模型，生成最终答案：

def generate_response(query, context, model="meta-llama/Llama-3.2-3B-Instruct"):
    system_prompt = """You are a helpful assistant that answers questions based on the provided context.
    The context consists of document segments that have been retrieved as relevant to the user's query.
    Use the information from these segments to provide a comprehensive and accurate answer.
    If the context doesn't contain relevant information to answer the question, say so clearly."""
    user_prompt = f"""
Context:
{context}

Question: {query}

Please provide a helpful answer based on the context provided.
"""
    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ],
        temperature=0
    )
    return response.choices[0].message.content