台中市网站建设_网站建设公司_全栈开发者_seo优化

本文介绍如何结合DeepSeek-V3.2和Docling搭建极简Agentic RAG工作流。DeepSeek-V3.2通过稀疏注意力机制高效处理长文本，Docling将PDF转化为结构化Markdown。系统实现"检索-研究-验证"闭环：RelevanceChecker判断文档相关性，ResearchAgent生成答案，VerificationAgent进行事实核查，有效减少模型幻觉，以低代码成本实现具备自纠错能力的文档问答系统。

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 osimport hashlibimport picklefrom datetime import datetime, timedeltafrom pathlib import Pathfrom typing importListfrom docling.document_converter import DocumentConverterfrom langchain_text_splitters import MarkdownHeaderTextSplitterfrom config import constantsfrom config.settings import settingsfrom utils.logging import loggerclassDocumentProcessor: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_chunksdef_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.pyfrom langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.output_parsers import StrOutputParserfrom langchain_deepseek import ChatDeepSeekfrom config.settings import settingsclassRelevanceChecker: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 ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.output_parsers import StrOutputParserfrom typing importDict, Listfrom langchain_core.documents import Documentfrom langchain_deepseek import ChatDeepSeekimport loggingfrom config.settings import settingslogger = 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}")raisereturn {"draft_answer": answer,"context_used": context }

7、验证智能体

以下代码创建了一个VerificationAgent类，用于对 AI 生成的答案进行事实核查，以识别虚假信息。在__init__方法中，初始化了一个温度为 0（表示完全确定性）的 DeepSeek-V3.2 模型，创建了一个提示模板，指示 LLM 使用结构化的响应格式验证四个方面（直接事实支持、未经证实的说法、矛盾和相关性），然后构建了一个 LangChain 链。

在check()函数中，接收一个响应字符串和一个Document对象列表，将所有文档文本用双换行符连接成一个上下文字符串，然后使用响应字符串和上下文字符串调用链式调用以获取验证报告。在try-except块中，记录报告和上下文字符串以进行调试，最后返回一个包含验证报告和上下文字符串的字典，用于溯源。

from langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.output_parsers import StrOutputParserfrom typing importDict, Listfrom langchain_core.documents import Documentfrom langchain_deepseek import ChatDeepSeekimport loggingfrom config.settings import settingslogger = 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}")raisereturn {"verification_report": verification,"context_used": context }

​最后

我在一线科技企业深耕十二载，见证过太多因技术卡位而跃迁的案例。那些率先拥抱 AI 的同事，早已在效率与薪资上形成代际优势，我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在大模型的学习中的很多困惑。

我整理出这套 AI 大模型突围资料包：

• ✅AI大模型学习路线图
• ✅Agent行业报告
• ✅100集大模型视频教程
• ✅大模型书籍PDF
• ✅DeepSeek教程
• ✅AI产品经理入门资料

完整的大模型学习和面试资料已经上传带到CSDN的官方了，有需要的朋友可以扫描下方二维码免费领取【保证100%免费】👇👇
​​

为什么说现在普通人就业/升职加薪的首选是AI大模型？

人工智能技术的爆发式增长，正以不可逆转之势重塑就业市场版图。从DeepSeek等国产大模型引发的科技圈热议，到全国两会关于AI产业发展的政策聚焦，再到招聘会上排起的长队，AI的热度已从技术领域渗透到就业市场的每一个角落。

智联招聘的最新数据给出了最直观的印证：2025年2月，AI领域求职人数同比增幅突破200%，远超其他行业平均水平；整个人工智能行业的求职增速达到33.4%，位居各行业榜首，其中人工智能工程师岗位的求职热度更是飙升69.6%。

AI产业的快速扩张，也让人才供需矛盾愈发突出。麦肯锡报告明确预测，到2030年中国AI专业人才需求将达600万人，人才缺口可能高达400万人，这一缺口不仅存在于核心技术领域，更蔓延至产业应用的各个环节。

​​

资料包有什么？

①从入门到精通的全套视频教程⑤⑥

包含提示词工程、RAG、Agent等技术点

② AI大模型学习路线图（还有视频解说）

全过程AI大模型学习路线

③学习电子书籍和技术文档

市面上的大模型书籍确实太多了，这些是我精选出来的

④各大厂大模型面试题目详解

⑤ 这些资料真的有用吗?

这份资料由我和鲁为民博士共同整理，鲁为民博士先后获得了北京清华大学学士和美国加州理工学院博士学位，在包括IEEE Transactions等学术期刊和诸多国际会议上发表了超过50篇学术论文、取得了多项美国和中国发明专利，同时还斩获了吴文俊人工智能科学技术奖。目前我正在和鲁博士共同进行人工智能的研究。

所有的视频教程由智泊AI老师录制，且资料与智泊AI共享，相互补充。这份学习大礼包应该算是现在最全面的大模型学习资料了。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。

智泊AI始终秉持着“让每个人平等享受到优质教育资源”的育人理念‌，通过动态追踪大模型开发、数据标注伦理等前沿技术趋势‌，构建起"前沿课程+智能实训+精准就业"的高效培养体系。

课堂上不光教理论，还带着学员做了十多个真实项目。学员要亲自上手搞数据清洗、模型调优这些硬核操作，把课本知识变成真本事‌！

​​​​

如果说你是以下人群中的其中一类，都可以来智泊AI学习人工智能，找到高薪工作，一次小小的“投资”换来的是终身受益！

应届毕业生‌：无工作经验但想要系统学习AI大模型技术，期待通过实战项目掌握核心技术。

零基础转型‌：非技术背景但关注AI应用场景，计划通过低代码工具实现“AI+行业”跨界‌。

业务赋能 ‌突破瓶颈：传统开发者（Java/前端等）学习Transformer架构与LangChain框架，向AI全栈工程师转型‌。

👉获取方式：

😝有需要的小伙伴，可以保存图片到wx扫描二v码免费领取【保证100%免费】🆓**

​