DeepSeek-V3.2 的发布为复杂推理任务提供了极佳的开源选择，其长文本处理能力也让深度文档解析变得更高效。而在 RAG（检索增强生成）实践中，如何精准提取 PDF 结构依然是关键。IBM 的开源工具 Docling 改变了传统 OCR 简单提取字符串的方式，通过视觉语义分析将文档转化为结构化的 Markdown。
本文将分享如何将两者结合，搭建一套极简的 Agentic RAG 工作流。不同于传统的单向检索，该系统通过“检索-研究-验证”的智能体闭环，实现了具备自纠错能力的文档问答，能以极低的代码成本有效减少模型幻觉。
1、DeepSeek-V3.2 的独特之处是什么？
大多数强大的 AI 模型都面临一个共同问题：随着文件长度的增加，模型的执行速度会显著下降，而成本则会急剧上升。这是因为传统模型会尝试将每个词与其他所有词进行比较，以理解上下文。
DeepSeek-V3.2 通过引入一种名为 DeepSeek 稀疏注意力（DeepSeek Sparse Attention，DSA）的新方法来解决这个问题。可以将其想象成一位研究人员在图书馆进行研究：
- • 传统方法（密集注意力）： 研究人员为了回答一个问题，需要逐页阅读书架上的每一本书。虽然这种方法很全面，但速度极慢，而且需要耗费大量精力。
- • 新方法（DeepSeek-V3.2）： 研究人员使用数字索引（Lightning Indexer）快速找到关键页面，然后只阅读这些页面。这种方法同样准确，但速度却快得多。
2、Docling 的独特之处是什么？
Docling 之所以能从现有工具中脱颖而出，最大的原因在于其设计理念基于与生成式人工智能（特别是检索增强生成，RAG）的协作。
现代人工智能应用需要的不仅仅是提取文本。为了让人工智能深入理解文档内容并生成准确的答案，需要了解文档的语义，包括：
- • 这句话是论文的"摘要"还是"结论"？
- • 这串数字不仅仅是文本，而是一个"表格"，那么每个单元格代表什么？
- • 这张图片的"标题"是什么？
PyMuPDF 和 Tesseract 将文本提取为"字符串"，而 Docling 则利用视觉-语言模型（VLM）的强大功能来分析这些结构和关系，并将其输出为包含丰富信息的"DoclingDocument"对象。
这些结构化数据是显著提升 RAG 检索和答案生成质量的关键。
3、配置环境
首先安装支持该模型的库，执行 pip install requirements 命令。
pip install requirements
下一步是常规操作：导入相关的库，其重要性将在后续步骤中逐渐显现。
- •
DocumentConverter：一个高级 Python 类，用于将文档转换为结构化的DoclingDocument格式。 - •
EnsembleRetriever：一个集成检索器，它使用加权倒数排序融合（weighted reciprocal rank fusion）来聚合和排序多个检索器的结果。
4、DocLing
以下代码创建了一个 VerificationAgent 类，用于根据源文档对 AI 生成的答案进行事实核查。在 \_\_init\_\_ 函数中，实例化了一个温度为零（用于确定性输出）的 DeepSeek-V3.2 模型，并构建了一个提示模板。这个模板要求 LLM 从四个特定方面验证答案：声明是否得到直接支持、哪些内容不被支持、哪些内容相互矛盾以及它是否相关，并强制使用结构化响应格式以确保解析的一致性。
在 check() 函数中，接收答案字符串和一个 Document 对象列表，提取并连接所有文档文本到一个上下文字符串中，然后创建一个 LangChain 管道（提示 → LLM → 字符串解析器），并使用答案和上下文调用该管道以获取验证报告。
记录报告和上下文以进行调试，如果有任何错误发生，则重新抛出。最后，返回一个包含验证报告文本和使用的上下文字符串的字典——这样做的目的是通过检查 RAG 系统生成的答案是否确实得到源文档的支持来捕获"幻觉"（hallucinations）。
import os
import hashlib
import pickle
from datetime import datetime, timedelta
from pathlib import Path
from typing importList
from docling.document\_converter import DocumentConverter
from langchain\_text\_splitters import MarkdownHeaderTextSplitter
from config import constants
from config.settings import settings
from utils.logging import logger
classDocumentProcessor:
def\_\_init\_\_(self):
self.headers = [("#", "Header 1"), ("##", "Header 2")]
self.cache\_dir = Path(settings.CACHE\_DIR)
self.cache\_dir.mkdir(parents=True, exist\_ok=True)
defvalidate\_files(self, files: List) -> None:
"""验证上传文件的总大小。"""
total\_size = sum(os.path.getsize(f.name) for f in files)
if total\_size > constants.MAX\_TOTAL\_SIZE:
raise ValueError(f"总大小超过 {constants.MAX\_TOTAL\_SIZE//1024//1024}MB 限制")
defprocess(self, files: List) -> List:
"""处理文件，并进行缓存，以便后续查询使用。"""
self.validate\_files(files)
all\_chunks = []
seen\_hashes = set()
for file in files:
try:
# 为缓存生成基于内容的哈希值
withopen(file.name, "rb") as f:
file\_hash = self.\_generate\_hash(f.read())
cache\_path = self.cache\_dir / f"{file\_hash}.pkl"
ifself.\_is\_cache\_valid(cache\_path):
logger.info(f"从缓存加载：{file.name}")
chunks = self.\_load\_from\_cache(cache\_path)
else:
logger.info(f"处理并缓存：{file.name}")
chunks = self.\_process\_file(file)
self.\_save\_to\_cache(chunks, cache\_path)
# 跨文件进行块去重
for chunk in chunks:
chunk\_hash = self.\_generate\_hash(chunk.page\_content.encode())
if chunk\_hash notin seen\_hashes:
all\_chunks.append(chunk)
seen\_hashes.add(chunk\_hash)
except Exception as e:
logger.error(f"处理文件 {file.name} 失败：{str(e)}")
continue
logger.info(f"总计唯一块数：{len(all\_chunks)}")
return all\_chunks
def\_process\_file(self, file) -> List:
"""使用 Docling 进行原始处理逻辑。"""
ifnot file.name.endswith(('.pdf', '.docx', '.txt', '.md')):
logger.warning(f"跳过不支持的文件类型：{file.name}")
return []
converter = DocumentConverter()
markdown = converter.convert(file.name).document.export\_to\_markdown()
splitter = MarkdownHeaderTextSplitter(self.headers)
return splitter.split\_text(markdown)
def\_generate\_hash(self, content: bytes) -> str:
"""生成内容的 SHA256 哈希值。"""
return hashlib.sha256(content).hexdigest()
def\_save\_to\_cache(self, chunks: List, cache\_path: Path):
"""将处理过的块保存到缓存文件。"""
withopen(cache\_path, "wb") as f:
pickle.dump({
"timestamp": datetime.now().timestamp(),
"chunks": chunks
}, f)
def\_load\_from\_cache(self, cache\_path: Path) -> List:
"""从缓存文件加载块。"""
withopen(cache\_path, "rb") as f:
data = pickle.load(f)
return data["chunks"]
def\_is\_cache\_valid(self, cache\_path: Path) -> bool:
"""检查缓存文件是否有效（未过期）。"""
ifnot cache\_path.exists():
returnFalse
cache\_age = datetime.now() - datetime.fromtimestamp(cache\_path.stat().st\_mtime)
return cache\_age < timedelta(days=settings.CACHE\_EXPIRE\_DAYS)
5、RelevanceChecker
以下代码创建了一个 RelevanceChecker 类，通过将检索到的文档分类为三个类别来确定它们是否可以回答用户的问题。
在 \_\_init\_\_ 方法中，使用 API 密钥初始化了一个 DeepSeek-V3.2 模型，并创建了一个提示模板。该模板指示大型语言模型（LLM）将段落分类为"CAN_ANSWER"（完全回答）、“PARTIAL”（提及主题但不完整）或"NO_MATCH"（完全未讨论该主题），并强调任何主题提及都应归类为"PARTIAL"，而不是"NO_MATCH"。通过 提示 -> LLM -> 字符串解析器 的管道连接，构建了一个 LangChain 链。
在 check() 方法中，接收一个问题、一个检索器对象和一个参数 k（默认为 3），用于指定要分析的前 k 个文档的数量。使用问题调用检索器以获取相关的文本块，如果没有返回任何内容，则立即返回"NO_MATCH"。
为了便于查看，打印调试信息，显示文档计数以及前 k 个文本块的 200 字符预览。将前 k 个文档的文本合并成一个字符串，字符串之间用双换行符分隔，然后使用问题和合并后的内容调用 LLM 链，并获取一个分类字符串。
通过将响应转换为大写并检查是否符合有效选项来验证响应是否为三个有效标签之一，如果 LLM 返回了意外结果，则强制返回"NO_MATCH"。
最后，返回经过验证的分类结果，从而清楚地了解检索器是否找到了可用的文档，或者是否需要回退到其他方法，例如网络搜索。
# agents/relevance\_checker.py
from langchain\_openai import ChatOpenAI
from langchain\_core.prompts import ChatPromptTemplate
from langchain\_core.output\_parsers import StrOutputParser
from langchain\_deepseek import ChatDeepSeek
from config.settings import settings
classRelevanceChecker:
def\_\_init\_\_(self):
# self.llm = ChatOpenAI(api\_key=settings.OPENAI\_API\_KEY, model="gpt-4o")
self.llm = ChatDeepSeek(api\_key=settings.DEEPSEEK\_API\_KEY, model="deepseek-chat")
self.prompt = ChatPromptTemplate.from\_template(
"""给定一个用户问题和一些来自上传文档的段落。
请评估这些段落回答用户问题的程度。
请**准确选择**以下回答中的一个（**只**返回该标签）：
1) "CAN\_ANSWER":这些段落包含足够明确的信息来完整回答问题。
2) "PARTIAL": 这些段落提及或讨论了问题的主题（例如，相关年份、设施名称），
但没有提供回答问题所需的所有数据或细节。
3) "NO\_MATCH": 这些段落完全没有讨论或提及问题的主题。
重要提示：如果段落以**任何方式**提及或引用了问题的主题或时间范围，
即使信息不完整，也应返回 "PARTIAL"，而不是 "NO\_MATCH"。
问题: {question}
段落: {document\_content}
请**只**返回 "CAN\_ANSWER", "PARTIAL", 或 "NO\_MATCH"。
"""
)
self.chain = self.prompt | self.llm | StrOutputParser()
defcheck(self, question: str, retriever, k=3) -> str:
"""
1. 从全局检索器中检索 top-k 文档块。
2. 将它们组合成一个单独的文本字符串。
3. 将该文本 + 问题传递给 LLM 链进行分类。
返回: "CAN\_ANSWER" 或"PARTIAL" 或 "NO\_MATCH"。
"""
print(f"[DEBUG] RelevanceChecker.check 被调用，问题='{question}'，k={k}")
# 从检索器中检索文档块
top\_docs = retriever.invoke(question)[:k] # 只使用前 k 个文档
ifnot top\_docs:
print("[DEBUG] 检索器.invoke() 未返回任何文档。分类为 NO\_MATCH。")
return"NO\_MATCH"
print(f"[DEBUG] 检索器返回了 {len(top\_docs)} 个文档。")
# 快速显示每个块的片段以进行调试
for i, doc inenumerate(top\_docs):
snippet = doc.page\_content[:200].replace("\n", "\\n")
print(f"[DEBUG] 块 #{i+1}预览（前 200 个字符）：{snippet}...")
# 将前 k 个块的文本合并成一个字符串
document\_content = "\n\n".join(doc.page\_content for doc in top\_docs)
print(f"[DEBUG] 前 {k} 个块的组合文本长度：{len(document\_content)} 字符。")
# 调用 LLM
response = self.chain.invoke({
"question": question,
"document\_content": document\_content
}).strip()
print(f"[DEBUG] LLM原始分类响应：'{response}'")
# 转换为大写，检查是否是有效标签之一
classification = response.upper()
valid\_labels = {"CAN\_ANSWER", "PARTIAL", "NO\_MATCH"}
if classification notin valid\_labels:
print("[DEBUG] LLM 未返回有效标签。强制设为 'NO\_MATCH'。")
classification = "NO\_MATCH"
else:
print(f"[DEBUG] 分类被识别为 '{classification}'。")
return classification
6、ResearchAgent
以下代码创建了一个名为 ResearchAgent 的类，使用检索到的文档作为上下文来生成问题的答案。
代码创建了一个提示模板，要求 LLM 根据提供的上下文回答问题，做到准确且基于事实，并指示 LLM 在上下文不足时明确表示"我无法根据提供的文档回答此问题"。
在 generate() 方法中，接收一个问题字符串和一个 Document 对象列表，然后提取所有文档文本，并使用双换行符作为分隔符将其连接成一个上下文字符串。
使用问题和上下文调用该链，将其替换到模板中，向 DeepSeek 发送请求，并将生成的答案作为字符串返回。此过程被包装在 try-except 语句中，以便记录答案和完整的上下文进行调试，并重新引发任何发生的异常。
最后，返回一个包含草稿答案和所用上下文的字典，从而同时获得生成的响应和用于创建它的源材料的可追溯性。
from langchain\_openai import ChatOpenAI
from langchain\_core.prompts import ChatPromptTemplate
from langchain\_core.output\_parsers import StrOutputParser
from typing importDict, List
from langchain\_core.documents import Document
from langchain\_deepseek import ChatDeepSeek
import logging
from config.settings import settings
logger = logging.getLogger(\_\_name\_\_)
classResearchAgent:
def\_\_init\_\_(self):
"""用 OpenAI 模型初始化研究智能体。"""
# self.llm = ChatOpenAI(
# model="gpt-4-turbo",
# temperature=0.3,
# api\_key=settings.OPENAI\_API\_KEY # 在这里传递 API 密钥
# )
self.llm = ChatDeepSeek(
model="deepseek-chat",
temperature=0.3,
api\_key=settings.DEEPSEEK\_API\_KEY # 在这里传递 API 密钥
)
self.prompt = ChatPromptTemplate.from\_template(
"""根据提供的上下文回答以下问题。请精确且基于事实。
问题: {question}
上下文:
{context}
如果上下文不足，请回答: "我无法根据提供的文档回答此问题。"
"""
)
defgenerate(self, question: str, documents: List[Document]) -> Dict:
"""使用提供的文档生成初步答案。"""
context = "\n\n".join([doc.page\_content for doc in documents])
chain = self.prompt | self.llm | StrOutputParser()
try:
answer = chain.invoke({
"question": question,
"context": context
})
logger.info(f"生成的答案: {answer}")
logger.info(f"使用的上下文: {context}")
except Exception as e:
logger.error(f"生成答案时出错: {e}")
raise
return {
"draft\_answer": answer,
"context\_used": context
}
7、验证智能体
以下代码创建了一个 VerificationAgent 类，用于对 AI 生成的答案进行事实核查，以识别虚假信息。在 \_\_init\_\_ 方法中，初始化了一个温度为 0（表示完全确定性）的 DeepSeek-V3.2 模型，创建了一个提示模板，指示 LLM 使用结构化的响应格式验证四个方面（直接事实支持、未经证实的说法、矛盾和相关性），然后构建了一个 LangChain 链。
在 check() 函数中，接收一个响应字符串和一个 Document 对象列表，将所有文档文本用双换行符连接成一个上下文字符串，然后使用响应字符串和上下文字符串调用链式调用以获取验证报告。在 try-except 块中，记录报告和上下文字符串以进行调试，最后返回一个包含验证报告和上下文字符串的字典，用于溯源。
from langchain\_openai import ChatOpenAI
from langchain\_core.prompts import ChatPromptTemplate
from langchain\_core.output\_parsers import StrOutputParser
from typing importDict, List
from langchain\_core.documents import Document
from langchain\_deepseek import ChatDeepSeek
import logging
from config.settings import settings
logger = logging.getLogger(\_\_name\_\_)
classVerificationAgent:
def\_\_init\_\_(self):
# self.llm = ChatOpenAI(
# model="gpt-4-turbo",
# temperature=0,
# api\_key=settings.OPENAI\_API\_KEY # 在这里传递 API 密钥
# )
self.llm = ChatDeepSeek(
model="deepseek-chat",
temperature=0,
api\_key=settings.DEEPSEEK\_API\_KEY # 在这里传递 API 密钥
)
self.prompt = ChatPromptTemplate.from\_template(
"""根据提供的上下文验证以下答案。检查以下各项：
1. 直接事实支持 (是/否)
2. 未经支持的主张 (列出)
3. 矛盾之处 (列出)
4. 与问题的相关性 (是/否)
请按以下格式回应：
支持: 是/否
未经支持的主张: [项目]
矛盾之处: [项目]
相关性: 是/否
答案: {answer}
上下文: {context}
"""
)
defcheck(self, answer: str, documents: List[Document]) -> Dict:
"""根据提供的文档验证答案。"""
context = "\n\n".join([doc.page\_content for doc in documents])
chain = self.prompt | self.llm | StrOutputParser()
try:
verification = chain.invoke({
"answer": answer,
"context": context
})
logger.info(f"验证报告: {verification}")
logger.info(f"使用的上下文: {context}")
except Exception as e:
logger.error(f"验证答案时出错: {e}")
raise
return {
"verification\_report": verification,
"context\_used": context
}
添加微信，备注” LLM “进入大模型技术交流群
如果你觉得这篇文章对你有帮助，别忘了点个赞、送个喜欢
/ 作者：ChallengeHub小编
/ 作者：欢迎转载，标注来源即可