我要投稿

提高企业 RAG 准确性的分步指南

发布日期：2025-03-05 11:47:25 浏览次数： 1801 作者：barry的异想世界

从PDF文件生成的知识图谱

在我之前的博客中，我写到如何使用像Gemini Flash 2.0这样具有非常大上下文大小的新模型进行语义分块，可以显著提高从非结构化数据（如PDF）中的整体检索准确性。

在探索这一点时，我开始研究其他策略，以进一步提高响应的准确性，因为在大多数大型企业中，对不准确的容忍度几乎为零，并且应该如此。在这个追求中，我最终尝试了许多不同的东西，在这篇博客中，让我们看看最终帮助提升准确性的整体步骤。

在进入步骤之前，让我们从稍高的层面看一下整个过程，并理解为了获得更准确的结果，我们必须在以下两个方面显著做得更好：

1. 提取 — 给定一组文档，以一种本质上有序的方式提取数据和知识，以便于更好和更准确的检索。
2. 检索 — 当查询到达时，查看一些检索前和检索后的步骤，并通过“知识”进行更好的上下文化，以获得更好的结果。

现在，让我们看看反映当前项目状态的具体步骤。我在每个步骤中放入了一些伪代码，以使整体文章更易于理解。对于那些寻找代码级具体内容的人，可以查看我的Github仓库，我在其中发布了代码。

1. 从PDF中提取知识

流程

当 PDF 进入系统时，需要发生几件事情：它被存储、处理、分块、嵌入，并与结构化知识进行丰富。以下是整个过程的展开：

步骤 1：上传与记录创建

• 用户上传一个 PDF（其他文件类型如音频和视频文件即将推出）。
• 系统将文件保存到磁盘（不久的将来，我将把它移动到 AWS S3 存储桶，以更好地支持企业用例）。
• 一条记录被插入到数据库中，并创建一个处理状态条目。对于数据库，我使用 SingleStore，因为它支持多种数据类型和混合搜索以及单次检索）。
• 一个后台任务被排队以异步处理 PDF。这让我深入研究了整体步骤所需的时间，最终决定使用 Redis 和 Celery 进行任务处理和跟踪。这在部署时确实有些麻烦，但我们可以稍后再讨论。

# 伪代码 save_file_to_disk(pdf) db_insert(document_record, status=”started”) queue_processing_task(pdf)

第2步：解析与分块 PDF

• 文件被打开并验证大小限制或密码保护，原因是我们希望在文件不可读时尽早失败该过程。
• 内容被提取为文本/markdown。这是另一个大话题。我之前使用 PyMudf 进行整体提取，但后来我发现了 Llamaparse，切换后让我生活变得轻松许多。Llamaparse 的免费版本允许每天解析 1000 个文档，并具有许多附加功能，可以以不同格式返回响应，并更好地提取 PDF 中的表格和图像。
• 文档结构被分析（例如，目录、标题等）。
• 使用语义方法将文本拆分为有意义的块。这是我使用 Gemini Flash 2.0 的地方，因为它具有巨大的上下文大小和显著更低的定价。
• 如果语义分块失败，系统会回退到更简单的分段。
• 在块之间添加重叠以保持上下文。

validate_pdf(pdf)
text = extract_text(pdf)
chunks = semantic_chunking(text) or fallback_chunking(text)
add_overlaps(chunks)

第3步：生成嵌入

• 每个块都通过嵌入模型转换为高维向量。我使用1536维，因为我使用了OpenAI的大型ada模型。
• 接下来，块及其嵌入都存储在数据库中。在SingleStore中，我们将块和文本存储在同一个表中的两个不同列中，以便于维护和检索。

# 伪代码 for chunk in chunks: vector = generate_embedding(chunk.text) db_insert(embedding_record, vector)

第4步：使用LLMs提取实体和关系

• 这是对整体准确性影响很大的事情。在此步骤中，我将语义组织的块发送给OpenAI，并通过一些特定的提示请求它从每个块中返回实体和关系。结果包括关键实体（名称、类型、描述、别名）。
• 实体之间的关系被映射出来。在这里，如果我们找到多个实体，我们每次都用丰富的数据更新类别，而不是添加重复项。
• 提取的“知识”现在存储在结构化表中。

# 伪代码 for chunk in chunks: entities, relationships = extract_knowledge(chunk.text) db_insert(entities) db_insert(relationships)

第5步：最终处理状态

• 如果一切处理正确，状态将更新为“已完成”。这样前端可以持续轮询，并在任何时候显示正确的状态。
• 如果出现故障，状态将标记为“失败”，并清理任何临时数据。

# 伪代码 if success: update_status(“completed”) else: update_status(“failed”) cleanup_partial_data()

当这些步骤完成后，我们现在拥有语义块、相应的嵌入以及在文档中找到的实体和关系，这些都在相互引用的表格中。

我们现在准备好进行下一步，即检索。

2. 检索知识 (RAG 管道)

流程

现在数据已经结构化并存储，我们需要在用户提问时有效地检索这些数据。系统处理查询，找到相关信息，并生成响应。

步骤 1：用户查询

• 用户向系统提交查询。

第2步：预处理与扩展查询

• 系统对查询进行规范化（去除标点，规范空格，使用同义词进行扩展）。在这里我再次使用LLM（Groq以加快处理速度）

query = preprocess_query(query) expanded_query = expand_query(query)

第3步：嵌入查询和搜索向量

• 查询被嵌入到一个高维向量中。我使用之前用于提取的相同ada模型。
• 系统使用语义搜索在文档嵌入数据库中搜索最接近的匹配项。我在SingleStore中使用dot_product来实现这一点。

query_vector = generate_embedding(expanded_query) top_chunks = vector_search(query_vector)

第4步：全文搜索

• 进行并行全文搜索以补充向量搜索。在SingleStore中，我们使用MATCH语句来实现这一点。

text_results = full_text_search(query)

第5步：合并与排名结果

• 向量和文本搜索结果被合并并根据相关性重新排名。我们可以在这里调整的配置之一是前 k 个结果。我在 top k = 10 或更高时得到了更好的结果。
• 低置信度的结果被过滤掉。

merged_results = merge_and_rank(top_chunks, text_results) filtered_results = filter_low_confidence(merged_results)

第6步：检索实体和关系

• 接下来，如果检索到的块存在实体和关系，则将其包含在响应中。

# 伪代码 for result in filtered_results: entities, relationships = fetch_knowledge(result) enrich_result(result, entities, relationships)

第7步：生成最终答案

• 现在我们根据整体上下文，通过提示增强上下文，并将相关数据发送给LLM（我使用的是gpt3o-mini）以生成最终响应。

final_answer = generate_llm_response(filtered_results)

第8步：将答案返回给用户

• 系统将响应作为结构化的 JSON 负载发送回去，并附带原始数据库搜索结果，以便在需要时识别源进行进一步调试和调整。

# 伪代码 return_response(final_answer)

现在，关键来了。总体而言，检索过程对我来说大约需要 8 秒，这是不可接受的。

在追踪调用时，我发现最大响应时间来自 LLM 调用（大约 1.5 到 2 秒）。SingleStore 数据库查询始终在 600 毫秒或更短的时间内返回。切换到 Groq 进行一些 LLM 调用后，总体响应时间降至 3.5 秒。我认为如果我们进行一些并行调用而不是串行调用，这可以进一步改善，但那是另一个项目。

最后，关键来了。

鉴于我们正在使用 SingleStore，我想看看是否可以只进行一次查询来完成整个检索，这样不仅更容易管理、更新和改进，而且因为我希望从数据库获得更好的响应时间。这里的假设是 LLM 模型在不久的将来会变得更好、更快，而我无法控制这些（当然，如果你真的对延迟非常认真，可以在同一网络中部署本地 LLM）。

最后，这里是代码（为了方便，单个文件）现在执行单次检索查询。

import os
import json
import mysql.connector
from openai import OpenAI

DB\_CONFIG = {
"host": os.getenv("SINGLESTORE\_HOST", "localhost"),
"port": int(os.getenv("SINGLESTORE\_PORT", "3306")),
"user": os.getenv("SINGLESTORE\_USER", "root"),
"password": os.getenv("SINGLESTORE\_PASSWORD", ""),
"database": os.getenv("SINGLESTORE\_DATABASE", "knowledge\_graph")
}
defget\_query\_embedding(query: str) -\> list:
"""
 Generate a 1536-dimensional embedding for the query using OpenAI embeddings API.
 """
 client = OpenAI(api\_key=os.getenv("OPENAI\_API\_KEY"))
 response = client.embeddings.create(
 model="text-embedding-ada-002",
input=query
 )
return response.data\[0\].embedding 
defretrieve\_rag\_results(query: str) -\> list:
"""
 Execute the hybrid search SQL query in SingleStore and return the top-ranked results.
 """
 conn = mysql.connector.connect(\*\*DB\_CONFIG)
 cursor = conn.cursor(dictionary=True)

 query\_embedding = get\_query\_embedding(query)
 embedding\_str = json.dumps(query\_embedding) 

 cursor.execute("SET @qvec = %s", (embedding\_str,))


 sql\_query = """
SELECT 
 d.doc\_id,
 d.content,
 (d.embedding <\*\> @qvec) AS vector\_score,
 MATCH(TABLE Document\_Embeddings) AGAINST(%s) AS text\_score,
 (0.7 \* (d.embedding <\*\> @qvec) + 0.3 \* MATCH(TABLE Document\_Embeddings) AGAINST(%s)) AS combined\_score,
 JSON\_AGG(DISTINCT JSON\_OBJECT(
 'entity\_id', e.entity\_id,
 'name', e.name,
 'description', e.description,
 'category', e.category
 )) AS entities,
 JSON\_AGG(DISTINCT JSON\_OBJECT(
 'relationship\_id', r.relationship\_id,
 'source\_entity\_id', r.source\_entity\_id,
 'target\_entity\_id', r.target\_entity\_id,
 'relation\_type', r.relation\_type
 )) AS relationships
FROM Document\_Embeddings d
LEFT JOIN Relationships r ON r.doc\_id = d.doc\_id
LEFT JOIN Entities e ON e.entity\_id IN (r.source\_entity\_id, r.target\_entity\_id)
WHERE MATCH(TABLE Document\_Embeddings) AGAINST(%s)
GROUP BY d.doc\_id, d.content, d.embedding
ORDER BY combined\_score DESC
LIMIT 10;
 """

 cursor.execute(sql\_query, (query, query, query))
 results = cursor.fetchall()
 cursor.close()
 conn.close()
return results