微信扫码
添加专属顾问
我要投稿
提升企业数据检索效率的必备指南,RAG系统准确性提升全流程解析。 核心内容: 1. PDF文件上传与记录创建,优化存储和处理流程 2. PDF解析与分块,提升文本提取和语义分割准确性 3. 嵌入模型应用与实体关系提取,增强检索精准度
在企业环境中,准确高效地从大量非结构化数据(如 PDF 文件)中检索信息至关重要。基于检索增强生成(RAG)的系统在这方面发挥着重要作用,但提升其准确性是一个复杂且具有挑战性的任务。本文将详细介绍提高企业 RAG 准确性的逐步指南,涵盖从数据提取到检索的各个关键步骤。
用户上传 PDF 文件(未来还将支持音频和视频等其他文件类型)。系统将文件保存到磁盘(近期计划迁移到 AWS S3 存储桶以更好地满足企业需求),并在数据库中插入记录,创建处理状态条目。使用支持多种数据类型、混合搜索和单次检索的 SingleStore 数据库。同时,将处理 PDF 的任务放入后台队列,通过 Redis 和 Celery 进行异步处理和跟踪。
# Pseudo-codesave_file_to_disk(pdf)db_insert(document_record, status=”started”)queue_processing_task(pdf)
打开文件并验证大小限制和密码保护,若文件不可读则尽早终止处理。将文件内容提取为文本或 Markdown 格式,可使用 Llamaparse(来自 Llamaindex)替代之前的 PyMudf,其免费版每天支持 1000 次文档解析,并能更好地处理表格和图像提取。分析文档结构(如目录、标题等),利用 Gemini Flash 2.0 基于语义将文本分割成有意义的块,若语义分块失败,则采用简单的分割方式,并在块之间添加重叠部分以保持上下文连贯性。
# Pseudo-codevalidate_pdf(pdf)text = extract_text(pdf)chunks = semantic_chunking(text) or fallback_chunking(text)add_overlaps(chunks)
# Pseudo-codefor chunk in chunks: vector = generate_embedding(chunk.text) db_insert(embedding_record, vector)
这一步对整体准确性影响很大。将语义组织好的文本块发送给 OpenAI,通过特定的提示要求其返回每个块中的实体和关系,包括关键实体(名称、类型、描述、别名)。映射实体之间的关系,避免重复添加数据,将提取的 “知识” 存储在结构化表中。
# Pseudo-codefor chunk in chunks: entities, relationships = extract_knowledge(chunk.text) db_insert(entities) db_insert(relationships)
若所有步骤处理正确,将状态更新为 “已完成”,以便前端随时轮询并显示正确状态;若处理失败,则标记为 “失败”,并清理临时数据。
# Pseudo-codeif success: update_status(“completed”)else: update_status(“failed”) cleanup_partial_data()
用户向系统提交查询请求。
# Pseudo-codequery = get_user_query()
系统对查询进行规范化处理,去除标点符号、标准化空白字符,并使用大语言模型(如 Groq,处理速度更快)扩展同义词。
# Pseudo-codequery = preprocess_query(query)expanded_query = expand_query(query)
使用与提取时相同的 ada 模型将查询嵌入为高维向量,在文档嵌入数据库中使用语义搜索(如在 SingleStore 中使用点积运算)查找最匹配的文本块。
# Pseudo-codequery_vector = generate_embedding(expanded_query)top_chunks = vector_search(query_vector)
并行进行全文搜索以补充向量搜索,在 SingleStore 中可使用 MATCH 语句实现。
# Pseudo-codetext_results = full_text_search(query)
将向量搜索和全文搜索的结果合并,并根据相关性重新排序,可调整返回的前 k 个结果数量(如 k = 10 或更高效果更好),过滤掉低置信度的结果。
# Pseudo-codemerged_results = merge_and_rank(top_chunks, text_results)filtered_results = filter_low_confidence(merged_results)
若检索到的文本块存在实体和关系,则将其包含在响应中。
# Pseudo-codefor result in filtered_results: entities, relationships = fetch_knowledge(result) enrich_result(result, entities, relationships)
利用提示增强上下文,将相关数据发送给大语言模型(如 gpt3o - mini)生成最终响应。
# Pseudo-codefinal_answer = generate_llm_response(filtered_results)
系统将响应作为结构化 JSON 有效负载返回,并附带原始数据库搜索结果,以便在需要时进行调试和调整。
# Pseudo-codereturn_response(final_answer)
在实践中,发现检索过程的响应时间过长(约 8 秒),主要瓶颈在于大语言模型的调用(每次约 1.5 - 2 秒),而 SingleStore 数据库查询时间通常在 600 毫秒以内。切换到 Groq 进行部分大语言模型调用后,响应时间缩短至 3.5 秒。为进一步优化,可尝试并行调用而非串行调用。
为简化管理和提高数据库响应时间,实现了单次检索查询的代码。通过 OpenAI 的 Embeddings API 生成查询嵌入向量,在 SingleStore 中执行混合搜索 SQL 查询,同时获取文本块、向量得分、文本得分、综合得分以及相关的实体和关系信息。
import os
import json
import mysql.connector
from openai import OpenAI
# Define database connection parameters (assumed from env vars)
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 # Extract embedding vector
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)
# Generate query embedding
query_embedding = get_query_embedding(query)
embedding_str = json.dumps(query_embedding) # Convert to JSON for SQL compatibility
# Set the query embedding session variable
cursor.execute("SET @qvec = %s", (embedding_str,))
# Hybrid Search SQL Query (same as provided earlier)
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;
"""
# Execute the query
cursor.execute(sql_query, (query, query, query))
results = cursor.fetchall()
cursor.close()
conn.close()
return results # Return list of retrieved documents with entities and relationships
提高 RAG 准确性并保持低延迟是一项艰巨的任务,尤其是在处理结构化数据时。未来可从以下方面进行改进:
53AI,企业落地大模型首选服务商
产品:场景落地咨询+大模型应用平台+行业解决方案
承诺:免费场景POC验证,效果验证后签署服务协议。零风险落地应用大模型,已交付160+中大型企业
2024-10-27
2024-09-04
2024-07-18
2024-05-05
2024-06-20
2024-06-13
2024-07-09
2024-07-09
2024-05-19
2024-07-07
2025-04-01
2025-04-01
2025-03-30
2025-03-28
2025-03-27
2025-03-27
2025-03-25
2025-03-19