提高企业 RAG 准确性的分步指南
从PDF文件生成的知识图谱
在我之前的博客中，我写到如何使用像Gemini Flash 2.0这样具有非常大上下文大小的新模型进行语义分块，可以显著提高从非结构化数据（如PDF）中的整体检索准确性。
在探索这一点时，我开始研究其他策略，以进一步提高响应的准确性，因为在大多数大型企业中，对不准确的容忍度几乎为零，并且应该如此。在这个追求中，我最终尝试了许多不同的东西，在这篇博客中，让我们看看最终帮助提升准确性的整体步骤。
在进入步骤之前，让我们从稍高的层面看一下整个过程，并理解为了获得更准确的结果，我们必须在以下两个方面显著做得更好：
- 提取 — 给定一组文档，以一种本质上有序的方式提取数据和知识，以便于更好和更准确的检索。
- 检索 — 当查询到达时，查看一些检索前和检索后的步骤，并通过“知识”进行更好的上下文化，以获得更好的结果。
现在，让我们看看反映当前项目状态的具体步骤。我在每个步骤中放入了一些伪代码，以使整体文章更易于理解。对于那些寻找代码级具体内容的人，可以查看我的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")
}
def get\_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
def retrieve\_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% 的准确率又是另一回事。现在再加入结构化数据，你已经深入到一个项目中，这几乎成了一份全职工作 😅
我计划继续调整和改进，并为这个项目写博客，短期内，我正在寻找一些接下来要探索的想法。
精度增强
提取：
- 外部化并实验实体提取提示。
- 在处理之前对块进行总结。我感觉这可能会产生非平凡的影响。
- 为不同步骤中的失败添加更好的重试机制。
检索
- 使用更好的查询扩展技术（自定义词典、行业特定术语）。
- 微调向量与文本搜索的权重（这已经在配置 YAML 文件中外部化）。
- 添加第二次 LLM 处理以重新排序顶级结果（考虑到延迟权衡，我对此持谨慎态度）。
- 调整检索窗口大小，以优化召回率与相关性。
- 生成块级摘要，而不是将原始文本发送给 LLM。
总结
在许多方面，我记录这些内容是为了提醒自己在构建企业级 RAG 或 KAG 时需要考虑的企业需求。如果作为读者的你发现我做的一些事情非常幼稚，或者有其他想法可以改进我，请随时在这里或在 LinkedIn 上与我联系，以便我们可以一起合作。
✌️
