Agentic-GraphRAG应用开发实战

课程说明：

体验课内容节选自《2025大模型Agent智能体开发实战》(12月班) 完整版付费课程

体验课时间有限，若想深度学习大模型技术，欢迎大家报名由我主讲的《2025大模型Agent智能体开发实战》(12月班)

《2025大模型Agent智能体开发实战》(12月班) 为【100+小时】体系大课，总共20大模块精讲精析，零基础直达大模型企业级应用！

课程完整介绍

部分课程成果演示

from IPython.display import Video

Dify+DeepSeek搭建智能微信语音客服

Video("https://ml2022.oss-cn-hangzhou.aliyuncs.com/2f1b47f42c65fd59e8d3a83e6cb9f13b_raw.mp4", width=800, height=400)

Coze自动图文视频创作流程

Video("https://ml2022.oss-cn-hangzhou.aliyuncs.com/Coze%E5%8A%A8%E6%80%81%E8%A7%86%E9%A2%91%E7%94%9F%E6%88%90%E5%AE%9E%E4%BE%8B.mp4", width=800, height=400)

可视化数据分析Multi-Agent

Video("https://ml2022.oss-cn-hangzhou.aliyuncs.com/%E5%8F%AF%E8%A7%86%E5%8C%96%E6%95%B0%E6%8D%AE%E5%88%86%E6%9E%90Multi-Agent%E6%95%88%E6%9E%9C%E6%BC%94%E7%A4%BA%E6%95%88%E6%9E%9C.mp4", width=800, height=400)

高效微调全自动数据集创建

Video("https://ml2022.oss-cn-hangzhou.aliyuncs.com/easy_daset_yanshi.mp4", width=800, height=400)

MateGen Pro 项目功能演示

Video("https://ml2022.oss-cn-hangzhou.aliyuncs.com/MG%E6%BC%94%E7%A4%BA%E8%A7%86%E9%A2%91.mp4", width=800, height=400)

智能客服项目展示

Video("https://ml2022.oss-cn-hangzhou.aliyuncs.com/%E6%99%BA%E8%83%BD%E5%AE%A2%E6%9C%8D%E6%A1%88%E4%BE%8B%E8%A7%86%E9%A2%91.mp4", width=800, height=400)

GraphRAG+多模态文档检索

Video("https://ml2022.oss-cn-hangzhou.aliyuncs.com/7%E6%9C%8817%E6%97%A5%281%29%20%E8%BF%9B%E5%BA%A6%E6%9D%A1.mp4", width=800, height=400)

此外，若是对大模型底层原理感兴趣，也欢迎报名由我和菜菜老师共同主讲的《2025大模型原理与实战课程》(秋季班)

详细信息扫码添加助教，回复“大模型”，即可领取课程大纲&查看课程详情👇

《大模型Agent智能体开发实战》12月班封班 · 体验课

垂直领域 Agentic-GraphRAG 开发实战

本节公开课，我们将带大家深入理解垂直领域智能问答系统的核心技术，从 RAG 痛点分析到 GraphRAG 解决方案，再到 Agentic-GraphRAG 的完整实现路径。通过理论讲解与实战演练相结合的方式，帮助大家掌握构建企业级智能问答系统的核心技能。

一、垂直领域 RAG 痛点与 GraphRAG 落地产品

RAG，Retrieval-Augmented Generation，也被称作检索增强生成技术，最早在 Facebook AI（Meta AI）在 2020 年发表的论文《Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks》（ https://arxiv.org/abs/2005.11401 ）中正式提出，这种方法的核心思想是借助一些文本检索策略，让大模型每次问答前都带入相关文本，以此来改善大模型回答时的准确性。这项技术刚发布时并未引发太大关注，而伴随2022年大模型技术大爆发，RAG技术才逐渐进入人们视野，并且由于早期大模型技术应用均已“知识库问答”为主，而RAG技术是最易上手、并且上限极高的技术，因此很快就成为了大模型技术人必备的技术之一。

论文地址：https://arxiv.org/pdf/2312.10997

Naive RAG最简单的RAG技术实现流程：需要围绕给定的文档（往往是非常长的文档）先进行切分，然后将切分的文档转化为计算机能识别的形式，也就是将其转化为一个数值型向量（也被称为词向量），然后当用户询问问题的时候，我们再将用户的问题转化为词向量，并和段落文档的词向量进行相似度匹配，借此找出和当前用户问题最相关的原始文档片段，然后将用户的问题和匹配的到的原文片段都带入大模型，进行最终的问答。由此便可实现一次完整的文档检索增强执行流程。

Naive RAG存在两个主要不足：其一，它难以捕捉跨文档的碎片化信息关联。当答案涉及多个文档的知识点时，纯粹基于语义相似度的检索往往无法将零散信息串联起来，导致答案不够全面。其二，随着数据规模增长，朴素 RAG 的性能和检索质量容易下降，可能出现无关内容干扰答案的情况。这在行业中被称为"上下文中毒"，即检索到的片段中混入了不相关的信息，导致模型生成的答案看似有道理却牛头不对马嘴。以上痛点在医疗、法律等垂直领域尤为突出——这些领域的问题往往要求结合多个证据点，多跳推理才能得出正确结论。

设想一个医疗研究员提出一个复杂问题，需要从多份医学报告中提取关联信息才能回答。如果仅靠传统搜索或单轮问答，很可能出现答案片面或遗漏关键细节的情况。在垂直领域（如医疗、金融），企业内部有大量专有知识，但直接让大模型了解这些专业资料往往困难重重。这就引出了检索增强生成（Retrieval-Augmented Generation，简称 RAG）技术的应用：通过检索外部知识来增强大模型的回答。然而很多团队发现，在垂直场景中直接采用朴素 RAG存在明显痛点，需要我们寻找更有效的改进方案。

针对上述挑战，GraphRAG 应运而生。GraphRAG 是将知识图谱引入 RAG 检索的一种新型技术路线，即 Graph + RAG。它通过从文本中提取实体及其关系，构建成图谱形式，以图结构来增强检索和推理能力。简单来说，GraphRAG 不再把文档视作孤立的句子向量集合，而是将其中蕴含的人物、事件、概念等实体"节点"挖掘出来，并根据关联构建"边"，形成一个语义网络。用户提问时，GraphRAG 可以识别问题涉及的实体并沿图谱关系寻找相关联的知识。这种方式能帮助模型获取更广阔而有结构的上下文视图，把分散在不同文件里的相关信息串联起来，特别适合需要多跳推理的复杂问答场景。

从通俗的角度来理解一下GraphRAG 和传统RAG 的区别。如下图所示：

GraphRAG 的运行机制

我们深入来看 GraphRAG 的运作机制。在索引阶段，GraphRAG 会对原始文档进行文本分析和图谱构建：利用大型模型或特定算法抽取出文中的实体（如人名、术语、数值等）及它们之间的关系（例如因果关系、组织结构、时间顺序等）。这些实体被作为图谱中的节点，关系作为边，可能还会附带权重或类型信息。接下来对大量节点进行语义聚类，将紧密相关的一组节点归入同一个"社区"，形成层次化的簇。这一过程可以看作对知识图谱的分块，方便后续快速锁定相关区域。

到了查询阶段，用户问题首先会通过自然语言处理识别其中的关键词或实体，然后 GraphRAG 算法会匹配图谱中相应的社区或子图。系统提取该子图的相关信息，通常包括社区内若干节点及它们的连接路径，并对这些内容进行局部摘要提炼。为了确保答案既准确又不遗漏全局背景，GraphRAG 常常采用分层汇总：先生成各相关社区的局部答案，再将它们合成为全局答案。微软研究提出的 GraphRAG 方案中，就结合了本地与全局两级摘要，并给每个答案片段打分筛选，最后通过多阶段递归（Map-Reduce）生成最终解答。这种图谱增强的流程显著提升了复杂查询时的表现：实验表明 GraphRAG 对复杂问答的准确性和响应质量相比传统 RAG 有大幅提升，某些场景下答案综合性能提升可达3倍以上。

GraphRAG 整个Indexing过程可以通过以下简单的方式来理解：

类似于 Baseline RAG，将源文档分块为较小的子文档；
执行两个并行提取：实体提取用于识别人名、地名、组织名等实体，关系提取：查找不同数据块中实体之间的关系，比如朋友、同事，员工等；
创建知识图谱，其中节点表示实体，边表示它们之间的关系，比如张三是李四的朋友，张三是王五的同事；
通过识别密切相关的实体来构建社区；
生成不同社区级别的分层摘要；
使用 reduce-map 方法通过逐步组合块来创建摘要，直到实现整体概览；

需要说明的是：GraphRAG 是一种技术范式，其目的是将知识图谱与大语言模型结合起来，以增强信息检索和生成能力。在这一技术范式下，衍生出非常多的开源项目、框架或工具会去实现这一流程。比较受欢迎的工具和项目主要如下：

热门GraphRAG项目

项目名称	描述	GitHub链接	论文链接
Microsoft GraphRAG	一个能够将知识图谱和 RAG 结合起来的数据工作流和转换工具，可以提供数据处理，检索问答能力	GitHub	论文
LightRAG	一个轻量级的RAG框架，支持图增强文本索引、增量更新算法等，比 Microsoft GraphRAG 更适合个人使用，成本相对较低	GitHub	论文
Fast-GraphRAG	更低延迟的GraphRAG实现，动态数据生成和增量更新等，成本约为Microsoft GraphRAG的1/6	GitHub	-
RAGFlow	基于深度文档理解构建的开源 RAG（Retrieval-Augmented Generation）引擎	GitHub	-
kotaemon	集成 GraphRAG 及混合检索等方法的具备本地知识库问答的项目	GitHub	-

GraphRAG 的优势固然突出，但我们也要看到其实现成本和局限性。

成本非常高

由于要构建和维护知识图谱，GraphRAG 在计算开销上远高于朴素 RAG：需要额外的LLM调用来抽取实体关系，并执行多轮查询和摘要，处理长文本时费用惊人（有报告称处理一本3.2万字的书用 GPT-4 构建图谱约耗费$6-$7 美元）。

运行时间过长

此外，GraphRAG 当前的流程偏离线批处理：如果我们添加新的知识文件，往往需要重新构建或更新整张知识图谱，无法像向量库那样即时增量更新。检索速度方面，由于涉及图数据库查询和复杂过滤，也会比直接向量近邻搜索更慢。

中间过程不可控

在学习成本上，GraphRAG 方案涉及 NLP 信息抽取、图数据库、检索算法等多方面知识，开发和调优难度较高。这些因素使得直接落地 GraphRAG 对很多企业来说门槛较高。为此，业界也在探索改进方案，例如 2024 年香港大学发布了LightRAG 项目，专注于通过轻量化的图谱简化提升效率。LightRAG 依然提取实体和关系但采用键值结构存储，增加去冗余合并节点的步骤，使知识图谱更紧凑。在检索阶段引入双层检索机制：既关注实体邻域精细答案，又考虑全局关系获取主题背景。得益于这些优化，LightRAG 号称将 GraphRAG 所需的 API 调用次数减少了约90%，整体 Token 开销缩减为 GraphRAG 的六千分之一！更重要的是，LightRAG 支持图谱的增量更新。从评测看，在大多数任务上LightRAG 的回答质量可媲美甚至略优于GraphRAG。总的来说，GraphRAG 和 LightRAG 的出现，表明通过知识图谱增强RAG是提升垂直领域问答质量的有效路径，但实现时需要权衡性能与成本。

二、Agentic-GraphRAG 技术方案

考虑到GraphRAG方案的复杂性，我们在垂直领域应用中可以采用更务实的实现方式。一个思路是借助大语言模型的Agent能力，结合知识图谱思路，实现Agentic-GraphRAG的效果。这也是本课程的核心内容：不是手动构建庞大的图数据库和复杂检索算法，而是利用 LangChain 等框架中现有组件，把知识检索、关系推理融合在智能 Agent 的决策过程中。

如上所示的技术架构就是本节课程我们会给大家详细讲解的技术实现思路。先通过OCR和信息抽取把文档转成结构化知识，然后把这些知识作为查询工具接入Agent。在问答时，Agent会自主选择是查询向量语义库还是查询构建的知识图谱，甚至调用计算工具，对结果进行验证。这种方案下，Agent相当于一个灵活的指挥官，可以根据问题难度自动规划多步检索。例如，对于简单直问，Agent直接语义检索获取答案；但若问题涉及推理整合，Agent 会先提取相关实体，再多步挖掘它们之间的关联，最后综合得到答案。我们将在后续章节通过实际代码演示如何组装这样的 Agent 管理流程。在进入技术细节前，请牢记：无论实现方式如何，RAG 思想始终是大模型应用的基石。即使是在 Agent 场景下，可靠的检索仍然是回答准确性的保障。因此理清 RAG 的原理和增强方法（如 GraphRAG）对我们开发任何垂直智能问答系统都至关重要。

从工程角度看，要实现Agentic-GraphRAG，我们需要解决几项关键技术：

其一，定义Agent可用的工具集，其中必然包括我们自建的垂直领域知识库检索接口（例如基于向量数据库或图数据库的查询函数）。

其二，设计Agent的Prompt模板，指导大模型何时该用哪个工具、如何解析工具返回结果。这通常采用ReAct（先思考再行动）或Plan-and-Execute等提示策略，使模型先输出思考步骤，再逐步调用动作。LangChain 1.1 提供了全新的 create_agent 接口，能够让开发者以更简洁的方式注册自定义工具并生成Agent，大大降低实现难度。我们可以在 Prompt 中列出如"SearchKnowledge"和"QueryGraph"等命令，模型就会在回答时尝试这些操作。本课程稍后会通过代码演示如何使用 LangChain 1.1 和 LangGraph 来打造这样一个能自主推理的 Agent。

其三，Agent得到最终答案后，我们还需关注来源溯源(Source Grounding)：确保回答中的每个关键事实都有据可查。这可以通过让Agent在输出答案时附上引用链接或文本片段来实现，也是验证Agent决策正确性的关键一环。总而言之，Agentic-GraphRAG 技术的优势在于融合了GraphRAG的深度检索与Agent的自主规划，使系统既"聪明"又"勤奋"。这为复杂业务场景下的大模型应用提供了一种可行方案，能够更有效地处理那些需要多步思考、多数据源整合的问题。

三、非结构化数据清洗：OCR 与 LangExtract 技术方案拆解

在一个医疗问答系统中，如果我们想让AI回答复杂的临床问题，就必须让它读懂海量的病历、检测报告；在一个金融分析应用里，AI需要消化各种财报、法规文件。然而，这些珍贵的信息往往以非结构化形式存在：扫描版PDF、图像表格、手写签字等等，直接输入给大模型显然不现实。如果我们不加处理就把这些内容喂给RAG系统，效果可想而知——OCR识别错误、文本杂乱无章，模型无法抓住重点。这一切凸显了数据清洗的重要性：只有将非结构化数据转化为结构化、可检索的知识，我们的Agentic-GraphRAG系统才能有效运转。

接下来，我们就需要首先从OCR识别和信息抽取两方面，拆解如何构建一个工业级的垂直领域文本清洗流水线。

3.1 OCR 数据解析

OCR（Optical Character Recognition，光学字符识别）技术用于将图像中的文字转化为数字文本，是非结构化数据清洗的第一步。如果把扫描文件比作埋在矿石中的金子，OCR 就是一台挖掘机，先将文字信息"挖掘"出来。但传统OCR往往只能产出"纯文本"，大量格式、表格结构和语义信息在这一过程中丢失。幸运的是，近年来出现了针对长文档和复杂版面的OCR新方案，例如 DeepSeek-OCR 等多模态模型。这些新OCR模型引入了视觉大模型，将整页图像编码为紧凑的视觉Token，再由轻量级的解码器输出文字。相比传统逐字识别，这种方法对版面理解更好、可以在更小模型上处理超长文档。简单来说，新一代OCR能更高效地将复杂版面压缩成可供大模型直接利用的文本语料。

现代文档不仅仅是文本。它们包含多列布局、数学公式、半扫描表格、多语言文本以及分辨率奇数的图表。像 GPT-4o 或 Qwen-VL 这样的端到端模型可以解析它们，但它们速度慢、布局混乱，并且耗费 GPU 内存。所以企业环境下往往会选择更小、更紧凑的视觉模型来为解析工作提供支持。主流的企业级OCR项目应用如下：

MinerU:点击进入

MinerU 是由 OpenDataLab（上海人工智能实验室下属团队）发起的一个开源工具，目标是将 PDF（含扫描件、复杂版式、多栏、多表格、多公式）转换为可机读的结构化格式（如 Markdown、JSON）以便进一步下游使用。项目的定位更偏「文档内容抽取／结构化」而不仅仅是传统 OCR。其取向是“将 PDF → Markdown/JSON”这一流程，而不仅“图片 → 文字”。

MinerU的主要工作流程分为以下几个阶段：

输入：接收PDF 格式文本，可以是简单的纯文本，也可以是包含双列文本、公式、表格或图等多模态PDF文件;
文档预处理（Document Preprocessing）：检查语言、页面大小、文件是否被扫描以及加密状态；
内容解析（Content Parsing）：
- 局分析：区分文本、表格和图像。
- 公式检测和识别：识别公式类型（内联、显示或忽略）并将其转换为 LaTeX 格式。
- 表格识别：以 HTML/LaTeX 格式输出表格。
- OCR：对扫描的 PDF 执行文本识别。
内容后处理（Content Post-processing）：修复文档解析后可能出现的问题。比如解决文本、图像、表格和公式块之间的重叠，并根据人类阅读模式重新排序内容，确保最终输出遵循自然的阅读顺序。
格式转换（Format Conversion）：以 Markdown 或 JSON 格式生成输出。
输出（Output）：高质量、结构良好的解析文档。

如下是MinerU项目官方给出的配置参考：

VLM-Transformer则是直接利用transformers库中的Vision-Language模型处理图像+文本的多模态输入，其流程如下：

而VLM-Sglang后端则是利用sglang高性能推理引擎，优化了GPU加速及分布式部署的基础上同时支持实时流式输出，其流程如下：

MinerU 项目非常适用于：

需要从大量 PDF 文档中抽取结构化内容（例如学术文献、技术白皮书、报告）用于知识库或训练语料。
对版式结构（如章节、列表、表格、公式）要求较高，而不只是 OCR 文本识别。
希望输出 Markdown／JSON 供后续自动化流水线使用。
PaddleOCR：点击进入

PaddleOCR 是由 Baidu (及其生态) 基于其深度学习框架 PaddlePaddle 提供的开源 OCR 工具箱。支持从 PDF 或图像文档转为结构化数据（适配 AI 场景），支持 100+ 语言。最新版本 3.0 在其技术报告中提出：PP-OCRv5、PP-StructureV3、PP-ChatOCRv4 三大解决方案，覆盖文字识别、多版式文档解析、关键信息提取。

在早期版本（如 PP-OCRv3）中，其结构可概括为：“检测 (Detection) → 分类 (Classification of orientation) → 识别 (Recognition)”。使用多种模型，例如检测模型（DBNet 等）、识别模型（如 SVTR），在 3.0 版本中，其“PP-StructureV3”整合了布局分析、表格识别、结构抽取。同时还最新推出了还推出了PaddleOCR-VL 的 Vision-Language 模型版本（0.9B 参数的 VLM），用于多语种文档解析。

PaddleOCR-VL 是推出的一个专注于“文档解析／视觉-语言模型 (Vision-Language Model, VLM)”功能的新模块，采用了视觉-语言模型架构以应对更高阶的需求。在解析多模态数据方面，PaddleOCR将这项工作分为两部分：

首先检测并排序布局元素。
使用紧凑的视觉语言模型精确识别每个元素。

该系统分为两个明确的阶段运行。

第一阶段是执行布局分析（PP-DocLayoutV2），此部分标识文本块、表格、公式和图表。它使用：

RT-DETR 用于物体检测（基本上是边界框 + 类标签）。
指针网络（6 个转换器层）可确定元素的读取顺序，从上到下、从左到右等。

最终输出统一模式的图片标注数据，如下图所示：

第二阶段则是元素识别（PaddleOCR-VL-0.9B），这就是视觉语言模型发挥作用的地方。它使用：

NaViT 风格编码器（来自 Keye-VL），可处理动态图像分辨率。无平铺，无拉伸。
一个简单的 2 层 MLP，用于将视觉特征与语言空间对齐。
ERNIE-4.5–0.3B 作为语言模型，该模型规模虽小但速度很快，并且采用 3D-RoPE 进行位置编码

最终模型输出结构化 Markdown 或 JSON 格式的文件用于后续的处理。

这个小小的设计决策，将布局和识别分离，使得 PaddleOCR-VL 比通常的一体化系统更快、更稳定。同时根据实际的测试，其运行和解析速度也更快。在 A100 GPU 上，吞吐量为 1.22 页/秒，。比 MinerU2.5 快 15.8%， VRAM 比 dots.ocr 少约 40%。

DeepSeek-OCR：点击进入

DeepSeek 于 2025年10月21日发布了 DeepSeek-OCR 模型，仅需7G的显存，就能完成高精度的表格、公式识别，图片语义识别，并且在多项评测指标中一举拿下SOTA成绩。其 Github 相较于其他项目相对比较"简陋"，仅仅提供了Transformers 和 vLLM 启动 DeepSeek-OCR的服务示例代码文件：

dots.ocr：点击进入

dots.ocr 是由 rednote‑hilab（HiLab团队）开源的多语种文档布局解析工具。官方介绍中强调：“一个统一的 Vision-Language 模型（≈1.7 B 参数）即可完成布局检测 + 内容识别 +阅读顺序排序”。支持文本、表格、公式、以及多语言输入。

dots.ocr 的特点是用一个 VLM（1.7 B 参数）来统一布局解析+内容识别，而不是传统将检测、识别、结构分开。用户可通过不同 prompt 来切换任务（如“请输出版式元素的 bbox、类别、文本”）→ 即说明模型采用 prompt + VLM 的方式。

非常适合需要快速处理多语种、混版式文档，且希望用一个统一模型／prompt 来搞定。虽然表现不错，但对于极复杂的表格（如跨页表、合并单元格）或特殊版式效果并不是很理想。

关于主流的MinerU、PaddleOCR-VL、DeepSeek-OCR等解析方案，均有本地部署版本，大家可以找到助教老师领取资料：

当然，无论OCR技术多先进，输出的文本仍然是非结构化的:段落杂糅、缺少标注。这就需要下一步的信息抽取来赋予其结构。

3.2 LangExtract信息抽取

信息抽取（Information Extraction）是将非结构化文本转化为结构化数据的过程。在传统NLP中，这通常包括命名实体识别（NER）、关系抽取（RE）、事件抽取等任务。然而，以往这些任务需要训练复杂模型、编写规则，对于垂直领域还面临着大量人工标注成本。

假设你是一家大型医院的数据分析师,每天医生会产生数千份病历记录。这些记录都是自由文本形式：

"患者张某,男,45岁,主诉胸痛3天。既往有高血压病史10年,目前服用缬沙坦80mg每日一次。体格检查:血压150/95mmHg..."

现在医院想要统计所有患者的用药情况、剂量分布、疾病共病模式,怎么做?

人工标注? 数千份病历,至少需要数周时间,且容易疲劳出错
正则表达式? 药物名称、剂量的表述千变万化,"每日一次"、"qd"、"1次/天"都是同一个意思
训练专用模型? 需要大量标注数据、算力资源,还要针对不同科室分别训练

这就是非结构化文本处理的核心痛点——信息价值高,但提取成本更高。

传统信息提取方法对比

方法类型	优点	局限性	适用场景
正则表达式/规则	精准、可控	规则复杂、维护困难、泛化能力差	格式固定的简单文本
传统NER模型	自动学习模式	需要大量标注数据、跨领域迁移差	有训练数据的特定领域
通用NLP工具	开箱即用	准确率低、无法定制	通用场景的粗粒度提取
人工标注	最准确	成本高、速度慢、无法规模化	小规模高价值数据

当然，除了 Microsoft GraphRAG 等项目，还有更轻量的工具推荐使用 - LangExtract。

LangExtract 是 Google 开源的一个 Python 库,它的核心使命可以用一句话概括:利用大型语言模型,将非结构化文本转换为结构化数据,并确保每个提取结果都精确对应回原文位置。

能力一:零代码定义提取任务(Prompt-Driven)

在传统方法中,提取药物信息可能需要编写数百行正则表达式或训练一个NER模型。而在 LangExtract 中,只需用自然语言描述任务:

FENCE0

能力二:少样本学习(Few-Shot Learning)

仅有 prompt 还不够,LangExtract 支持提供少量示例来"教会"模型输出的具体格式:

FENCE0

能力三:精确的来源定位(Source Grounding)

这是 LangExtract 最突出的特性之一。每个提取结果都会自动标注其在原文中的字符偏移位置。

这主要可以用于审查场景，例如：

医疗审计:监管部门需要核查AI提取的诊断结论是否真的出现在病历中
法律合规:合规审查时必须能追溯每条规定来自合同的哪一页哪一行
知识管理:构建企业知识库时,需要链接回原始文档以供查证
能力四:结构化输出保证(Schema Enforcement)

直接使用大模型的最大的痛点之一是输出格式不可控。同一个任务,模型可能这次返回 JSON,下次返回表格,再下次又变成了自然语言描述。LangExtract 通过两种机制保证输出的结构化:

1. 基于示例的格式约束

通过 Few-Shot 示例,模型会学习并模仿示例的结构。这在所有模型上都有效。

2. 受控生成(Controlled Generation)

对于支持的模型(如 Google Gemini),LangExtract 会利用模型原生的 schema 约束功能,强制要求输出必须符合预定义的 JSON Schema。在实际应用中,可以批量处理成千上万份文档,无需担心某一份的输出格式突然出错导致整个流程崩溃。

能力五:长文档多轮提取(Multi-Pass Extraction)

大模型都有上下文长度限制。即使是支持 100K tokens 的模型,面对一份 500 页的法律合同也会捉襟见肘。更重要的是,长文档中的信息密度高,单次扫描很容易遗漏细节。LangExtract 采用了智能的处理策略:

1. 自动分块(Automatic Chunking)

将长文档切分为适合模型处理的小块
保留块与块之间的上下文重叠,避免信息断层
并行处理多个分块,提高速度

2. 多轮提取(Multi-Pass)

第一轮:全面扫描,提取明显的信息
第二轮:针对第一轮可能遗漏的内容,调整 prompt 再次扫描
第三轮及以后:可以针对特定类型的信息进行深度挖掘

这种策略在合规审查、法律分析等不允许遗漏任何细节的场景中至关重要。

四、【实战】Prompt Engineering 对比 LangExtract

上面我们已经了解了 LangExtract 的基本原理和核心能力。但在实际项目中，可能大部分同学都会有疑问："我直接用提示词让大模型提取不就行了吗？为什么还要用 LangExtract？"所以，我们就先通过一个真实的新闻信息提取场景，先运行纯提示词方式，看到它的提取结果，再运行 LangExtract 方式，看到它的提取结果，最后进行深度对比分析（重复实体、完整性、原文对齐能力）。

首先需要安装langextract库：

%pip install langextract

%pip install openai python-dotenv

Step 1.导入所有依赖库

首先我们导入本章需要的所有 Python 库。注意这里同时导入了纯提示词所需的 OpenAI 客户端和 LangExtract 框架，

# 导入所有需要的库
import os
import time
import json
from pathlib import Path
from collections import Counter

# 纯提示词方式所需
from openai import OpenAI

# LangExtract 方式所需
import langextract as lx
from langextract.providers.openai import OpenAILanguageModel
from langextract.prompt_validation import PromptValidationLevel

# 环境变量加载
from dotenv import load_dotenv

print("所有依赖库导入成功")

可以看到，所有库导入成功。如果这里报错，请确保已安装：pip install langextract openai python-dotenv

Step 2：加载环境变量与 API 配置

接下来加载 DeepSeek API 的配置。这一步确保两种方案使用完全相同的模型和配置。

# 加载环境变量
load_dotenv()

# 读取 API 配置
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
DEEPSEEK_BASE_URL = os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com")
DEFAULT_MODEL = os.getenv("DEFAULT_MODEL", "deepseek-chat")

# 创建输出目录
OUTPUT_DIR = Path("./outputs")
OUTPUT_DIR.mkdir(exist_ok=True)

print("环境变量加载完成")
print(f"  模型: {DEFAULT_MODEL}")
print(f"  API: {DEEPSEEK_BASE_URL}")

API 配置已加载。可以看到我们将使用相同的 DeepSeek 模型进行对比。

Step 3：准备测试文本

接下来定义测试文本。这是一段真实新闻风格的复杂文本，包含多个时间、机构、人物、地点、事件和指标，用于压力测试两种方案。

# 定义测试文本
input_text = """
2025年12月22日上午，国家统计局在北京国务院新闻办公室举行新闻发布会，公布2025年前11个月国民经济运行情况。
国家统计局新闻发言人付凌晖表示，规模以上工业增加值同比增长6.1%，社会消费品零售总额增长7.3%。

同日，国家发展改革委（以下简称"发改委"）在例行发布会上介绍，将于2026年起对"人工智能+制造"试点城市给予专项资金支持，首批覆盖上海、深圳、成都等10个城市。
发改委副主任李春临称，资金将重点投向算力基础设施和工业软件，并与地方财政配套安排相衔接。

22日傍晚，中国人民银行（央行）公告，自12月23日起下调金融机构存款准备金率0.25个百分点。
央行副行长宣昌能在答记者问时称，此举旨在保持流动性合理充裕，并支持中小微企业融资。

对此，清华大学经济管理学院教授李稻葵认为，上述措施有助于稳定市场预期，但需警惕部分行业"低价竞争"风险。
他同时提到，地方政府隐性债务化解仍是明年宏观政策的重要约束。
""".strip()

print("✓ 测试文本已准备")
print(f"文本长度: {len(input_text)} 字符")
print(f"\n原文预览（前200字符）：\n{input_text[:200]}...")

测试文本已准备好。这是一段高信息密度的新闻文本，接下来将用它来测试两种方案。

重点关注：文本中包含"2026年起"、"12月23日起"等时间表达，我们将观察两种方案如何处理它们。

接下来我们运行纯提示词方式。这个方案直接调用 DeepSeek API，通过精心设计的 Prompt 来提取信息。

Step 4：准备 Few-shot 示例

首先定义 Few-shot 示例。这个示例会告诉大模型我们要提取哪些类别、输出什么格式。

# 定义 Few-shot 示例文本
example_text = (
    "2025年6月3日，工业和信息化部在北京发布《算力基础设施高质量发展行动计划》。"
    "工信部副部长张云明表示，到2027年全国算力总规模将达到300 EFLOPS。"
)

# 定义示例的标准输出格式
example_output = """{
  "extractions": [
    {"class": "时间", "text": "2025年6月3日"},
    {"class": "机构", "text": "工业和信息化部"},
    {"class": "地点", "text": "北京"},
    {"class": "人物", "text": "张云明"},
    {"class": "事件", "text": "发布《算力基础设施高质量发展行动计划》"},
    {"class": "指标", "text": "300 EFLOPS"}
  ]
}"""

print("Few-shot 示例已准备")
print(f"示例文本: {example_text}")
print(f"\n示例输出:\n{example_output}")

Few-shot 示例定义完成。这个示例包含了所有实体类别（时间、地点、机构、人物、事件、指标），可以有效引导 LLM 的提取行为。

Step 5：构建完整的 Prompt

接下来构建完整的 Prompt。这个 Prompt 包含任务说明、输出格式要求、Few-shot 示例和待提取的文本。

# 构建完整 Prompt
prompt = f"""
请从下面的新闻文本中提取结构化信息。

【抽取类别】
- 时间
- 地点
- 机构
- 人物
- 事件
- 指标（数值/增速/比例/数量等）

【要求】
1. 所有 `text` 必须是原文中的精确子串，不要改写
2. 去重并按原文出现顺序输出

【输出格式】
请严格输出 JSON（不要添加任何解释、Markdown、代码块）：
{{
  "extractions": [
    {{
      "class": "类别",
      "text": "原文精确文本"
    }}
  ]
}}

【Few-shot 示例】
示例文本：
{example_text}

示例输出：
{example_output}

---

现在请抽取这段新闻文本：
{input_text}
"""

print("Prompt 构建完成")
print(f"Prompt 总长度: {len(prompt)} 字符")

Prompt 构建完成。接下来我们将用这个 Prompt 调用 DeepSeek API。

Step 6：创建 DeepSeek 客户端

创建 OpenAI 客户端（兼容 DeepSeek API）。

# 创建 DeepSeek 客户端
client = OpenAI(
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL
)

print("DeepSeek 客户端创建成功")

客户端创建成功。接下来调用 API 进行信息提取。

Step 7. 调用 DeepSeek API 进行提取

现在正式调用大模型，让它按照我们的 Prompt 提取信息。这一步会发送网络请求，通常耗时 2-5 秒。

# 调用 DeepSeek API
print("正在调用 DeepSeek API...")
start_time = time.time()

response = client.chat.completions.create(
    model=DEFAULT_MODEL,
    messages=[
        {
            "role": "system",
            "content": "你是一个文本信息抽取助手，请严格按照用户指定的 JSON 格式输出，不要添加多余解释。"
        },
        {
            "role": "user",
            "content": prompt
        }
    ],
    temperature=0.3,
    max_tokens=2000,
    stream=False
)

result_prompt = response.choices[0].message.content
elapsed = time.time() - start_time

print(f"提取完成（耗时 {elapsed:.2f} 秒）")

API 调用成功！接下来查看提取结果。

Step 8. 查看纯提示词的提取结果

打印大模型返回的原始结果。

# 打印原始结果
print("纯提示词提取结果:")
print("=" * 80)
print(result_prompt)
print("=" * 80)

可以看到，大模型返回了一个 JSON 格式的提取结果。我们解析并统计纯提示词的结果：

# 解析 JSON 并统计
prompt_result = json.loads(result_prompt)
prompt_extractions = prompt_result["extractions"]

# 基本统计
total_count_prompt = len(prompt_extractions)
type_counts_prompt = Counter(ext["class"] for ext in prompt_extractions)

print(f"纯提示词统计:")
print(f"  总实体数: {total_count_prompt}")
print(f"  按类型分布:")
for entity_type, count in sorted(type_counts_prompt.items()):
    print(f"    - {entity_type}: {count} 个")

统计完成。可以看到纯提示词提取了大量实体。但数量多不代表质量高——接下来我们将检查重复问题。

Step 10. 检查纯提示词的重复问题

检查是否存在重复提取的实体。

# 检查重复
text_counts = Counter(ext["text"] for ext in prompt_extractions)
duplicates = {text: count for text, count in text_counts.items() if count > 1}

if duplicates:
    print("发现重复实体:")
    for text, count in duplicates.items():
        print(f"  - '{text}' 重复了 {count} 次")
    
    total_duplicates = sum(count - 1 for count in duplicates.values())
    unique_count_prompt = total_count_prompt - total_duplicates
    
    print(f"\n去重后实际数量: {unique_count_prompt} 个（减少了 {total_duplicates} 个重复）")
else:
    print("✓ 未发现重复实体")
    unique_count_prompt = total_count_prompt

可以看到，纯提示词存在明显的重复问题！这会导致：

数量虚高
需要人工去重

接下来我们运行 LangExtract 方式。这个方案使用 LangExtract 框架，它会自动处理去重、原文对齐等细节。

Step 11：定义 LangExtract 的任务描述

LangExtract 的 Prompt 比纯提示词简洁得多，因为它内置了很多最佳实践。

# 定义 LangExtract 的任务描述
langextract_prompt = """
从新闻文本中提取以下信息：
- 时间
- 地点
- 机构
- 人物
- 事件
- 指标（数值/增速/比例/数量等）

要求：
1. 使用原文中的完整表述
2. 不要重复
3. 按出现顺序提取
"""

print("LangExtract 任务描述已定义")
print(f"Prompt 长度: {len(langextract_prompt)} 字符（比纯提示词简洁得多）")

任务描述定义完成。注意它比纯提示词的 Prompt 简洁很多，因为 LangExtract 会自动处理输出格式、去重等细节。

Step 12. 定义 LangExtract 的 Few-shot 示例

使用 LangExtract 的数据结构定义 Few-shot 示例（与纯提示词使用完全相同的示例文本）。

# 定义 LangExtract 的 Few-shot 示例
examples = [
    lx.data.ExampleData(
        text=example_text,
        extractions=[
            lx.data.Extraction("时间", "2025年6月3日"),
            lx.data.Extraction("机构", "工业和信息化部"),
            lx.data.Extraction("地点", "北京"),
            lx.data.Extraction("人物", "张云明"),
            lx.data.Extraction("事件", "发布《算力基础设施高质量发展行动计划》"),
            lx.data.Extraction("指标", "300 EFLOPS"),
        ]
    )
]

print("LangExtract Few-shot 示例已定义")
print(f"示例数量: {len(examples)}")
print(f"示例实体数: {len(examples[0].extractions)}")

Few-shot 示例定义完成。注意这里使用了与纯提示词完全相同的示例文本和标准答案，确保对比的公平性。

Step 13：初始化 LangExtract 模型

创建 LangExtract 的模型实例（使用相同的 DeepSeek API）。

# 创建 LangExtract 模型
model = OpenAILanguageModel(
    model_id=DEFAULT_MODEL,
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL
)

print("LangExtract 模型创建成功")
print(f"使用模型: {DEFAULT_MODEL}（与纯提示词相同）")

模型创建成功。此时 LangExtract 已准备好调用 DeepSeek API。

Step 14：执行 LangExtract 提取

现在正式执行 LangExtract 提取。这一步会自动完成：构建优化的 Prompt、调用 LLM、解析结果、对齐原文位置、自动去重。

result = lx.extract(
    text_or_documents=input_text,
    prompt_description=prompt,
    examples=examples,
    model=model,
    fence_output=True,
    use_schema_constraints=False,
    prompt_validation_level=PromptValidationLevel.OFF,  

)

LangExtract 提取完成！接下来查看结果。

Step 15. 查看 LangExtract 的提取结果

打印 LangExtract 的提取结果。注意观察每个实体是否有精确的位置信息。

# 打印 LangExtract 结果
print("LangExtract 提取结果:")
print("=" * 80)
for ext in result.extractions:
    if ext.char_interval:
        pos_info = f"[{ext.char_interval.start_pos}-{ext.char_interval.end_pos}]"
    print(f"[{ext.extraction_class}] {ext.extraction_text} {pos_info}")
print("=" * 80)

可以看到，LangExtract 的每个实体都有精确的位置信息！这意味着我们可以回到原文验证、可视化高亮展示。

Step 16. 统计 LangExtract 的结果

统计 LangExtract 提取了多少个实体，并按类型分布。

# 统计 LangExtract 结果
langextract_extractions = result.extractions
total_count_lang = len(langextract_extractions)
type_counts_lang = Counter(ext.extraction_class for ext in langextract_extractions)

# 统计有位置信息的实体数
with_position = sum(1 for ext in langextract_extractions if ext.char_interval is not None)

print(f"LangExtract 统计:")
print(f"  总实体数: {total_count_lang}")
print(f"  按类型分布:")
for entity_type, count in sorted(type_counts_lang.items()):
    print(f"    - {entity_type}: {count} 个")

[demo:graph1:打开对比展示]

五、LangExtract 长文本提取实战

在前面的章节中，我们已经学习了 LangExtract 的基本用法和与纯提示词的对比。本章我们将深入探讨 LangExtract 处理长文本的核心能力。

我们将使用完整的《罗密欧与朱丽叶》中文版（约 54,000 字符）作为案例，演示如何从长文档中提取人物、情感和关系信息。

本章核心要点：

长文本处理策略：分块 + 多轮 + 并行
戏剧文本特点：角色对话、情感丰富、关系复杂
数据分析与可视化：从 1,889 个实体中挖掘价值
核心参数调优：extraction_passes、max_workers、max_char_buffer

5.1 配置运行环境

首先我们需要导入必要的 Python 库。这一步非常基础，但如果依赖缺失，后续所有代码都将无法运行。

我们需要导入的库包括：

os 和 Path：文件和目录操作
textwrap：格式化多行字符串
Counter 和 defaultdict：统计和数据聚合
langextract：核心提取框架
OpenAILanguageModel：DeepSeek 兼容的模型接口
dotenv：加载环境变量

import os
import textwrap
from pathlib import Path
from collections import Counter, defaultdict

from dotenv import load_dotenv
import langextract as lx
from langextract.providers.openai import OpenAILanguageModel
from langextract.prompt_validation import PromptValidationLevel

print("所有依赖库导入成功！")

# 加载环境变量
load_dotenv()

print("环境变量加载成功！")

load_dotenv() 会从项目根目录的 .env 文件中读取配置信息，主要是 API Key 和相关配置。

接下来，我们配置 API 相关的参数。

# 配置 DeepSeek API
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY", "")
DEEPSEEK_BASE_URL = os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com")
DEFAULT_MODEL = os.getenv("DEFAULT_MODEL", "deepseek-chat")

# 配置输出目录
OUTPUT_DIR = Path("./outputs")
OUTPUT_DIR.mkdir(exist_ok=True)

print(f"配置完成！")
print(f"  模型: {DEFAULT_MODEL}")
print(f"  API 端点: {DEEPSEEK_BASE_URL}")
print(f"  输出目录: {OUTPUT_DIR}")

这里我们完成了三件事：

读取 API 配置：从环境变量中获取 DeepSeek API Key 和端点
设置默认模型：使用 deepseek-chat 模型
创建输出目录：OUTPUT_DIR.mkdir(exist_ok=True) 会创建 outputs 目录，如果已存在则不报错

准备工作完成后，接下来进入核心部分。

5.2 读取长文本数据

我们将从本地文件读取《罗密欧与朱丽叶》的完整中文文本。这个文本约 54,000 字符，是一个典型的长文档。

由于中文文本可能有不同的编码格式，我们需要尝试多种编码来读取文件。

# 读取《罗密欧与朱丽叶》中文文本
romeo_file = Path("罗密欧与朱丽叶.txt")

if romeo_file.exists():
    # 尝试多种编码
    for encoding in ['gbk', 'gb2312', 'utf-8', 'utf-16']:
        try:
            with open(romeo_file, 'r', encoding=encoding) as f:
                input_text = f.read()
            print(f"✓ 成功使用 {encoding} 编码读取文件")
            break
        except:
            continue
else:
    print("错误: 未找到 罗密欧与朱丽叶.txt 文件")
    print("请确保文件存在于当前目录")

print(f"✓ 成功读取 {len(input_text):,} 字符")
print(f"\n文本预览（前 200 字符）：")
print("-" * 60)
print(input_text[:200])
print("...")

可以看到，文件成功读取。这里有几个关键点：

多编码尝试：我们依次尝试 gbk、gb2312、utf-8、utf-16，确保能正确读取中文文本
字符数统计：约 54,000 字符，这是一个典型的长文档场景
文本预览：通过 input_text[:200] 查看前 200 字符，确认文本格式正确

接下来，我们需要定义提取任务的 Prompt 描述。

4.3 定义提取任务 Prompt

Prompt 是告诉模型"要提取什么"的核心指令。对于戏剧文本，我们关注三类信息：

人物：剧中角色
情感：情感表达和心理活动
关系：人物之间的关系描述

同时，我们要明确告诉模型戏剧脚本的特点：角色名称后跟冒号。

# 定义提取任务的 Prompt 描述
prompt = textwrap.dedent("""\
    从中文戏剧文本中提取人物、情感和关系信息。
    
    为每个实体添加有意义的属性以增加上下文。
    
    重要: extraction_text 必须是原文的精确子串，不要改写。
    按出现顺序提取实体，不要重叠。
    
    注意: 在戏剧脚本中，角色名称后跟冒号。""")

print("Prompt 定义完成！")
print("\n" + "=" * 60)
print("Prompt 内容：")
print("=" * 60)
print(prompt)

这个 Prompt 包含了几个关键要求：

提取目标：人物、情感、关系
属性要求：为每个实体添加属性，增加上下文信息
精确对齐：extraction_text 必须是原文精确子串
戏剧格式：角色名称后跟冒号，这是戏剧脚本的特殊格式

接下来，我们需要提供 Few-shot 示例，引导大模型正确提取。

5.4 定义 Few-shot 示例

Few-shot 示例是 LangExtract 的核心。通过提供高质量的示例，我们可以引导模型理解"什么是好的提取"。我们将使用一段经典的罗密欧与朱丽叶对话作为示例。

# 定义 Few-shot 示例
examples = [
    lx.data.ExampleData(
        text=textwrap.dedent("""\
            罗密欧: 轻声！那边窗子里亮起来的是什么光？
            那就是东方，朱丽叶就是太阳。
            朱丽叶: 啊！罗密欧，罗密欧！为什么你偏偏是罗密欧呢？"""),
        extractions=[
            lx.data.Extraction(
                extraction_class="人物",
                extraction_text="罗密欧",
                attributes={"情感状态": "惊叹"}
            ),
            lx.data.Extraction(
                extraction_class="情感",
                extraction_text="轻声！",
                attributes={"情感": "温柔敬畏", "人物": "罗密欧"}
            ),
            lx.data.Extraction(
                extraction_class="关系",
                extraction_text="朱丽叶就是太阳",
                attributes={"类型": "比喻", "人物1": "罗密欧", "人物2": "朱丽叶"}
            ),
            lx.data.Extraction(
                extraction_class="人物",
                extraction_text="朱丽叶",
                attributes={"情感状态": "渴望"}
            ),
            lx.data.Extraction(
                extraction_class="情感",
                extraction_text="为什么你偏偏是罗密欧呢？",
                attributes={"情感": "渴望的疑问", "人物": "朱丽叶"}
            ),
        ]
    )
]

print("Few-shot 示例定义完成！")
print(f"\n示例数量: {len(examples)}")
print(f"示例文本长度: {len(examples[0].text)} 字符")
print(f"示例提取数量: {len(examples[0].extractions)} 个")
print("\n" + "=" * 60)
print("示例文本：")
print("=" * 60)
print(examples[0].text)

这个 Few-shot 示例展示了高质量提取的标准：

人物提取：提取"罗密欧"和"朱丽叶"，并添加"情感状态"属性
情感提取：提取"轻声！"和"为什么你偏偏是罗密欧呢？"，标注情感类型和归属人物
关系提取：提取"朱丽叶就是太阳"这个比喻，标注类型和涉及的人物

注意每个 extraction_text 都是原文的精确子串，这是 LangExtract 的核心要求。

准备工作完成，接下来创建模型。

5.5 创建 LangExtract 模型

我们使用 OpenAILanguageModel 来对接 DeepSeek API。LangExtract 的设计让我们可以轻松地使用任何 OpenAI 兼容的 API。

# 创建 DeepSeek 模型实例
model = OpenAILanguageModel(
    model_id=DEFAULT_MODEL,
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL
)

print("模型创建成功！")
print(f"  模型 ID: {DEFAULT_MODEL}")
print(f"  API 端点: {DEEPSEEK_BASE_URL}")

模型实例创建完成。这里我们传入了三个参数：

model_id：模型名称，这里是 deepseek-chat
api_key：API 密钥，从环境变量读取
base_url：API 端点，DeepSeek 的 API 端点

接下来进入最关键的部分：运行长文本提取。

5.6 运行长文本提取（核心）

这是本章的核心部分。我们将使用 lx.extract() 函数，配合长文本处理的三大核心参数：

长文本处理核心参数

参数名	值	作用
`extraction_passes`	3	多轮提取，提高召回率
`max_workers`	20	并行处理，加快速度
`max_char_buffer`	1000	分块大小，保持准确性

这三个参数是 LangExtract 处理长文本的核心策略：分块 + 多轮 + 并行。

注意：运行这个 Cell 会调用 DeepSeek API，处理 54,000 字符的文本，大约需要 1-2 分钟，请耐心等待。

# 运行长文本提取
print("开始运行长文本提取...")
print(f"   输入文本: {len(input_text):,} 字符")
print(f"   提取轮数: 3 轮")
print(f"   并行数: 20 workers")
print(f"   分块大小: 1000 字符")
print("\n正在处理，请稍候...\n")

result = lx.extract(
    text_or_documents=input_text,
    prompt_description=prompt,
    examples=examples,
    model=model,
    extraction_passes=3,      # 多轮提取提高召回率
    max_workers=20,           # 并行处理加速
    max_char_buffer=1000      # 较小的上下文提高准确性
)

print("\n" + "=" * 60)
print("提取完成！")
print("=" * 60)
print(f"文本长度: {len(result.text):,} 字符")
print(f"提取实体总数: {len(result.extractions)} 个")

提取完成！从 54,000 字符的文本中，我们提取了约 1,889 个实体。

这里发生了什么？让我们理解 LangExtract 的长文本处理策略：

分块处理：max_char_buffer=1000 将文本分成多个 ~1000 字符的块
并行执行：max_workers=20 同时处理 20 个块，大幅提速
多轮提取：extraction_passes=3 执行 3 轮独立提取，将不重叠的结果合并

[demo:graph2:打开对比展示]

5.7 保存提取结果

LangExtract 使用 JSONL 格式保存结果。JSONL 是一种人类可读的格式，每行是一个独立的 JSON 对象，便于解析和分享。

# 保存提取结果为 JSONL 格式
lx.io.save_annotated_documents(
    [result], 
    output_name="罗密欧与朱丽叶_extractions.jsonl", 
    output_dir=str(OUTPUT_DIR)
)

output_jsonl_path = OUTPUT_DIR / "罗密欧与朱丽叶_extractions.jsonl"
print(f"结果已保存: {output_jsonl_path}")

JSONL 文件已保存到 outputs 目录。这个文件包含了：

原始文本
所有提取的实体
每个实体的位置信息（start_pos、end_pos）
每个实体的属性

接下来，我们生成交互式可视化。LangExtract 提供了强大的可视化功能。通过 lx.visualize() 函数，我们可以生成一个交互式 HTML 页面，用于查看和分析提取结果。

# 生成交互式可视化
html_content = lx.visualize(str(output_jsonl_path))

html_file = OUTPUT_DIR / "罗密欧与朱丽叶_visualization.html"
with open(html_file, "w", encoding="utf-8") as f:
    if hasattr(html_content, 'data'):
        f.write(html_content.data)
    else:
        f.write(html_content)

print(f"交互式可视化已生成: {html_file}")
print("\n提示: 在浏览器中打开 HTML 文件可以:")
print("   - 查看高亮显示的所有提取结果")
print("   - 按类别筛选实体")
print("   - 查看每个实体的详细属性")

可视化 HTML 文件已生成。这个文件提供了：

原文高亮显示
按类别筛选
点击实体查看详细信息
可搜索、可导航

[demo:graph3:打开对比展示]

六、Agentic GraphRAG 构建与溯源实战

在前面的章节中，我们已经学习了 LangExtract 的基本用法、与纯提示词的对比、以及长文本处理能力。本章我们将进入一个更加实战化的场景：如何构建一个带溯源能力的 Agentic GraphRAG 系统。

这个系统的完整流程包括：

PDF 解析：使用 MinerU API 将 PDF 转换为结构化的 Markdown
知识提取：使用 LangExtract 提取实体、关系、属性，并保留原文位置信息
向量存储：使用 ChromaDB 存储提取结果的向量表示
知识图谱构建：从提取结果构建实体-关系图谱
Agent 问答：构建智能问答系统，并实现完整的溯源链路

本章的核心亮点是：每一个提取的实体，都能精确定位到原文的字符位置，实现真正的可追溯、可验证。

说明：这是一个完整的生产级系统架构，适合用于企业文档问答、法律文书分析、学术论文检索等场景。

6 .1 环境准备与依赖导入

在开始构建系统之前，我们需要先理解每个依赖库的作用：

langextract：核心提取库，负责从文本中提取结构化信息
requests：用于调用 MinerU API 进行 PDF 解析
chromadb + langchain-chroma：向量数据库，存储和检索提取结果
langchain-openai：提供 Embeddings 向量化能力
openai：调用 DeepSeek 等兼容 OpenAI 接口的 LLM

接下来我们逐步导入这些依赖。注意，我们会分批导入，每一批都有明确的功能分组。

步骤 1：导入标准库

首先导入 Python 标准库，这些是进行文件操作、时间统计、数据处理的基础工具。

# 标准库导入
import os
import json
import textwrap
import time
import zipfile
import io
import uuid
from pathlib import Path
from typing import Optional
from dataclasses import dataclass, field, asdict
from collections import defaultdict, Counter

print("标准库导入成功")

执行结果说明：

pathlib.Path：用于跨平台的文件路径处理
dataclass：用于定义数据结构类，简化代码
uuid：为每个向量生成唯一 ID
Counter：统计实体类型分布

注意：这里没有报错就说明环境正常，可以继续。

步骤 2：加载环境变量

在实际项目中，API Key 等敏感信息不应该硬编码在代码中，而应该通过 .env 文件管理。这里我们使用 python-dotenv 来加载环境变量。

在使用 DeepSeek 和 MinerU 之前，我们需要获取并配置 API 密钥。

DeepSeek API Key：访问 DeepSeek 开放平台，注册账号后获取 API Key；
MinerU API Key：访问 MinerU 官网，需要先进行API申请，然后创建 Token 即可。

# 加载环境变量
try:
    from dotenv import load_dotenv
    load_dotenv()
    print("已加载 .env 配置文件")
except ImportError:
    print("未安装 python-dotenv，将使用系统环境变量")

执行结果说明：

如果显示 已加载 .env 配置文件，说明环境变量加载成功
如果显示警告信息，说明 python-dotenv 未安装，但程序会尝试使用系统环境变量

提示：.env 文件应该包含 MINERU_API_KEY、DEEPSEEK_API_KEY、DASHSCOPE_API_KEY 三个配置。

步骤 3：导入第三方核心库

现在导入项目的核心依赖：HTTP 请求库和 LangExtract。

# HTTP 请求库（用于调用 MinerU API）
import requests

# LangExtract 核心库
import langextract as lx
from langextract.providers.openai import OpenAILanguageModel
from langextract.prompt_validation import PromptValidationLevel

print("LangExtract 导入成功")
print(f"  版本信息: {lx.__version__ if hasattr(lx, '__version__') else '未知'}")

执行结果说明：

OpenAILanguageModel：LangExtract 的模型适配器，支持 DeepSeek 等兼容 OpenAI 接口的 LLM
PromptValidationLevel：控制提示词验证级别，我们会设置为 OFF 以提高灵活性

说明：如果这一步报错，请检查是否已安装 langextract 库。

步骤 4：导入向量数据库相关库

向量数据库是 RAG 系统的核心组件，用于存储和检索向量化的知识。我们使用 ChromaDB 作为向量数据库，LangChain 提供集成接口。

%pip install chromadb langchain-chroma langchain_openai

# ChromaDB 向量数据库
import chromadb
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_chroma import Chroma

print("ChromaDB 和 LangChain 导入成功")

执行结果说明：

chromadb：轻量级向量数据库，支持持久化存储
OpenAIEmbeddings：向量化工具，将文本转换为向量表示
Chroma：LangChain 对 ChromaDB 的封装，提供更友好的 API

注意：如果这一步报错，请确保已安装 chromadb、langchain-chroma、langchain-openai。

6.2 系统配置与参数定义

一个生产级系统需要清晰的配置管理。我们将所有的 API Key、模型参数、文件路径等配置项集中定义，便于后续调整和维护。

配置项分为四类：

MinerU 配置：PDF 解析服务
DeepSeek 配置：用于 LangExtract 提取
阿里云百炼配置：用于向量化（Embeddings）
ChromaDB 配置：向量数据库存储路径

步骤 1：定义 MinerU API 配置

MinerU 是一个 PDF 解析服务，能够将 PDF 转换为结构化的 Markdown 格式，保留文档的标题、段落、表格等结构信息。

# MinerU API 配置
MINERU_API_KEY = os.getenv("MINERU_API_KEY", "")
MINERU_BASE_URL = "https://mineru.net"

print("MinerU 配置:")
if MINERU_API_KEY:
    print(f"  API Key: {MINERU_API_KEY[:10]}... (已配置)")
    print(f"  API URL: {MINERU_BASE_URL}")
else:
    print(" API Key 未配置")

执行结果说明：

如果显示 API Key: eyJ0eXBlIj... (已配置)，说明 MinerU API Key 配置成功
如果显示 API Key 未配置，说明需要在 .env 文件中添加 MINERU_API_KEY
步骤 2：定义 DeepSeek LLM 配置

DeepSeek 是一个兼容 OpenAI 接口的大语言模型，我们用它来驱动 LangExtract 进行知识提取，以及后续的 Agent 问答。

# DeepSeek LLM 配置
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY", "")
DEEPSEEK_BASE_URL = os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com")
DEFAULT_MODEL = os.getenv("DEFAULT_MODEL", "deepseek-chat")

print("\nDeepSeek 配置:")
if DEEPSEEK_API_KEY:
    print(f"  API Key: {DEEPSEEK_API_KEY[:10]}... (已配置)")
    print(f"  API URL: {DEEPSEEK_BASE_URL}")
    print(f"  模型: {DEFAULT_MODEL}")
else:
    print(" API Key 未配置")

执行结果说明：

DEFAULT_MODEL：默认使用 deepseek-chat 模型，你也可以根据需要更换为其他兼容模型
如果显示 API Key 未配置，后续的 LangExtract 提取和 Agent 问答将无法正常工作
步骤 3：定义阿里云百炼 Embeddings 配置

向量化（Embeddings）是将文本转换为数值向量的过程，这是向量检索的基础。我们使用阿里云百炼的 text-embedding-v4 模型进行向量化。

# 阿里云百炼 Embeddings 配置
DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY", "")
DASHSCOPE_BASE_URL = os.getenv("DASHSCOPE_BASE_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1")
EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "text-embedding-v4")

print("\n阿里云百炼 Embeddings 配置:")
if DASHSCOPE_API_KEY:
    print(f"  API Key: {DASHSCOPE_API_KEY[:10]}... (已配置)")
    print(f"  API URL: {DASHSCOPE_BASE_URL}")
    print(f"  Embedding 模型: {EMBEDDING_MODEL}")
else:
    print(" API Key 未配置")

执行结果说明：

text-embedding-v4：阿里云百炼的向量化模型，支持中文和英文，向量维度为 1024
如果显示 API Key 未配置，后续的向量存储和检索将无法正常工作

注意：阿里云百炼的 Embeddings API 有批量大小限制（单次最多 10 条），我们会在后续代码中设置 chunk_size=10。

步骤 4：定义 ChromaDB 和输出目录配置

最后，我们定义向量数据库的存储路径和输出文件的保存目录。

# ChromaDB 配置
CHROMA_PERSIST_DIR = "./chroma_db_chapter5"
COLLECTION_NAME = "pdf_knowledge_graph"

# 输出目录配置
OUTPUT_DIR = Path("./outputs")
OUTPUT_DIR.mkdir(exist_ok=True)

print("\n存储配置:")
print(f"  ChromaDB 存储路径: {CHROMA_PERSIST_DIR}")
print(f"  集合名称: {COLLECTION_NAME}")
print(f"  输出目录: {OUTPUT_DIR}")

执行结果说明：

CHROMA_PERSIST_DIR：ChromaDB 的持久化存储路径，数据会保存在磁盘上，重启程序后依然可用
COLLECTION_NAME：向量集合的名称，类似于数据库中的表名
OUTPUT_DIR：所有中间文件（Markdown、提取结果、知识图谱等）都会保存在这个目录

说明：如果输出目录不存在，mkdir(exist_ok=True) 会自动创建，不会报错。

6.3 定义数据结构

在本系统中，我们需要定义一个核心的数据结构：KnowledgeExtraction，用于存储每一个提取的实体或关系，并且包含完整的溯源信息。

步骤：定义 KnowledgeExtraction 数据类

这个数据类包含了一个知识实体的所有必要信息：来源文档、实体类型、提取文本、原文位置、属性信息等。

# 定义知识提取结果的数据结构
@dataclass
class KnowledgeExtraction:
    """知识提取结果（带溯源）"""
    doc_id: str                              # 文档 ID
    doc_title: str                           # 文档标题
    extraction_class: str                    # 提取类型（实体、关系、事件等）
    extraction_text: str                     # 提取的原文文本
    char_interval: Optional[dict] = None     # 原文字符区间（溯源的核心）
    attributes: dict = field(default_factory=dict)  # 属性信息
    
    def to_dict(self) -> dict:
        """转换为字典格式"""
        return asdict(self)
    
    def to_searchable_text(self) -> str:
        """生成用于向量化的可搜索文本"""
        parts = [
            f"类型: {self.extraction_class}",
            f"内容: {self.extraction_text}",
            f"来源: {self.doc_title}"
        ]
        if self.attributes:
            for k, v in self.attributes.items():
                parts.append(f"{k}: {v}")
        return " | ".join(parts)

print("KnowledgeExtraction 数据结构定义完成")
print("\n字段说明:")
print("  - doc_id: 文档唯一标识")
print("  - doc_title: 文档名称")
print("  - extraction_class: 实体类型（如'实体'、'关系描述'、'数据指标'）")
print("  - extraction_text: 从原文提取的精确文本")
print("  - char_interval: {'start_pos': 起始位置, 'end_pos': 结束位置} ← 溯源关键")
print("  - attributes: 实体的属性信息（如类型、角色等）")

执行结果说明：

char_interval 是溯源机制的核心字段，记录了提取文本在原 Markdown 中的精确位置
to_searchable_text() 方法将结构化数据转换为一段便于向量化和检索的文本
使用 @dataclass 装饰器可以自动生成 __init__、__repr__ 等方法，简化代码

重点：char_interval 的存在使得我们能够在 Agent 回答时，明确告知用户"这个信息来自原文的哪个位置"，实现真正的可追溯。

6.4 MinerU PDF 解析

MinerU 是一个专业的 PDF 解析服务，能够：

识别文档结构（标题、段落、表格、公式）
将 PDF 转换为结构化的 Markdown 格式
保留原文的语义关系

接下来，我们将逐步实现一个完整的 MinerU API 调用流程。

MinerU 在线API的工作流程如下：

FENCE0

大家可以在MinerU的官网看到详细的API接口文档：

下面我们来实现一个MinerU 客户端：

步骤 1：准备 PDF 文件路径

首先，我们需要指定要解析的 PDF 文件路径。请确保该文件存在。

# 指定要解析的 PDF 文件
PDF_PATH = "test.pdf"  # 修改为你的 PDF 文件路径

# 检查文件是否存在
pdf_file = Path(PDF_PATH)
if pdf_file.exists():
    file_size = pdf_file.stat().st_size
    print(f"PDF 文件存在: {PDF_PATH}")
    print(f"  文件大小: ({file_size/1024:.2f} KB)")
else:
    print(f"PDF 文件不存在: {PDF_PATH}")
    print("  请修改 PDF_PATH 变量为有效的文件路径")

执行结果说明：

如果显示 PDF 文件存在，说明文件路径正确，可以继续
如果显示 PDF 文件不存在，请检查路径是否正确，或者将 PDF 文件复制到当前目录
步骤 2：请求预签名上传 URL

这一步向 MinerU API 发送请求，告诉服务器我们要上传一个 PDF 文件，并获取一个临时的上传 URL 和批次 ID。

# 记录开始时间
parse_start_time = time.time()

# 准备 HTTP 请求头
headers = {
    "Authorization": f"Bearer {MINERU_API_KEY}",
    "Content-Type": "application/json",
}

# 步骤1：请求预签名上传 URL
print("步骤1：请求上传 URL...")
response = requests.post(
    f"{MINERU_BASE_URL}/api/v4/file-urls/batch",
    headers=headers,
    json={
        "files": [{"name": pdf_file.name, "data_id": pdf_file.name}],
        "model_version": "vlm",  # 使用视觉语言模型版本
    }
)
response.raise_for_status()  # 如果请求失败，抛出异常

# 解析响应
data = response.json()["data"]
batch_id = data["batch_id"]
upload_url = data["file_urls"][0]

print(f"  批次 ID: {batch_id}")
print(f"  上传 URL: {upload_url[:50]}...")

执行结果说明：

batch_id：这是一个唯一标识符，后续查询解析进度时需要用到
upload_url：一个临时的预签名 URL，用于上传 PDF 文件，通常有时效性（如 1 小时）
model_version: vlm：指定使用视觉语言模型进行解析，能更好地处理图表和公式

注意：如果这一步报错（如 401 Unauthorized），请检查 MINERU_API_KEY 是否正确。

步骤 3：上传 PDF 文件

现在我们将本地的 PDF 文件上传到刚才获取的预签名 URL。

# 步骤2：上传 PDF 文件
print("\n步骤2：上传 PDF...")

with open(PDF_PATH, "rb") as f:
    file_content = f.read()
    upload_response = requests.put(upload_url, data=file_content)
    upload_response.raise_for_status()

print(f"   上传速度: {len(file_content) / (time.time() - parse_start_time) / 1024:.2f} KB/s")

执行结果说明：

这里使用 HTTP PUT 请求将 PDF 文件的二进制内容上传到预签名 URL
上传速度取决于网络环境，通常在几秒到几十秒之间

说明：预签名 URL 是一种安全的文件上传方式，客户端直接上传到对象存储（如 S3），不需要经过 API 服务器中转，提高效率。

步骤 4：轮询等待解析完成

PDF 上传后，MinerU 服务器会开始解析。这个过程需要一定时间（通常几秒到几分钟，取决于 PDF 的复杂度和页数）。我们需要定期查询解析进度，直到完成。

# 步骤3：轮询等待解析
print("\n步骤3：等待解析...")

max_wait = 600  # 最长等待时间（秒）
poll_interval = 3  # 轮询间隔（秒）
elapsed = 0
full_zip_url = None

while elapsed < max_wait:
    # 查询解析状态
    status_response = requests.get(
        f"{MINERU_BASE_URL}/api/v4/extract-results/batch/{batch_id}",
        headers={"Authorization": f"Bearer {MINERU_API_KEY}"}
    )
    result = status_response.json()["data"]["extract_result"][0]
    state = result["state"]
    
    if state == "done":
        # 解析完成
        full_zip_url = result["full_zip_url"]
        print(f"   ✓ 解析完成！(耗时 {elapsed}秒)")
        break
    elif state == "failed":
        # 解析失败
        error_msg = result.get('err_msg', '未知错误')
        print(f"   解析失败: {error_msg}")
        break
    elif state == "running":
        # 正在解析
        progress = result.get("extract_progress", {})
        extracted = progress.get("extracted_pages", 0)
        total = progress.get("total_pages", "?")
        print(f"   {extracted}/{total} 页 (已等待 {elapsed}秒)", end="\r")
    
    time.sleep(poll_interval)
    elapsed += poll_interval

if elapsed >= max_wait:
    print(f"\n   解析超时 (超过 {max_wait}秒)")

执行结果说明：

state 字段有三种可能的值：
- running：正在解析中
- done：解析完成
- failed：解析失败
extract_progress 显示了当前的解析进度（已处理页数 / 总页数）
我们每 3 秒查询一次状态，直到完成或超时

注意：对于复杂的 PDF（如包含大量图表、公式的学术论文），解析时间可能较长。如果经常超时，可以增加 max_wait 的值。

# 步骤4：下载并提取 Markdown
if full_zip_url:
    print("\n步骤4：下载解析结果...")
    
    # 下载 ZIP 包
    zip_response = requests.get(full_zip_url)
    
    # 解压并提取 Markdown 文件
    with zipfile.ZipFile(io.BytesIO(zip_response.content)) as zf:
        # 查找 .md 文件
        md_files = [f for f in zf.namelist() if f.endswith('.md')]
        
        if md_files:
            # 优先选择文件名包含 'full' 的文件，否则选第一个
            target_file = next((f for f in md_files if 'full' in f.lower()), md_files[0])
            
            # 读取 Markdown 内容
            with zf.open(target_file) as f:
                markdown_text = f.read().decode('utf-8')
            
            print(f"   成功提取 Markdown: {target_file}")
            print(f"   Markdown 长度: {len(markdown_text):,} 字符")
        else:
            print("   ZIP 包中未找到 .md 文件")
            markdown_text = None
    
    # 统计解析总耗时
    parse_time = time.time() - parse_start_time
    print(f"\nPDF 解析完成 (总耗时 {parse_time:.2f}秒)")
else:
    print("\n未能获取解析结果")
    markdown_text = None

执行结果说明：

markdown_text 变量现在包含了从 PDF 提取出的完整 Markdown 文本
这个 Markdown 保留了文档的结构信息（标题、段落、列表等），是后续知识提取的输入
ZIP 包中可能包含多个 .md 文件，我们优先选择文件名包含 full 的完整版本

说明：MinerU 还会提取 PDF 中的图片，保存在 ZIP 包中。如果需要分析图片，可以进一步处理这些图片文件。

步骤 6：保存 Markdown 到文件

为了方便后续查看和调试，我们将提取出的 Markdown 保存到本地文件。

# 保存 Markdown 到文件
if markdown_text:
    md_file = OUTPUT_DIR / "parsed_document.md"
    with open(md_file, "w", encoding="utf-8") as f:
        f.write(markdown_text)
    
    print(f"Markdown 已保存: {md_file}")
    print(f"\nMarkdown 预览（前500字符）:")
    print("-" * 80)
    print(markdown_text[:500])
    print("-" * 80)
else:
    print("没有 Markdown 内容可保存")

6.5 LangExtract 知识提取

现在我们有了结构化的 Markdown 文本，接下来需要从中提取出结构化的知识。与前几章不同，本章我们要提取的不仅仅是简单的实体，还包括：

实体：人物、机构、地点、时间、概念、技术术语等
数据指标：数值、百分比、统计数据等
关系描述：实体之间的关系（合作、隶属、引用等）
事件：重要事件和行为

同时，最关键的是，每个提取的实体都会记录其在原 Markdown 文本中的精确位置（char_interval），实现完整的溯源。

定义提取 Prompt

Prompt 是告诉 LLM"我们要提取什么、怎么提取"的指令。一个好的 Prompt 应该：

明确列出要提取的类别
规定提取的要求和约束
简洁清晰，避免歧义

# 定义知识提取的 Prompt
extraction_prompt = textwrap.dedent("""\
    从文档中提取以下结构化知识:
    
    - 实体: 人物、机构、地点、时间、概念、技术术语
    - 数据指标: 数值、百分比、统计数据
    - 关系描述: 实体之间的关系（合作、隶属、引用等）
    - 事件: 重要事件和行为
    
    要求:
    1. extraction_text 必须是原文的精确子串
    2. 为每个提取添加丰富的属性信息
    3. 关系类型必须在 attributes 中标注涉及的主体
    4. 保持原文出现顺序
    """)

print("提取 Prompt 定义完成")
print("\nPrompt 内容:")
print(extraction_prompt)

Prompt 设计要点：

明确类别：我们明确列出了 4 大类要提取的知识
精确子串：要求 extraction_text 必须是原文的精确子串，这是实现溯源的前提
丰富属性：鼓励为每个实体添加属性，增加知识的语义深度
关系标注：对于关系类型，要求在 attributes 中标注涉及的主体（如"主体1"、"主体2"）

说明：这个 Prompt 是通用的，适合各类文档。在实际应用中，可以根据具体场景（如法律文书、学术论文）进行调整。

定义 Few-shot 示例

Few-shot 示例是给大模型的"范例"，通过展示输入和期望的输出，引导大模型按照我们期望的格式进行提取。

# 定义 Few-shot 示例
example_text = textwrap.dedent("""\
    # 人工智能发展报告
    
    ## 作者
    张三（清华大学）、李四（北京大学）
    
    ## 摘要
    本报告分析了2024年人工智能的发展趋势。研究显示，大模型参数量增长了300%，
    推理成本下降了50%。清华大学与北京大学联合开展了此项研究。
    """).strip()

# 期望的提取结果
example_extractions = [
    lx.data.Extraction(
        extraction_class="实体",
        extraction_text="张三",
        attributes={"类型": "人物", "机构": "清华大学"}
    ),
    lx.data.Extraction(
        extraction_class="实体",
        extraction_text="清华大学",
        attributes={"类型": "机构", "类别": "高校"}
    ),
    lx.data.Extraction(
        extraction_class="数据指标",
        extraction_text="增长了300%",
        attributes={"指标": "大模型参数量", "类型": "增长率"}
    ),
    lx.data.Extraction(
        extraction_class="关系描述",
        extraction_text="清华大学与北京大学联合开展了此项研究",
        attributes={
            "类型": "合作关系",
            "主体1": "清华大学",
            "主体2": "北京大学",
            "关系": "联合研究"
        }
    ),
]

# 组装为 ExampleData
examples = [
    lx.data.ExampleData(
        text=example_text,
        extractions=example_extractions
    )
]

print("Few-shot 示例定义完成")
print(f"\n示例文本长度: {len(example_text)} 字符")
print(f"示例提取数量: {len(example_extractions)} 个")

Few-shot 示例设计要点：

覆盖所有类别：示例中包含了"实体"、"数据指标"、"关系描述"三种类别，确保 LLM 理解每种类别的格式
属性丰富：每个提取都包含了详细的属性信息，引导 LLM 也这样做
关系结构化：对于"关系描述"，我们明确标注了"主体1"、"主体2"、"关系"等字段，便于后续构建知识图谱

说明：Few-shot 示例的质量直接影响提取效果。建议根据实际文档类型，精心设计 1-3 个高质量示例。 image.png

创建 LangExtract 模型并执行提取

现在我们创建 LangExtract 的语言模型适配器，连接到 DeepSeek API，然后执行知识提取。

# 创建 LangExtract 模型
langextract_model = OpenAILanguageModel(
    model_id=DEFAULT_MODEL,
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL,
)

print("LangExtract 模型创建成功")
print(f"  模型: {DEFAULT_MODEL}")
print(f"  API: {DEEPSEEK_BASE_URL}")

# 准备提取参数
doc_id = "doc_001"
doc_title = PDF_PATH

# 如果 Markdown 文本过长，截断到 8000 字符（避免超过 LLM 上下文限制）
max_length = 8000
if markdown_text and len(markdown_text) > max_length:
    print(f"\n Markdown 文本较长 ({len(markdown_text):,} 字符)，截断至 {max_length:,} 字符")
    extraction_text = markdown_text[:max_length]
else:
    extraction_text = markdown_text

print(f"\n准备提取:")
print(f"  文档 ID: {doc_id}")
print(f"  文档标题: {doc_title}")
print(f"  文本长度: {len(extraction_text):,} 字符" if extraction_text else "  文本长度: 0 字符")

# 执行 LangExtract 提取
print("\n正在执行 LangExtract 提取...\n")
extract_start_time = time.time()

if extraction_text:
    result = lx.extract(
        text_or_documents=extraction_text,
        prompt_description=extraction_prompt,
        examples=examples,
        model=langextract_model,
        fence_output=True,  # 要求 LLM 输出用代码块包裹，避免格式错误
        use_schema_constraints=False,  # 不使用严格的 schema 约束，提高灵活性
        prompt_validation_level=PromptValidationLevel.OFF,  # 关闭提示词验证
        show_progress=True,  # 显示提取进度
    )
    
    extract_time = time.time() - extract_start_time
    print(f"\n提取完成 (耗时 {extract_time:.2f}秒)")
else:
    print("没有可提取的文本")
    result = None

LangExtract 返回的原始结果需要转换为我们之前定义的 KnowledgeExtraction 数据结构，特别是要提取出 char_interval 字段，这是溯源的关键。

# 转换为 KnowledgeExtraction 对象列表
extractions = []

if result:
    for ext in result.extractions:
        # 提取 char_interval（溯源的核心）
        if ext.char_interval:
            char_interval = {
                "start_pos": ext.char_interval.start_pos,
                "end_pos": ext.char_interval.end_pos
            }
        else:
            char_interval = None
        
        # 创建 KnowledgeExtraction 对象
        ke = KnowledgeExtraction(
            doc_id=doc_id,
            doc_title=doc_title,
            extraction_class=ext.extraction_class,
            extraction_text=ext.extraction_text,
            char_interval=char_interval,
            attributes=ext.attributes or {}
        )
        extractions.append(ke)

    print(f"已转换为 KnowledgeExtraction 对象")
    print(f"  提取实体总数: {len(extractions)}")
else:
    print("没有提取结果")

# 统计实体类型分布
if extractions:
    type_counts = Counter(e.extraction_class for e in extractions)
    
    print("\n提取结果统计:")
    print("\n实体类型分布:")
    for entity_type, count in type_counts.most_common():
        percentage = count / len(extractions) * 100
        print(f"   - {entity_type}: {count} 个 ({percentage:.1f}%)")
    
    # 统计溯源情况
    aligned_count = sum(1 for e in extractions if e.char_interval)
    alignment_rate = aligned_count / len(extractions) * 100 if extractions else 0
    
    print(f"\n溯源统计:")
    print(f"   - 有原文位置: {aligned_count} 个")
    print(f"   - 无原文位置: {len(extractions) - aligned_count} 个")
    print(f"   - 对齐率: {alignment_rate:.1f}%")
    
    if alignment_rate < 80:
        print("   对齐率较低，可能影响溯源质量")
    else:
        print("   对齐率良好，溯源质量较高")
    
    # 统计属性情况
    with_attrs = sum(1 for e in extractions if e.attributes)
    attrs_rate = with_attrs / len(extractions) * 100 if extractions else 0
    
    print(f"\n属性统计:")
    print(f"   - 有属性: {with_attrs} 个")
    print(f"   - 无属性: {len(extractions) - with_attrs} 个")
    print(f"   - 属性率: {attrs_rate:.1f}%")
    
    if attrs_rate < 60:
        print("   属性较少，建议优化 Few-shot 示例")
    else:
        print("   属性丰富，有利于知识图谱构建")
else:
    print("没有提取结果可统计")

# 展示前5个提取示例（含溯源验证）
if extractions and extraction_text:
    print("\n提取示例（前5个，含溯源验证）:")
    print("=" * 80)
    
    for i, ext in enumerate(extractions[:5], 1):
        print(f"\n[{i}] {ext.extraction_class}: {ext.extraction_text[:50]}{'...' if len(ext.extraction_text) > 50 else ''}")
        
        if ext.attributes:
            print(f"    属性: {ext.attributes}")
        
        if ext.char_interval:
            interval = ext.char_interval
            print(f"    原文位置: 字符 {interval['start_pos']} - {interval['end_pos']}")
            
            # 验证：从 Markdown 中提取对应位置的文本
            original_text = extraction_text[interval['start_pos']:interval['end_pos']]
            if original_text == ext.extraction_text:
                print(f"    溯源验证通过")
            else:
                print(f"    溯源不匹配: 原文为 '{original_text[:30]}...'")
        else:
            print(f"    无原文位置信息")
    
    print("\n" + "=" * 80)
    
    # 保存提取结果
    extractions_file = OUTPUT_DIR / "chapter5_extractions.json"
    with open(extractions_file, "w", encoding="utf-8") as f:
        json.dump([e.to_dict() for e in extractions], f, ensure_ascii=False, indent=2)
    
    print(f"\n提取结果已保存: {extractions_file}")
    print(f"   文件大小: {extractions_file.stat().st_size / 1024:.2f} KB")
else:
    print("没有提取结果可展示")

6.6 ChromaDB 向量存储

提取出的结构化知识需要存储起来，以便后续进行语义检索。向量数据库能够：

将文本转换为高维向量表示
支持相似度搜索，找到语义相关的知识
在检索时保留完整的元数据（包括溯源信息）

接下来，我们将提取结果存入 ChromaDB，并确保所有溯源信息（char_interval）都保存在 metadata 中。

步骤 1：初始化 ChromaDB 和 Embeddings

首先初始化 ChromaDB 客户端和 Embeddings 模型。

%pip install langchain dashscope

# 初始化 Embeddings
embeddings = OpenAIEmbeddings(
    model=EMBEDDING_MODEL,
    api_key=DASHSCOPE_API_KEY,
    base_url=DASHSCOPE_BASE_URL,
    check_embedding_ctx_length=False,  # 禁用 token 长度检查
    chunk_size=10,  # 阿里云百炼限制批量大小不超过 10
)
print(f"Embeddings: {EMBEDDING_MODEL}")

# 初始化 ChromaDB
persist_dir = Path(CHROMA_PERSIST_DIR)
persist_dir.mkdir(parents=True, exist_ok=True)
client = chromadb.PersistentClient(path=str(persist_dir))
print(f"ChromaDB: {persist_dir}")

# 重建集合（删除旧数据）
try:
    client.delete_collection(name=COLLECTION_NAME)
    print(f"  已删除旧集合")
except:
    pass

# 创建 LangChain 的 Chroma 向量存储
vectorstore = Chroma(
    collection_name=COLLECTION_NAME,
    embedding_function=embeddings,
    client=client,
)
print(f"  已创建集合: {COLLECTION_NAME}\n")

步骤 2：将提取结果存入向量库（含溯源信息）

这是关键步骤：我们需要将每个提取结果转换为向量，并在 metadata 中保存完整的溯源信息（char_interval）。

# 准备向量化数据
if extractions:
    print(f"存储到 ChromaDB...")
    
    texts = [ext.to_searchable_text() for ext in extractions]
    metadatas = []
    ids = []
    
    for ext in extractions:
        ids.append(str(uuid.uuid4()))
        # 关键：在 metadata 中保存溯源信息
        metadatas.append({
            "doc_id": ext.doc_id,
            "doc_title": ext.doc_title,
            "extraction_class": ext.extraction_class,
            "extraction_text": ext.extraction_text,
            "char_interval": json.dumps(ext.char_interval) if ext.char_interval else "",  # 溯源关键
            "attributes": json.dumps(ext.attributes, ensure_ascii=False),
        })
    
    # 批量插入向量
    vectorstore.add_texts(texts=texts, metadatas=metadatas, ids=ids)
    print(f"✓ 已索引 {len(texts)} 条记录\n")
    
    # 查询集合信息
    collection = client.get_collection(COLLECTION_NAME)
    count = collection.count()
    print(f"集合统计:")
    print(f"   - 向量数量: {count}")
    print(f"   - 集合名称: {COLLECTION_NAME}")
else:
    print("没有提取结果可存储")

6.7 知识图谱构建

知识图谱由实体和关系组成。我们从提取结果中：

提取所有实体（人物、机构、概念等）
提取所有关系（合作、隶属、引用等）
保留每个实体的提及位置（溯源信息）

# 构建知识图谱
knowledge_graph = {
    "entities": {},
    "relations": []
}

if extractions:
    for ext in extractions:
        if ext.extraction_class == "关系描述":
            # 提取关系
            attrs = ext.attributes
            knowledge_graph["relations"].append({
                "text": ext.extraction_text,
                "type": attrs.get("类型", "未知"),
                "subject": attrs.get("主体1"),
                "object": attrs.get("主体2"),
                "relation": attrs.get("关系"),
                "source": ext.doc_title
            })
        elif ext.extraction_class in ["实体", "数据指标"]:
            # 提取实体（保留溯源信息）
            entity_name = ext.extraction_text
            if entity_name not in knowledge_graph["entities"]:
                knowledge_graph["entities"][entity_name] = {
                    "type": ext.extraction_class,
                    "attributes": ext.attributes,
                    "mentions": []  # 存储所有提及位置（溯源）
                }
            # 添加提及位置
            knowledge_graph["entities"][entity_name]["mentions"].append({
                "source": ext.doc_title,
                "position": ext.char_interval  # 溯源关键
            })
    
    print(f"知识图谱统计:")
    print(f"   - 实体数: {len(knowledge_graph['entities'])}")
    print(f"   - 关系数: {len(knowledge_graph['relations'])}")
    
    # 保存知识图谱
    kg_file = OUTPUT_DIR / "knowledge_graph.json"
    with open(kg_file, "w", encoding="utf-8") as f:
        json.dump(knowledge_graph, f, ensure_ascii=False, indent=2)
    print(f"\n知识图谱已保存: {kg_file}")
    
    # 展示部分知识图谱（含溯源信息）
    print(f"\n实体示例（前5个，含溯源信息）:")
    for i, (entity, data) in enumerate(list(knowledge_graph['entities'].items())[:5], 1):
        print(f"   [{i}] {entity}")
        print(f"       类型: {data['type']}")
        if data['attributes']:
            print(f"       属性: {data['attributes']}")
        if data['mentions']:
            print(f"       提及次数: {len(data['mentions'])}")
            for j, mention in enumerate(data['mentions'][:2], 1):
                if mention.get('position'):
                    pos = mention['position']
                    print(f"         提及{j}: 字符 {pos['start_pos']}-{pos['end_pos']} (来源: {mention['source']})")
else:
    print("没有提取结果可构建知识图谱")

6.8 Agent 智能问答与溯源

Agent 问答系统的工作流程：

用户提问：输入自然语言问题
向量检索：从 ChromaDB 中检索相关的知识
构建上下文：将检索结果（含溯源信息）组织成上下文
LLM 生成回答：基于上下文生成答案
展示溯源：在回答中明确展示每个信息的原文位置

这是整个系统的核心亮点：每个回答都能追溯到原文的具体位置。

步骤 1：定义图检索函数（含溯源信息）

这个函数从 ChromaDB 中检索相关知识，并返回完整的溯源信息。

# 定义向量检索函数（含溯源信息）
def search_knowledge(query: str, top_k: int = 5):
    """从向量库检索相关知识，返回含溯源信息的结果"""
    if not extractions:
        return []
    
    # 执行相似度搜索
    results = vectorstore.similarity_search_with_score(query, k=top_k)
    
    # 格式化结果（包含溯源信息）
    formatted_results = []
    for doc, score in results:
        similarity_score = 1 / (1 + score)  # 转换为相似度
        
        # 解析溯源信息
        char_interval_str = doc.metadata.get("char_interval", "")
        char_interval = json.loads(char_interval_str) if char_interval_str else None
        
        formatted_results.append({
            "score": similarity_score,
            "doc_title": doc.metadata.get("doc_title"),
            "extraction_class": doc.metadata.get("extraction_class"),
            "extraction_text": doc.metadata.get("extraction_text"),
            "char_interval": char_interval,  # 溯源关键
            "attributes": json.loads(doc.metadata.get("attributes", "{}")),
        })
    
    return formatted_results

print("向量检索函数定义完成")

# =============================================================================
# 图检索函数（GraphRAG 核心）
# =============================================================================

def graph_search(query_entity: str, hop: int = 1):
    """
    从知识图谱中检索实体及其关联关系
    
    Args:
        query_entity: 查询的实体名称（支持模糊匹配）
        hop: 跳数，1表示直接关联，2表示二度关联
    
    Returns:
        相关实体和关系
    """
    results = {
        "matched_entities": [],
        "related_relations": [],
        "connected_entities": set()
    }
    
    # 1. 模糊匹配实体
    for entity_name, entity_data in knowledge_graph["entities"].items():
        if query_entity.lower() in entity_name.lower():
            results["matched_entities"].append({
                "name": entity_name,
                **entity_data
            })
    
    # 2. 查找相关关系
    matched_names = [e["name"] for e in results["matched_entities"]]
    
    for relation in knowledge_graph["relations"]:
        subject = relation.get("subject", "")
        obj = relation.get("object", "")
        
        for name in matched_names:
            if name.lower() in str(subject).lower() or name.lower() in str(obj).lower():
                results["related_relations"].append(relation)
                if subject:
                    results["connected_entities"].add(subject)
                if obj:
                    results["connected_entities"].add(obj)
                break
    
    # 3. 二度关联
    if hop > 1 and results["connected_entities"]:
        for connected in list(results["connected_entities"]):
            for relation in knowledge_graph["relations"]:
                subject = relation.get("subject", "")
                obj = relation.get("object", "")
                if connected.lower() in str(subject).lower() or connected.lower() in str(obj).lower():
                    if relation not in results["related_relations"]:
                        results["related_relations"].append(relation)
    
    results["connected_entities"] = list(results["connected_entities"])
    return results

print("图检索函数定义完成")

# =============================================================================
# LangChain 1.1 GraphRAG Tools
# =============================================================================

from langchain.tools import tool
from langchain.agents import create_agent
from langchain.messages import HumanMessage

# Tool 1: 向量检索
@tool
def vector_search_tool(query: str) -> str:
    """
    向量语义检索：根据问题搜索相关知识片段。
    返回语义相似的文档内容和溯源信息。
    适合：查找与问题语义相关的内容。
    """
    results = search_knowledge(query, top_k=5)
    
    if not results:
        return "未找到相关信息"
    
    output_parts = []
    for i, r in enumerate(results, 1):
        part = f"[V{i}] {r['extraction_class']}: {r['extraction_text']}"
        if r.get('char_interval'):
            interval = r['char_interval']
            part += f"\n     位置: 字符 {interval['start_pos']}-{interval['end_pos']}"
        part += f"\n     来源: {r['doc_title']}"
        if r.get('attributes'):
            part += f"\n     属性: {r['attributes']}"
        output_parts.append(part)
    
    return "\n\n".join(output_parts)


# Tool 2: 图谱检索
@tool
def graph_search_tool(entity: str) -> str:
    """
    知识图谱检索：根据实体名称查找相关实体和关系。
    用于发现实体之间的结构化关联。
    适合：查找某个实体相关的关系、关联实体。
    """
    results = graph_search(entity, hop=1)
    
    # 格式化输出
    output_parts = []
    
    if results["matched_entities"]:
        output_parts.append("【匹配的实体】")
        for e in results["matched_entities"][:5]:
            mentions = e.get("mentions", [])
            output_parts.append(f"  - {e['name']} (类型: {e.get('type', '未知')}, 提及次数: {len(mentions)})")
    
    if results["related_relations"]:
        output_parts.append("\n【相关关系】")
        for i, rel in enumerate(results["related_relations"][:5], 1):
            subject = rel.get("subject", "?")
            relation = rel.get("relation", rel.get("type", "相关"))
            obj = rel.get("object", "?")
            output_parts.append(f"  [G{i}] {subject} --[{relation}]--> {obj}")
            if rel.get("text"):
                output_parts.append(f"       原文: {rel['text'][:50]}...")
    
    if results["connected_entities"]:
        output_parts.append(f"\n【关联实体】: {', '.join(results['connected_entities'][:10])}")
    
    if not output_parts:
        return f"未找到与 '{entity}' 相关的图谱信息"
    
    return "\n".join(output_parts)


# Tool 3: 混合检索
@tool  
def hybrid_search_tool(query: str) -> str:
    """
    混合检索（GraphRAG）：同时进行向量语义检索和知识图谱检索。
    适合：复杂问题，需要结合语义相似和结构化关系。
    """
    # 向量检索
    vector_result = vector_search_tool.invoke(query)
    
    # 从问题中提取可能的实体进行图检索
    words = [w for w in query.replace("？", "").replace("?", "").replace("，", " ").replace("。", " ").split() if len(w) >= 2]
    
    graph_results = []
    for word in words[:3]:
        gr = graph_search_tool.invoke(word)
        if "未找到" not in gr:
            graph_results.append(gr)
    
    # 组合结果
    output = "=== 向量检索结果 ===\n" + vector_result
    
    if graph_results:
        output += "\n\n=== 图谱检索结果 ===\n" + "\n".join(graph_results)
    
    return output

print("✓ GraphRAG Tools 定义完成")
print("   - vector_search_tool: 向量语义检索")
print("   - graph_search_tool: 图谱关系检索")
print("   - hybrid_search_tool: 混合检索")

步骤 2：定义 Agent 问答函数（含溯源上下文）

这个函数将检索结果组织成上下文，并在上下文中包含溯源信息，然后调用大模型生成回答。

# =============================================================================
# 创建 LangChain 1.1 GraphRAG Agent
# =============================================================================

from langchain_openai import ChatOpenAI

# 创建 DeepSeek 模型实例
llm = ChatOpenAI(
    model=DEFAULT_MODEL,
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL,
    temperature=0.3
)

# 创建 Agent
graphrag_agent = create_agent(
    model=llm,
    tools=[vector_search_tool, graph_search_tool, hybrid_search_tool],
    system_prompt="""你是一个 GraphRAG 知识图谱问答助手。

你有以下工具可用：
1. vector_search_tool - 向量语义检索，找语义相似的内容
2. graph_search_tool - 图谱检索，根据实体名找关系
3. hybrid_search_tool - 混合检索，同时使用向量和图谱

回答策略：
- 简单的内容查询：用 vector_search_tool
- 查找实体关系：用 graph_search_tool  
- 复杂问题：用 hybrid_search_tool

回答要求：
1. 综合检索到的信息回答问题
2. 标注信息来源（如 [V1] 表示向量结果，[G1] 表示图谱关系）
3. 如果有溯源位置信息，也一并说明
"""
)

print("✓ GraphRAG Agent 创建完成")

# =============================================================================
# 封装 agent_query 函数
# =============================================================================

def agent_query(question: str, top_k: int = 5):
    """
    GraphRAG Agent 问答
    
    Args:
        question: 用户问题
        top_k: 检索数量
    
    Returns:
        包含 question, answer, evidence 的字典
    """
    # 调用 Agent
    result = graphrag_agent.invoke({
        "messages": [HumanMessage(content=question)]
    })
    
    # 提取最终回答
    answer = result["messages"][-1].content
    
    # 提取工具调用记录作为 evidence
    tool_calls = []
    for msg in result["messages"]:
        # 工具调用请求
        if hasattr(msg, "tool_calls") and msg.tool_calls:
            for tc in msg.tool_calls:
                tool_calls.append({
                    "type": "call",
                    "tool": tc.get("name", "unknown"),
                    "args": tc.get("args", {})
                })
        # 工具返回结果
        if hasattr(msg, "name") and msg.name:
            tool_calls.append({
                "type": "result",
                "tool": msg.name,
                "content": msg.content[:500] + "..." if len(msg.content) > 500 else msg.content
            })
    
    return {
        "question": question,
        "answer": answer,
        "evidence": tool_calls
    }

print("✓ agent_query 函数定义完成")

步骤 3：测试 Agent 问答（展示完整溯源）

现在我们来测试 Agent 问答功能，重点观察溯源信息是如何在回答中展示的。

# =============================================================================
# 测试 GraphRAG 问答
# =============================================================================

test_questions = [
    "民间借贷的利率上限是多少？",
    "借款合同违约如何处理？",
    "民法典第六百七十五条",
]

for question in test_questions:
    print(f"\n{'='*80}")
    print(f"问题: {question}")
    print(f"{'-'*80}")
    
    result = agent_query(question, top_k=3)
    
    print(f"\n回答:\n{result['answer']}\n")
    
    # 展示工具调用证据
    if result['evidence']:
        print(f"Agent 工具调用记录:")
        for i, ev in enumerate(result['evidence'], 1):
            if ev['type'] == 'call':
                print(f"   [{i}] 调用 {ev['tool']}")
                print(f"       参数: {ev['args']}")
            else:
                print(f"   [{i}] {ev['tool']} 返回:")
                # 只显示前200字符
                content = ev['content']
                if len(content) > 200:
                    content = content[:200] + "..."
                print(f"       {content}")
    
    print(f"\n{'='*80}")

通过本章的学习，我们完成了一个完整的 Agentic GraphRAG 系统，实现了：

PDF 解析：使用 MinerU API 将 PDF 转换为结构化的 Markdown
知识提取：使用 LangExtract 提取实体、关系、属性，并保留原文位置信息
向量存储：使用 ChromaDB 存储提取结果的向量表示，并在 metadata 中保存溯源信息
知识图谱构建：从提取结果构建实体-关系图谱，保留每个实体的提及位置
Agent 问答与溯源：构建智能问答系统，每个回答都能追溯到原文的具体位置

核心就是溯源机制：LangExtract 自动记录每个实体的 char_interval（字符区间），在数据存储时，将 char_interval 保存在 ChromaDB 的 metadata 中，构建知识图谱是，每个实体的 mentions 列表包含所有提及位置，就能保证在Agent 问答阶段明确展示每个信息的原文位置（字符 X - Y）

这使得整个系统具有可追溯、可验证的特性，非常适合用于法律文书分析、学术论文检索、企业文档问答等场景。

本期课程内容就到这里，感谢大家的观看，我们下期再见！

课程说明：​

课程完整介绍​

部分课程成果演示​

《大模型Agent智能体开发实战》12月班封班 · 体验课

垂直领域 Agentic-GraphRAG 开发实战​

一、垂直领域 RAG 痛点与 GraphRAG 落地产品​

二、Agentic-GraphRAG 技术方案​

三、非结构化数据清洗：OCR 与 LangExtract 技术方案拆解​

3.1 OCR 数据解析​

3.2 LangExtract信息抽取​

四、【实战】Prompt Engineering 对比 LangExtract​

五、LangExtract 长文本提取实战​

5.1 配置运行环境​

5.2 读取长文本数据​

4.3 定义提取任务 Prompt​

5.4 定义 Few-shot 示例​

5.5 创建 LangExtract 模型​

5.6 运行长文本提取（核心）​

5.7 保存提取结果​

六、Agentic GraphRAG 构建与溯源实战​

6 .1 环境准备与依赖导入​

6.2 系统配置与参数定义​

6.3 定义数据结构​

6.4 MinerU PDF 解析​

6.5 LangExtract 知识提取​

6.6 ChromaDB 向量存储​

6.7 知识图谱构建​

6.8 Agent 智能问答与溯源​

课程说明：

课程完整介绍

部分课程成果演示

垂直领域 Agentic-GraphRAG 开发实战

一、垂直领域 RAG 痛点与 GraphRAG 落地产品

二、Agentic-GraphRAG 技术方案

三、非结构化数据清洗：OCR 与 LangExtract 技术方案拆解

3.1 OCR 数据解析

3.2 LangExtract信息抽取

四、【实战】Prompt Engineering 对比 LangExtract

五、LangExtract 长文本提取实战

5.1 配置运行环境

5.2 读取长文本数据

4.3 定义提取任务 Prompt

5.4 定义 Few-shot 示例

5.5 创建 LangExtract 模型

5.6 运行长文本提取（核心）

5.7 保存提取结果

六、Agentic GraphRAG 构建与溯源实战

6 .1 环境准备与依赖导入

6.2 系统配置与参数定义

6.3 定义数据结构

6.4 MinerU PDF 解析

6.5 LangExtract 知识提取

6.6 ChromaDB 向量存储

6.7 知识图谱构建

6.8 Agent 智能问答与溯源