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

从PDF文件生成的知识图谱

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

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

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

1. 1. 提取 — 给定一组文档，以一种本质上有序的方式提取数据和知识，以便于更好和更准确的检索。
2. 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

经验教训

正如你所想，进行“简单的” RAG 与你的 pdf 聊天是一回事，而在保持低延迟的同时实现超过 80% 的准确率又是另一回事。现在再加入结构化数据，你已经深入到一个项目中，这几乎成了一份全职工作 ?

我计划继续调整和改进，并为这个项目写博客，短期内，我正在寻找一些接下来要探索的想法。

精度增强

提取：

1. 1. 外部化并实验实体提取提示。
2. 2. 在处理之前对块进行总结。我感觉这可能会产生非平凡的影响。
3. 3. 为不同步骤中的失败添加更好的重试机制。

检索

1. 1. 使用更好的查询扩展技术（自定义词典、行业特定术语）。
2. 2. 微调向量与文本搜索的权重（这已经在配置 YAML 文件中外部化）。
3. 3. 添加第二次 LLM 处理以重新排序顶级结果（考虑到延迟权衡，我对此持谨慎态度）。
4. 4. 调整检索窗口大小，以优化召回率与相关性。
5. 5. 生成块级摘要，而不是将原始文本发送给 LLM。

总结

在许多方面，我记录这些内容是为了提醒自己在构建企业级 RAG 或 KAG 时需要考虑的企业需求。如果作为读者的你发现我做的一些事情非常幼稚，或者有其他想法可以改进我，请随时在这里或在 LinkedIn 上与我联系，以便我们可以一起合作