Docling：逐步构建文档智能应用

传统的 PDF 提取工具（如 pypdf 或 PDFMiner）能够给出原始文本，但会丢失文档结构。表格会变成杂乱的文本，标题与正文混在一起，图像也不见了。对于 RAG 系统，这些凌乱的数据意味着糟糕的检索和不可靠的答案。Docling 是 IBM Research 推出的开源工具包，利用计算机视觉模型理解文档版面布局，保留表格、图像、标题与结构。它的处理速度比传统基于 OCR 的方法快最多 30 倍，并且可在本地机器上运行。

在本教程中，我们将使用 Docling 构建一个“文档智能助手”——一个基于 Streamlit 的 Web 应用，支持上传文档、可视化其结构，并通过 RAG 驱动的聊天机器人进行提问。您将学习如何用 Docling 处理多种格式的文档，提取并展示表格和图像，使用 ChromaDB 构建向量存储，并用 LangGraph 创建对话式智能体。完成后，您将拥有一个可用的应用，将复杂文档转化为结构化数据，并实现智能问答。

应用预览：

文档结构可视化：

先决条件

开始本教程之前，您需要具备：

技术技能：熟悉 Python 的类、装饰器、类型注解和上下文管理器。我们将贯穿使用异步操作和工厂模式。需要理解大语言模型如何与提示词、tokens 和向量嵌入协同工作。了解检索增强生成系统与向量数据库会更好，但不是必须——我们会在构建过程中解释核心概念。

开发环境：Python 3.10 或更高版本，并使用 pip 进行包管理。推荐使用 VS Code 等代码编辑器。您需要从 platform.openai.com 获取 OpenAI API 密钥——处理成本大约每份文档 $0.10–0.20。

时间投入：预计用时 60–90 分钟完成教程，包括阅读讲解、编写代码和测试应用。本文假设您具备中级 Python 能力。

了解 Docling：特性与能力

大多数文档处理工具将 PDF 视作图像文件或文本流。要么对每页运行 OCR，要么只提取纯文本却不了解其含义。Docling 的方法不同。它是 IBM Research 的开源工具包，使用计算机视觉模型像人一样理解文档结构。

当您将文档输入 Docling 时， 两个 AI 模型会对其进行分析：

• 版面分析：在 DocLayNet 上训练的模型通过分析页面布局识别标题、正文、表格和图像等不同元素
• 表格结构：TableFormer 负责处理表格并将其转换为结构化数据

这些模型理解文档并非一团文字，而是具有层级、关联与意义。

这种结构化理解对 RAG 系统至关重要。当您构建检索增强生成应用时，文档处理质量直接影响检索准确性。如果 PDF 提取把财务表格变成乱序文本，您的向量检索就会返回垃圾。Docling 保留结构，因此当您对文档分块并嵌入时，处理对象是干净、有组织的数据。

Docling 原生支持多种文档格式：

• PDF 文档
• Word 文档（DOCX）
• PowerPoint 演示文稿（PPTX）
• Excel 电子表格（XLSX）
• HTML 文件
• 图像

您还可以通过 EasyOCR、Tesseract 或 RapidOCR 等引擎为扫描件启用 OCR。该工具包可导出为多种格式，包括 Markdown（适合 LLM）、JSON（用于结构化数据管道）以及 DocTags（用于捕获数学公式、代码块等复杂元素的格式）。

除了格式灵活性，Docling 也具备性能优势。传统基于 OCR 的文档处理很慢，因为它把每页当作需要字符识别的图像。对于数字文档，Docling 跳过这一步，从而显著加速。它可以在普通硬件上本地运行，避免 API 成本，也不用将敏感文档发送给第三方服务。处理速度取决于文档复杂度、页数和硬件规格，在现代硬件上通常每页耗时从不到一秒到几秒不等。

IBM 持续改进 Docling 的能力。他们发布了 Granite-Docling，一款拥有 2.58 亿参数的视觉-语言模型，擅长处理复杂版面，并提供实验性的多语种支持（以英语为主，阿拉伯语、中文和日语处于早期支持阶段）。该工具包现已支持可配置分辨率的图像提取，我们将在应用中用它来展示 PDF 中的实际图像及其文本内容。

对于我们的场景，Docling 很合适，因为我们需要的是结构化数据，而不仅是原始文本。如果您只需要基础的文本提取，更简单的工具如 pypdf 可能就够了。但既然我们要构建一个用于分析与对话的 AI 文档处理应用，Docling 的结构感知处理是更好的选择。它在处理技术文档、研究论文或商业报告等对表格与版面依赖较强的文档时尤为有价值。

搭建开发环境

在开始构建文档处理器之前，您需要设置项目结构并安装所需依赖。本节涵盖初始设置：创建目录、安装依赖以及配置 API 密钥。

创建项目结构

首先为您的项目创建一个新目录：

mkdir docling-demo
cd docling-demo

在该目录内创建一个 src/ 文件夹用于存放 Python 模块：

mkdir src
touch src/__init__.py

这个 __init__.py 文件告知 Python src/ 是一个包，从而可以进行诸如 from src.document_processor import DocumentProcessor 这样的导入。

项目结构如下：

docling-demo/
├── src/
│   └── __init__.py

安装依赖

在项目根目录创建一个 requirements.txt 文件：

docling>=2.55.0
langchain-docling>=0.1.0
langchain>=0.3.0
langchain-openai>=0.2.0
langgraph>=0.2.0
langchain-chroma>=0.1.0
streamlit>=1.28.0
streamlit-extras>=0.7.0
python-dotenv>=1.0.0
chromadb>=0.4.22
tiktoken>=0.5.0
pandas>=2.0.0
numpy<2

之所以约束 numpy<2，是因为 Docling 的依赖（TensorFlow 和 Transformers）需要 NumPy 1.x。

安装这些包：

pip install -r requirements.txt

首次安装需要几分钟，因为 Docling 会下载预训练 AI 模型（约 500MB）。这些模型负责版面分析与表格结构识别。模型会缓存在本地，后续运行会更快。

配置环境变量

创建一个 .env 文件来存储您的 OpenAI API 密钥：

OPENAI_API_KEY=your-openai-api-key-here

从 platform.openai.com 获取您的 API 密钥。它将用于向量嵌入与聊天智能体。

创建一个 .env.example 模板：

OPENAI_API_KEY=your-openai-api-key-here

添加 .gitignore 以避免提交敏感数据：

# Environment variables
.env

# Python
__pycache__/
*.py[cod]
*.so
venv/
*.egg-info/

# Chroma
chroma_db/

创建主应用文件

在项目根目录创建 app.py。我们将随教程逐步完善该文件。现在先加入基础导入与配置：

import streamlit as st
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Page configuration
st.set_page_config(
    page_title="Document Intelligence Assistant",
    page_icon="📄",
    layout="wide"
)

st.title("Document Intelligence Assistant")
st.write("Application setup complete. We'll build the functionality next.")

测试您的环境：

streamlit run app.py

Streamlit 会在 http://localhost:8501 打开浏览器窗口，显示基础页面。

最终项目结构（在接下来的章节中我们会新增脚本）：

docling-demo/
├── .env
├── .env.example
├── .gitignore
├── requirements.txt
├── app.py
└── src/
   └── __init__.py

环境准备就绪后，我们可以构建利用 Docling 从上传文件中提取结构的文档处理器。

> 注意：下文将把应用脚本分块讲解。为便于把握全貌并跟随操作，建议在新标签页打开本项目的 GitHub 仓库。

构建核心文档处理器

📄 完整脚本： src/document_processor.py

文档处理器是 Docling 发挥作用的地方。该组件接收上传文件，将其转换为我们可用于 RAG 与可视化的结构化数据。我们需要两个输出：供向量存储使用的干净 markdown 文本，以及保留所有结构信息（表格、图像等）的原始 Docling 文档对象。

通过在 src/ 目录中新建 document_processor.py 来实现。我们将创建一个 DocumentProcessor 类，用于配置 Docling 的处理管线并处理文件上传。

配置管线选项

Docling 允许通过管线选项控制处理方式。对于 PDF，您可以为扫描件开启 OCR、启用表格结构识别，并提取图像：

from docling.document_converter import DocumentConverter, PdfFormatOption
from docling.datamodel.base_models import InputFormat
from docling.datamodel.pipeline_options import PdfPipelineOptions
from langchain_core.documents import Document

class DocumentProcessor:
    def __init__(self):
        # Configure pipeline options for PDF processing
        pipeline_options = PdfPipelineOptions()
        pipeline_options.do_ocr = True
        pipeline_options.do_table_structure = True
        pipeline_options.generate_picture_images = True
        pipeline_options.images_scale = 2.0

我们创建一个 PdfPipelineOptions 对象来配置 Docling 如何处理 PDF 文件。各选项控制具体功能：do_ocr 为扫描件启用光学字符识别；do_table_structure 激活表格检测与解析；generate_picture_images 指示 Docling 将内嵌图像提取为 PIL 对象；images_scale 设置提取图像的分辨率倍率。

# Initialize converter with PDF options
self.converter = DocumentConverter(
    format_options={InputFormat.PDF: PdfFormatOption(pipeline_options=pipeline_options)}
)

DocumentConverter 是 Docling 的主要处理引擎。我们将 PDF 管线选项封装在 PdfFormatOption 中进行初始化，使其专门应用于 PDF 输入文件。

下面解释各选项的作用。do_ocr 会启用 OCR。当处理带有文本层的数字 PDF 时，Docling 会自动跳过 OCR 以节省时间。对于扫描件或包含文字的图像，此设置会让 Docling 运行视觉模型来提取文本。

do_table_structure 选项用于启用表格结构识别。若不启用，表格会被当作纯文本提取而丢失格式。启用后，Docling 使用其 TableFormer 模型识别行、列、表头与单元格关系。这种结构化表示便于后续导出为 pandas DataFrame，保留表格形态。

将 generate_picture_images 设为 True 可启用图像提取。默认情况下，Docling 只记录图像位置，不会导出实际图像。启用后，您将获得可在 UI 中显示或供视觉模型处理的 PIL 图像对象。images_scale 控制提取分辨率——2.0 表示分辨率加倍，便于展示或进行进一步分析。

处理上传文件

管线配置完成后，我们添加通过 Streamlit 上传的文件处理方法。该方法处理 Streamlit 的文件对象，临时保存、运行 Docling 转换，并返回供 RAG 使用的 markdown 与用于可视化的 Docling 文档：

import os
import tempfile
from typing import List, Any

def process_uploaded_files(self, uploaded_files) -> tuple[List[Document], List[Any]]:
    documents = []
    docling_docs = []
    temp_dir = tempfile.mkdtemp()

    try:
        for uploaded_file in uploaded_files:
            # Save uploaded file to temporary location
            temp_file_path = os.path.join(temp_dir, uploaded_file.name)
            with open(temp_file_path, "wb") as f:
                f.write(uploaded_file.getbuffer())

Streamlit 将上传文件以内存文件对象形式提供，但 Docling 需要磁盘文件来处理。我们创建临时目录并将每个上传文件写入其中，保留原始文件名。

# Process the document with Docling
result = self.converter.convert(temp_file_path)
          
# Export to markdown
markdown_content = result.document.export_to_markdown()

converter.convert() 会运行 Docling 的完整文档分析管线。它识别文档版面、按需应用 OCR、检测表格与图像，并构建结构化表示。不启用图像提取时通常需 20–30 秒，启用图像时约 40–60 秒。

转换完成后，我们导出为 markdown 格式，得到干净且适合 LLM 的文本，并保留恰当的格式——标题仍是标题、列表保持结构、表格转换为 markdown 表格。

# Create LangChain document
doc = Document(
    page_content=markdown_content,
    metadata={
        "filename": uploaded_file.name,
        "file_type": uploaded_file.type,
        "source": uploaded_file.name,
    }
)

documents.append(doc)

# Store the Docling document for structure visualization
docling_docs.append({
    "filename": uploaded_file.name,
    "doc": result.document
})

我们为每个已处理文档创建两种表示。LangChain 的 Document 对象将 markdown 文本存放于 page_content 并附带元数据——它将用于 RAG 的向量存储。原始 Docling 文档对象则与文件名一同单独保存，用于后续的结构可视化（表格、图像、层级等）。

finally:
    import shutil
    shutil.rmtree(temp_dir)

return documents, docling_docs

finally 块确保不论处理成功与否，临时文件都会被清理。函数返回一个元组，包含两类表示：用于 RAG 的 LangChain 文档与用于结构可视化的 Docling 文档。

文档处理器已就绪。接下来我们构建 Streamlit 界面，让用户上传文件并查看处理状态。

实现文档结构可视化

📄 完整脚本： src/structure_visualizer.py

既然我们可以用 Docling 处理文档，就需要一种方式将提取结果可视化。原始 Docling 文档对象包含丰富的结构信息（标题、表格、图像和元数据），但其原生形式不够友好。我们将构建一个可视化层，将这些数据转化为包含四种视图的交互界面：摘要面板、分层大纲、交互式表格，以及提取的图像。

在 src/ 目录中新建 structure_visualizer.py。该组件将解析 Docling 的文档结构并整理用于展示。

构建可视化类

首先创建一个类，包装一个 Docling 文档并提供提取不同结构元素的方法：

from typing import List, Dict, Any
import pandas as pd
from docling_core.types.doc import DoclingDocument

class DocumentStructureVisualizer:
    def __init__(self, docling_document: DoclingDocument):
        self.doc = docling_document

初始化方法接收一个 DoclingDocument 对象（即 DocumentConverter.convert() 返回的对象）。该对象包含 Docling 提取的全部内容。texts 包含所有文本元素及其标签和位置；tables 保存带结构信息的表格数据；pictures 存储图像元数据与实际图像数据；pages 提供页级信息。

提取文档层级

文档不仅仅是段落。标题创建了帮助读者导航的层级结构。以下是如何提取该层级：

def get_document_hierarchy(self) -> List[Dict[str, Any]]:
    hierarchy = []

    if not hasattr(self.doc, "texts") or not self.doc.texts:
        return hierarchy

    for item in self.doc.texts:
        label = getattr(item, "label", None)

        if label and "header" in label.lower():
            text = getattr(item, "text", "")
            prov = getattr(item, "prov", [])
            page_no = prov[0].page_no if prov else None

            hierarchy.append({
                "type": label,
                "text": text,
                "page": page_no,
                "level": self._infer_heading_level(label)
            })

    return hierarchy

我们遍历文档中的所有文本项，筛选标签包含 header 的元素。每个文本项都有 Docling 在版面分析中分配的 label。常见标签包括 section_header、page_header、title 和普通 text。prov（来源信息）包含定位信息，包括元素所在页。我们提取标题文本、页码，并从标签类型推断其层级。

在展示大纲时，标题层级决定缩进。一个辅助方法将标签类型映射为数字层级：

def _infer_heading_level(self, label: str) -> int:
    if "title" in label.lower():
        return 1
    elif "section" in label.lower():
        return 2
    elif "subsection" in label.lower():
        return 3
    else:
        return 4

据此建立层级：文档标题为 1 级，章节标题为 2 级，小节为 3 级，其他标题默认 4 级。

将表格转换为 DataFrame

在提取了文档层级之后，我们来处理表格。不同于将表格变成杂乱字符串的简单文本提取，Docling 会保留其结构，这是它最有价值的特性之一：

def get_tables_info(self) -> List[Dict[str, Any]]:
    tables_info = []

    if not hasattr(self.doc, "tables") or not self.doc.tables:
        return tables_info

    for i, table in enumerate(self.doc.tables, 1):
        try:
            df = table.export_to_dataframe(doc=self.doc)

            prov = getattr(table, "prov", [])
            page_no = prov[0].page_no if prov else None

            caption_text = getattr(table, "caption_text", None)
            caption = caption_text if caption_text and not callable(caption_text) else None

            tables_info.append({
                "table_number": i,
                "page": page_no,
                "caption": caption,
                "dataframe": df,
                "shape": df.shape,
                "is_empty": df.empty
            })
        except Exception as e:
            print(f"Warning: Could not process table {i}: {e}")
            continue

    return tables_info

table.export_to_dataframe(doc=self.doc) 方法将 Docling 的表格表示转换为 pandas DataFrame。我们在可用时提取表格标题与页码。

用 PIL 提取图像

除了表格，Docling 还能从文档中提取实际图像数据。这是一个较新的特性，不仅记录图像位置——而是检索实际图像字节以供展示。（我们已在文档处理器的管线选项中启用该功能。）

def get_pictures_info(self) -> List[Dict[str, Any]]:
    pictures_info = []

    if not hasattr(self.doc, "pictures") or not self.doc.pictures:
        return pictures_info

    for i, pic in enumerate(self.doc.pictures, 1):
        prov = getattr(pic, "prov", [])

        if prov:
            page_no = prov[0].page_no
            bbox = prov[0].bbox

            caption_text = getattr(pic, "caption_text", None)
            caption = caption_text if caption_text and not callable(caption_text) else None

            pil_image = None
            try:
                if hasattr(pic, "image") and pic.image is not None:
                    if hasattr(pic.image, "pil_image"):
                        pil_image = pic.image.pil_image
            except Exception as e:
                print(f"Warning: Could not extract image {i}: {e}")

            pictures_info.append({
                "picture_number": i,
                "page": page_no,
                "caption": caption,
                "pil_image": pil_image,
                "bounding_box": {
                    "left": bbox.l,
                    "top": bbox.t,
                    "right": bbox.r,
                    "bottom": bbox.b
                } if bbox else None
            })

    return pictures_info

每张图片都有来源信息，包括页码与边界框坐标。边界框通过左、上、右、下坐标定义图像在页面中的位置。

启用图像提取时，图片对象会包含实际图像数据的 image 属性。我们通过 picture.image.pil_image 访问 PIL 图像对象，Streamlit 可直接用 st.image() 展示。try-except 块用于处理图像提取失败或未启用时的情况，优雅降级为仅显示元数据。

生成文档摘要

可视化器还需要一个方法：高层摘要，帮助用户快速了解文档结构：

def get_document_summary(self) -> Dict[str, Any]:
    pages = getattr(self.doc, "pages", {})
    texts = getattr(self.doc, "texts", [])
    tables = getattr(self.doc, "tables", [])
    pictures = getattr(self.doc, "pictures", [])

    text_types = {}
    for item in texts:
        label = getattr(item, "label", "unknown")
        text_types[label] = text_types.get(label, 0) + 1

    return {
        "name": self.doc.name,
        "num_pages": len(pages) if pages else 0,
        "num_texts": len(texts),
        "num_tables": len(tables),
        "num_pictures": len(pictures),
        "text_types": text_types
    }

我们统计文档中的页面、文本元素、表格与图像数量。text_types 字典按标签细分文本元素，显示 Docling 识别出的标题、页眉、段落等各类元素数量。这能让用户快速把握文档结构与复杂度。

将可视化器集成到 Streamlit

可视化器完成后，我们将其集成到 Streamlit 中，并提供四个选项卡：Summary、Hierarchy、Tables 和 Images。

def render_structure_viz():
    st.title("📊 Document Structure")

    if not st.session_state.docling_docs:
        st.info("👈 Please upload and process your documents first!")
        return

    doc_names = [doc["filename"] for doc in st.session_state.docling_docs]
    selected_doc_name = st.selectbox("Select document to analyze:", doc_names)

    selected_doc_data = next(
        (doc for doc in st.session_state.docling_docs if doc["filename"] == selected_doc_name),
        None
    )

    if not selected_doc_data:
        return

    visualizer = DocumentStructureVisualizer(selected_doc_data["doc"])

    tab1, tab2, tab3, tab4 = st.tabs(["📑 Summary", "🏗️ Hierarchy", "📊 Tables", "🖼️ Images"])

我们创建下拉菜单，允许用户选择要分析的文档（在上传多个文件时很有用）。从会话状态中取得所选 Docling 文档后，实例化可视化器，并创建四个视图选项卡。

with tab1:
    st.subheader("Document Summary")
    summary = visualizer.get_document_summary()

    col1, col2, col3, col4 = st.columns(4)
    with col1:
        st.metric("Pages", summary["num_pages"])
    with col2:
        st.metric("Tables", summary["num_tables"])
    with col3:
        st.metric("Images", summary["num_pictures"])
    with col4:
        st.metric("Text Items", summary["num_texts"])

    st.subheader("Content Types")
    text_types_df = pd.DataFrame([
        {"Type": k, "Count": v}
        for k, v in sorted(summary["text_types"].items(), key=lambda x: -x[1])
    ])
    st.dataframe(text_types_df, use_container_width=True)

with tab2:
    st.subheader("Document Hierarchy")
    hierarchy = visualizer.get_document_hierarchy()

    if hierarchy:
        for item in hierarchy:
            indent = "  " * (item["level"] - 1)
            st.markdown(f"{indent}**{item['text']}** _(Page {item['page']})_")
    else:
        st.info("No hierarchical structure detected")

with tab3:
    st.subheader("Tables")
    tables_info = visualizer.get_tables_info()

    if tables_info:
        for table_data in tables_info:
            st.markdown(f"### Table {table_data['table_number']} (Page {table_data['page']})")

            if table_data["caption"]:
                st.caption(table_data["caption"])

            if not table_data["is_empty"]:
                st.dataframe(table_data["dataframe"], use_container_width=True)
            else:
                st.info("Table is empty")

            st.divider()
    else:
        st.info("No tables found in this document")

with tab4:
    st.subheader("Images")
    pictures_info = visualizer.get_pictures_info()

    if pictures_info:
        for pic_data in pictures_info:
            st.markdown(f"**Image {pic_data['picture_number']}** (Page {pic_data['page']})")

            if pic_data["caption"]:
                st.caption(pic_data["caption"])

            if pic_data["pil_image"] is not None:
                st.image(pic_data["pil_image"], use_container_width=True)
            else:
                st.info("Image data not available")

            if pic_data["bounding_box"]:
                bbox = pic_data["bounding_box"]
                with st.expander("📐 Position Details"):
                    st.text(
                        f"Position: ({bbox['left']:.1f}, {bbox['top']:.1f}) - "
                        f"({bbox['right']:.1f}, {bbox['bottom']:.1f})"
                    )

            st.divider()
    else:
        st.info("No images found in this document")

结构可视化器已完成。用户可以上传文档并立即看到其“解剖图”——包含多少页、多少表格和图像，带有标题与章节的层级结构、可交互探索的表格，以及从文档中提取的实际图像。这种透明度有助于用户理解 Docling 的提取结果并建立对系统的信任。

文档处理与可视化就绪后，我们可以构建支持对这些处理后文档进行问答的 RAG 系统。

构建 RAG 向量存储与检索工具

📄 完整脚本： src/vectorstore.py | src/tools.py

完成文档处理和可视化后，我们可以构建问答功能。RAG（检索增强生成）使用户能够就其文档提问。系统将文档转换为向量嵌入，存储在向量数据库中，并检索相关片段来回答问题。

本节涵盖两个组件：负责文档分块与嵌入的向量存储管理器，以及用于检索相关信息的搜索工具。

设置向量存储

向量存储用于保存文档嵌入。当用户提问时，我们在该存储中搜索相关片段，并将其作为上下文提供给 LLM。

在 src/ 目录中新建 vectorstore.py：

from typing import List
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

这些导入包含 LangChain 的文档处理、文本分割、嵌入生成组件，以及 ChromaDB 向量存储。它们的协作方式如下：

class VectorStoreManager:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=100,
            length_function=len,
        )

我们初始化两个关键组件。OpenAI 的 text-embedding-3-small 模型用于将文本转为向量。它比 text-embedding-3-large 更小更快，适合嵌入数百个片段时使用。RecursiveCharacterTextSplitter 将文档拆分为 1000 字符片段并有 100 字符重叠，确保片段边界的重要信息不会被截断。

为何选择 1000 字符？

1000 字符在精度与上下文之间取得平衡——更小的片段检索更精确，但丢失上下文；更大的片段保留上下文，但相关性被稀释。

添加分块方法：

def chunk_documents(self, documents: List[Document]) -> List[Document]:
    """Split documents into smaller chunks for better retrieval."""
    chunks = self.text_splitter.split_documents(documents)
    return chunks

分割器会自动处理元数据，保留原文档的信息。每个片段都知道自己来自哪个文件，这在智能体回答中引用来源时非常重要。

接着添加创建向量存储的方法：

def create_vectorstore(self, chunks: List[Document]) -> Chroma:
    """Create a Chroma vector store from document chunks."""
    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=self.embeddings,
        collection_name="documents"
    )
    return vectorstore

ChromaDB 负责核心工作。它使用我们的嵌入模型对每个片段进行向量化并存储在内存中。执行相似度搜索时，ChromaDB 会计算查询向量与所有文档向量的余弦相似度，返回最接近的匹配结果。

创建搜索工具

向量存储已就绪，下一步需要为智能体提供查询方式。工具是 LLM 在需要外部信息时可以调用的函数。

在 src/ 目录中新建 tools.py：

from typing import Annotated
from langchain_core.tools import tool

def create_search_tool(vectorstore):
    """Create a search tool that has access to the vector store."""

该工厂函数使用闭包模式：接收一个向量存储并返回可搜索它的工具。工具无需依赖全局变量即可保持对向量存储的访问。

@tool
def search_documents(query: Annotated[str, "The search query or question about the documents"]) -> str:
    """Search the uploaded documents for relevant information."""

@tool 装饰器将此函数变为 LangChain 工具，供智能体调用。Annotated 类型注解描述参数，帮助 LLM 理解调用时需要传入的内容。

try:
   results = vectorstore.similarity_search(query, k=8)

   if not results:
      return "No relevant information found in the documents for this query."

我们从向量存储中检索 8 个相似片段（k=8）。这能为 LLM 提供足够上下文而不至于用冗余信息淹没它。最佳数量取决于文档类型——信息密度高的技术文档可能更适合较少片段（k=4–6），叙事类文档可能受益于更多片段（k=10–12）。

context_parts = []
for i, doc in enumerate(results, 1):
    source = doc.metadata.get("filename", doc.metadata.get("source", "Unknown source"))
    content = doc.page_content.strip()

    context_parts.append(
        f"[Source {i}: {source}]\n"
        f"Content: {content}\n"
    )

return "\n---\n".join(context_parts)

我们为每个片段添加来源文件名，并用分隔符拼接。LLM 接收这种结构化上下文，用其生成带来源引用的答案。

except Exception as e:
    return f"Error searching documents: {str(e)}"

return search_documents

错误处理确保检索失败时，智能体能收到清晰的信息而非崩溃。函数返回已配置好的工具，供智能体使用。

向量存储与搜索工具已就绪。接下来我们构建 LangGraph 智能体，编排检索与生成以回答用户问题。

创建支持流式输出的 LangGraph 智能体

📄 完整脚本： src/agent.py

搜索工具可以查询向量存储，但还需要一个智能协调者。LangGraph 智能体决定何时搜索、如何解读结果，并生成自然语言答案。我们还将实现流式输出，以实时展示进度。

构建 LangGraph 智能体

智能体接收问题，决定何时检索文档，并基于检索到的上下文生成答案。

在 src/ 目录中新建 agent.py：

from typing import List
from langchain_core.tools import BaseTool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver

我们导入工具处理、OpenAI 聊天模型、LangGraph 的 ReAct 智能体实现，以及会话记忆组件。

SYSTEM_PROMPT = """You are a helpful document intelligence assistant. You have access to documents that have been uploaded and processed.

GUIDELINES:
- Use the search_documents tool to find relevant information
- Keep it simple: one well-crafted search is usually sufficient
- Only search again if the first results are clearly incomplete
- Provide clear, accurate answers based on the document contents
- Always cite your sources with filenames
- If information isn't found, say so clearly
- Be concise but thorough

When answering:
1. Search the documents with a focused query
2. Synthesize a clear answer from the results
3. Include source citations (filenames)
4. Only search again if absolutely necessary
"""

def create_documentation_agent(tools: List[BaseTool], model_name: str = "gpt-4o-mini"):
    """Create a document intelligence assistant agent using LangGraph."""
    llm = ChatOpenAI(model=model_name, temperature=0)
    memory = MemorySaver()

我们使用 gpt-4o-mini 而非 gpt-5，因为它更快、更经济，同时足以胜任文档问答。温度设为 0 以获得一致、基于事实的回答。MemorySaver 为智能体提供会话记忆，让其在会话中记住先前的交流。

agent = create_react_agent(
    llm,
    tools=tools,
    prompt=SYSTEM_PROMPT,
    checkpointer=memory
)

return agent

LangGraph 的 create_react_agent 实现了 ReAct 模式（推理 + 行动）。智能体先推理所需步骤，再调用工具、观察结果，并循环直到得出答案。该模式非常适合 RAG，因为智能体可以自行决定何时检索以及如何使用检索到的上下文。

实现流式响应

智能体已经可以检索并生成答案，但用户不应在其工作时盯着空白界面等待 10 秒。流式输出可实时展示进度——首先显示“思考中”，接着“正在搜索”，然后答案逐字呈现。

在 app.py 中更新 render_chat() 函数，补全我们先前标记为 TODO 的响应处理：

if prompt:
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)

我们将用户消息加入会话历史，并在聊天界面展示。

with st.chat_message("assistant"):
    status_placeholder = st.empty()
    message_placeholder = st.empty()

    try:
        config = {"configurable": {"thread_id": "document_chat"}}

我们为状态提示和响应消息创建占位符。config 包含线程 ID，LangGraph 用它在多轮对话中维护会话记忆。

def generate_response():
    """Generator that yields tokens from LangGraph stream."""
    status_placeholder.markdown("🤔 **Thinking...**")
    first_content_token = True
    tool_call_detected = False
    final_answer_started = False

该生成器函数处理 LangGraph 的流式输出。我们用标志位跟踪状态，以便在智能体工作流不同阶段展示合适的状态信息。

for msg, metadata in st.session_state.agent.stream(
    {"messages": [HumanMessage(content=prompt)]},
    config=config,
    stream_mode="messages",
):
    langgraph_node = metadata.get("langgraph_node", "")

stream_mode="messages" 让我们获得真实的 LLM tokens，而不仅仅是最终输出。LangGraph 会在智能体执行过程中发出事件——如开始思考、调用工具、生成文本等。

if "tools" in langgraph_node.lower() or "tool" in langgraph_node.lower():
    if not tool_call_detected:
        status_placeholder.markdown("🔍 **Searching documents...**")
        tool_call_detected = True
    continue

if "agent" in langgraph_node.lower() and hasattr(msg, "content"):
    content = msg.content

if content:
    if first_content_token:
        status_placeholder.markdown("💬 **Generating answer...**")
        first_content_token = False
        final_answer_started = True

if final_answer_started:
    yield content

当 agent 节点开始生成最终答案时，我们更新状态并开始逐 token 输出内容。每个 token 会即时展示，形成顺滑的“打字机”效果。

status_placeholder.empty()

with message_placeholder.container():
    full_response = st.write_stream(generate_response())

流式输出完成后，我们清空状态提示。Streamlit 的 st.write_stream() 会自动逐 token 展示并累积更新 UI，带来流畅、可信的聊天体验。

st.session_state.messages.append({"role": "assistant", "content": full_response})

except Exception as e:
    st.error(f"Error generating response: {str(e)}")

我们将完整回答保存到消息历史，并处理流式生成过程中的任何错误。

至此，RAG 系统完成。用户可以上传文档，用 Docling 处理，探索其结构，并就内容进行自然对话。智能体能智能检索、引用来源，并以流式方式输出答案，体验顺畅。

测试与验证

您的文档智能助手已完成。在部署或分享之前，您应测试应用以验证各功能是否正常。

运行应用

在项目根目录启动应用：

streamlit run app.py

Streamlit 会在 http://localhost:8501 打开浏览器。您应能在侧边栏看到上传控件，主区域有两个选项卡。

测试文档处理管线

上传包含表格与图像的示例 PDF，以测试完整处理能力：

1. 点击侧边栏中的“Upload Documents”
2. 选择一个 PDF 文件（最好包含表格、图像和清晰的标题）
3. 点击“Process & Index”

观察处理指示：

• “Processing documents with Docling…”（20–60 秒，取决于文档大小）
• “Chunking documents…”
• “Creating vector store…”
• “Creating agent…”

完成后，您将在状态区域看到“✅ Ready to chat!”。

测试文档结构可视化

切换到 “Document Structure” 选项卡，验证 Docling 是否正确提取：

1. Summary 选项卡：检查计数是否与文档相符（页数、表格数、图像数、文本元素数）
2. Hierarchy 选项卡：验证标题顺序与缩进是否正确
3. Tables 选项卡：确认表格以交互式 DataFrame 显示而非杂乱文本
4. Images 选项卡：验证图像是否正常渲染（若启用了图像提取）

如果表格显示为空或图像未出现，请检查是否在管线选项中启用了 do_table_structure=True 和 generate_picture_images=True。

测试问答系统

回到 “Chat” 选项卡，用以下示例问题测试智能体：

良好开场问题：

• “这份文档讲的是什么？”
• “概括一下主要内容”
• “文档里包含哪些表格？”
• “列出所有图或图像及其标题”

文档特定问题（请根据您的文档调整）：

• “表 1 显示了哪些数据？”
• “解释一下第 3 节描述的方法论”
• “关键发现是什么？”
• “作者是谁？”

后续追问（测试会话记忆）：

• 在询问表格之后：“这些数据说明了什么？”
• 在得到答案之后：“能用更简单的方式解释一下吗？”

关注点

留意答案是否直接来自您的文档并带有来源引用、流式 tokens 是否平滑、状态指示（“Thinking…”，“Searching documents…”，“Generating answer…”）是否正常。注意是否出现不基于文档的泛泛回答、缺失来源引用、无状态指示的慢响应，或缺少 API 密钥的报错。

常见问题与修复

“No module named ‘docling’” 错误

pip install docling langchain langchain-openai langchain-chroma langgraph streamlit streamlit-extras pandas python-dotenv

“OpenAI API key not found” 错误

• 确认 .env 文件存在且包含 OPENAI_API_KEY=your-key-here

• 重启 Streamlit

处理超过 60 秒

• 首次运行（Docling 下载约 500MB 的模型）或文档较大时属正常

智能体给出泛泛答案

• 确认文档已处理并创建了向量存储
• 尝试更具体的问题

表格显示为空 DataFrame

• 确认在 PdfPipelineOptions 中设置了 do_table_structure=True

• 尝试一份包含原生表格的不同 PDF

应用现已通过验证，可投入使用。您可以用自己的文档进行测试，或将其部署供他人使用。

结语

您已经拥有一个可用的文档智能助手，能够在保留结构的前提下处理 PDF、Word、PowerPoint 与 HTML 文件。Docling 会提取文本、表格、图像与文档层级，用户可通过交互式选项卡进行探索。RAG 系统结合 ChromaDB 与 LangGraph，实现带来源引用的实时流式问答。这展示了相较于基础文本提取，结构感知的文档处理如何提升检索质量。完整源代码可在 GitHub 仓库获取。

在此基础上，您可以拓展多个方向：

• 加入批处理与持久化向量存储
• 启用不同引擎的OCR 以处理扫描件
• 部署到 Streamlit Cloud 或用 Docker 容器化
• 集成视觉-语言模型以分析提取的图表
• 构建领域特定的抽取规则或添加导出功能

若想进一步探索 RAG 系统与 LLM 应用，我们的 AI Engineering 路线将覆盖您在此用到的概念与模式。