使用 RAG 掌握查询应答：克服大规模会议数据中的关键挑战-Python教程-PHP中文网

在信息过载的数字时代，从大型数据集中提取可行的见解比以往任何时候都更加重要。最近，我踏上了利用检索增强生成 (RAG) 来解决一项重大挑战的旅程——从大量会议记录中提供准确的答案。本博客探讨了将我的基于 RAG 的查询应答系统转变为从非结构化会议数据中提取见解的强大工具的障碍、解决方案和成就。

问题陈述：使用 RAG 进行查询应答的挑战
主要挑战之一是构建一个能够在庞大的会议记录存储库中处理复杂的、特定于意图的查询的系统。传统的 RAG 查询应答模型经常返回不相关或不完整的信息，无法捕捉用户意图。会议数据的非结构化性质与多样化的查询类型相结合，需要更精细的解决方案。

初步方法：为有效的查询应答奠定基础
我从一个基础 RAG 模型开始，该模型旨在将检索和响应生成结合起来。最初使用的两种技术是：

分块：按句子边界将大文档分成更小的片段，通过缩小搜索范围来改进检索。
嵌入和向量存储：分块后，每个片段都被嵌入并存储在向量数据库中，从而实现高效搜索。

但是，这种设置有局限性。最初的分块方法通常会导致检索不相关的信息，并且生成的答案缺乏精度以及与每个查询的意图的一致性。

大规模 RAG 查询应答的挑战

处理复杂查询：某些复杂问题需要超出基本语义搜索的更深入的语义理解。
上下文不匹配：检索到的块通常在上下文上相似，但不够精确，无法满足查询的要求。
检索精度限制：检索一小组文档（例如五到十个）通常会导致缺乏相关性的有限结果。

这些挑战强调需要更先进的方法来提高 RAG 查询应答的准确性。

增强查询准确性的高级 RAG 技术（解决方案）
为了解决这些问题，我应用了几种先进的方法，迭代地完善系统：
语义分块
与传统分块不同，语义分块优先考虑每个片段中的含义，通过将检索到的信息与查询的意图对齐来增强相关性。

Mastering Query Answering with RAG: Overcoming Key Challenges in Large-Scale Meeting Data

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai.embeddings import OpenAIEmbeddings
from langchain.schema import Document

# Initialize OpenAI Embeddings with API key
openai_api_key = ""
embedder = OpenAIEmbeddings(openai_api_key=openai_api_key)
text_splitter = SemanticChunker(embedder)

def prepare_docs_for_indexing(videos):
    all_docs = []

    for video in videos:
        video_id = video.get('video_id')
        title = video.get('video_name')
        transcript_info = video.get('details', {}).get('transcript_info', {})
        summary = video.get('details', {}).get('summary')
        created_at = transcript_info.get('created_at')  # Getting the created_at timestamp

        # Get the full transcription text
        transcription_text = transcript_info.get('transcription_text', '')

        # Create documents using semantic chunking
        docs = text_splitter.create_documents([transcription_text])

        for doc in docs:
            # Add metadata to each document
            doc.metadata = {
                "created_at": created_at,
                "title": title,
                "video_id": video_id,
                "summary": summary
            }
            all_docs.append(doc)

    return all_docs


docs = prepare_docs_for_indexing(videos)

# Output the created documents
for doc in docs:
    print("____________")
    print(doc.page_content)

登录后复制

最大保证金检索
该方法通过区分相关数据和不相关数据来提高检索精度，确保只检索最匹配的数据块。

Lambda 评分
使用 Lambda 评分，我可以根据相关性对结果进行排名，优先考虑与查询意图更一致的响应，以获得更好的答案质量。

from langchain_community.vectorstores import OpenSearchVectorSearch
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()

docsearch = OpenSearchVectorSearch.from_documents(
    docs, embeddings, opensearch_url="http://localhost:9200"
)

query = "your query"
docs = docsearch.max_marginal_relevance_search(query, k=2, fetch_k=10, lambda_param=0.25)

登录后复制

多查询和 RAG 融合
对于复杂的问题，系统会生成多个子查询。然后，RAG Fusion 将不同的答案整合为一个统一的、有凝聚力的响应，从而提高响应质量并减少错误。

def generate_multi_queries(question: str):
    # Template to generate multiple queries
    template = """You are an AI language model assistant. Your task is to generate five 
    different versions of the given user question to retrieve relevant documents from a vector 
    database. By generating multiple perspectives on the user question, your goal is to help
    the user overcome some of the limitations of the distance-based similarity search. 
    Provide these alternative questions separated by newlines. Original question: {question}"""

    # Creating a prompt template for query generation
    prompt_perspectives = ChatPromptTemplate.from_template(template)

    # Generate the queries using ChatOpenAI and output parser
    generate_queries = (
        prompt_perspectives 
        | ChatOpenAI(temperature=0, openai_api_key=openai_api_key) 
        | StrOutputParser() 
        | (lambda x: x.split("\n"))
    )

    # Invoke the chain to generate queries
    multi_queries = generate_queries.invoke({"question": question})

    return multi_queries

登录后复制

def reciprocal_rank_fusion(results: list[list], k=60):
    """Applies Reciprocal Rank Fusion (RRF) to fuse ranked document lists."""
    fused_scores = {}
    for docs in results:
        for rank, doc in enumerate(docs):
            doc_str = dumps(doc)  # Convert to a serializable format
            if doc_str not in fused_scores:
                fused_scores[doc_str] = 0
            fused_scores[doc_str] += 1 / (rank + k)  # RRF formula

    # Sort documents by the fused score
    reranked_results = [
        (loads(doc), score)
        for doc, score in sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)
    ]
    return reranked_result

登录后复制

Mastering Query Answering with RAG: Overcoming Key Challenges in Large-Scale Meeting Data