任何格式RAG数据实现秒级转换！彻底解决RAG系统中最令人头疼的数据准备环节

PDF解析: PyPDF2 / pdfplumber / PyMuPDFOCR识别: Tesseract / PaddleOCR / EasyOCR  Word解析: python-docx / docx2txt音频转录: Whisper / Azure Speech表格识别: Camelot / Tabula分块策略: LangChain TextSplitter / LlamaIndex

DoclingDocument:  ├── metadata: 文档元数据  │   ├── title: 标题  │   ├── authors: 作者  │   ├── creation_date: 创建时间  │   └── source: 来源路径  │  ├── body: 文档主体内容  │   ├── texts: 文本块列表  │   │   ├── text_block_1  │   │   │   ├── content: 文本内容  │   │   │   ├── type: 类型（paragraph/heading/list）  │   │   │   └── level: 层级（如标题级别）  │   │   └── text_block_2  │   │  │   ├── tables: 表格列表  │   │   ├── table_1  │   │   │   ├── rows: 行数据  │   │   │   ├── headers: 表头  │   │   │   └── caption: 标题  │   │   └── table_2  │   │  │   └── figures: 图表列表  │       ├── figure_1  │       │   ├── image_data: 图像数据  │       │   ├── caption: 说明  │       │   └── classification: 分类（图表/图片）  │       └── figure_2  │  └── export_options: 导出配置      ├── format: 输出格式      └── options: 格式参数

PDF输入 → 布局分析模型 → 识别页面元素           ↓        文本区域 → 直接提取        表格区域 → 表格识别模型 → 结构化表格        图像区域 → 图像分类 → 判断类型        公式区域 → 公式识别 → LaTeX格式           ↓        DoclingDocument → 导出Markdown

输入：DoclingDocument + max_tokens + embedding_model
Step 1: 初步切分document → sentences[] 或 paragraphs[]
Step 2: 语义相似度计算for each adjacent pair (s_i, s_i+1):    similarity = cosine_similarity(        embed(s_i),         embed(s_i+1)    )
Step 3: 动态合并current_chunk = []for s in sentences:    if len(current_chunk) + len(s) <= max_tokens:        if similarity(current_chunk, s) > threshold:            current_chunk.append(s)  # 语义相关，合并        else:            output_chunks.append(current_chunk)  # 语义不相关，新块            current_chunk = [s]    else:        output_chunks.append(current_chunk)  # 超过最大长度，新块        current_chunk = [s]
输出：chunks[] (每个块都是语义连贯的文本)

from docling_core.transforms.chunker import HybridChunker
chunker = HybridChunker(    tokenizer="sentence-transformers/all-MiniLM-L6-v2",  # 嵌入模型    max_tokens=512,            # 最大Token数    merge_peers=True,          # 合并相似段落    include_metadata=True      # 包含元数据（标题、页码）)

安装前准备
在开始安装之前，请确保您的系统满足以下要求：- 操作系统：Linux (Ubuntu 20.04+) 或 macOS 12+- Python版本：3.8或更高- 硬盘空间：至少10GB可用空间- 内存：推荐16GB或以上
依赖安装
首先，您需要安装必要的系统依赖。对于Ubuntu系统，执行以下命令：sudo apt-get updatesudo apt-get install python3-dev build-essential
对于macOS系统，请使用Homebrew：brew install python@3.10
接下来，创建Python虚拟环境...

ounter(lineounter(lineounter(lineChunk 1: "安装前准备\n在开始安装之前，请确保您的系统满足以下要求：\n- 操作系统：Linux (Ubuntu 20.04+) 或 macOS 12+\n- Python版本：3.8或更高\n- 硬盘空间：至少10GB可用空间\n- 内存：推荐16GB或以上\n\n依赖安装\n首先，您需要安装必要的系统依"
Chunk 2: "赖。对于Ubuntu系统，执行以下命令：\nsudo apt-get update\nsudo apt-get install python3-dev build-essential\n\n对于macOS系统，请使用Homebrew：\nbrew install python@3.10\n\n接下来，创建Python虚拟环境..."

Chunk 1: "安装前准备\n在开始安装之前，请确保您的系统满足以下要求：\n- 操作系统：Linux (Ubuntu 20.04+) 或 macOS 12+\n- Python版本：3.8或更高\n- 硬盘空间：至少10GB可用空间\n- 内存：推荐16GB或以上"
Chunk 2: "依赖安装\n首先，您需要安装必要的系统依赖。对于Ubuntu系统，执行以下命令：\nsudo apt-get update\nsudo apt-get install python3-dev build-essential\n\n对于macOS系统，请使用Homebrew：\nbrew install python@3.10\n\n接下来，创建Python虚拟环境..."

# 检查Python版本（需要3.8+）python --version
# 如果版本过低，推荐使用pyenv管理多版本Python# macOS/Linux:curl https://pyenv.run | bashpyenv install 3.10.0pyenv global 3.10.0
# Windows:# 从 https://www.python.org/ 下载安装Python 3.10+

# 创建虚拟环境（推荐）python -m venv docling_envsource docling_env/bin/activate  # Linux/macOS# docling_env\Scripts\activate  # Windows
# 安装Doclingpip install docling
# 验证安装python -c "import docling; print(docling.__version__)"

# 安装FFmpeg（音频解码）# macOS:brew install ffmpeg
# Ubuntu/Debian:sudo apt-get updatesudo apt-get install ffmpeg
# Windows:# 从 https://ffmpeg.org/download.html 下载并配置PATH
# 安装Whisper模型（语音转文字）pip install openai-whisper

# CUDA（NVIDIA GPU）pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118
# Apple Silicon（M1/M2/M3 Mac）# Docling会自动使用MLX加速，无需额外配置

from docling.document_converter import DocumentConverter
# 创建转换器converter = DocumentConverter()
# 转换文档（支持本地路径或URL）source = "https://arxiv.org/pdf/2408.09869"  # Docling技术报告result = converter.convert(source)
# 导出为Markdownmarkdown_content = result.document.export_to_markdown()print(markdown_content)

ounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(line## Docling Technical Report
### Abstract
We present Docling, a comprehensive toolkit for document understanding...
### 1. Introduction
Document processing is a critical component of many AI applications...
| Model | Accuracy | Speed ||-------|----------|-------|| Heron | 94.2% | 2.3s/page || ...   | ...   | ...   |

from docling.document_converter import DocumentConverterimport os
# 文档列表（PDF、Word、Markdown）documents = [    "data/technical_manual.pdf",    "data/meeting_notes.docx",    "data/api_docs.md",    "data/quarterly_report.xlsx"]
# 创建转换器converter = DocumentConverter()
# 批量处理results = {}for doc_path in documents:    print(f"Processing: {doc_path}")
    try:        # 转换文档        result = converter.convert(doc_path)
        # 导出为Markdown        markdown = result.document.export_to_markdown()
        # 保存结果        output_path = f"output/{os.path.basename(doc_path)}.md"        os.makedirs("output", exist_ok=True)        with open(output_path, "w", encoding="utf-8") as f:            f.write(markdown)
        results[doc_path] = "Success"        print(f"  ✓ Saved to {output_path}")
    except Exception as e:        results[doc_path] = f"Error: {str(e)}"        print(f"  ✗ Failed: {str(e)}")
# 汇总报告print("\n=== Conversion Summary ===")for doc, status in results.items():    print(f"{doc}: {status}")

from docling.document_converter import DocumentConverter, PdfFormatOptionfrom docling.datamodel.base_models import InputFormatfrom docling.pipeline.standard_pdf_pipeline import StandardPdfPipelinefrom docling.pipeline.asr_pipeline import AsrPipeline
# 配置ASR流水线asr_options = {    "model": "openai/whisper-turbo",  # 使用Whisper Turbo模型    "device": "auto",                 # 自动选择设备（CPU/GPU/MLX）    "add_timestamps": True            # 添加时间戳}
# 创建转换器converter = DocumentConverter(    format_options={        InputFormat.AUDIO: AsrPipeline(**asr_options)    })
# 转换音频文件audio_path = "data/meeting_recording.mp3"result = converter.convert(audio_path)
# 导出为Markdown（包含时间戳）transcript = result.document.export_to_markdown()
# 保存转录结果with open("output/meeting_transcript.md", "w", encoding="utf-8") as f:    f.write(transcript)
print("Transcript saved!")print(f"Duration: {result.document.metadata.get('duration', 'N/A')} seconds")print(f"Word count: {len(transcript.split())}")

## Meeting Transcript
[00:00:00] Welcome everyone to today's product review meeting.
[00:00:05] Let's start with the Q4 roadmap discussion.
[00:00:12] Sarah, could you walk us through the key milestones?
[00:00:18] Sure, we have three major releases planned...

from docling.document_converter import DocumentConverterfrom docling_core.transforms.chunker import HybridChunkerfrom sentence_transformers import SentenceTransformerimport numpy as np
# Step 1: 转换文档为DoclingDocumentconverter = DocumentConverter()result = converter.convert("data/product_manual.pdf")doc = result.document
# Step 2: 配置混合分块器chunker = HybridChunker(    tokenizer="sentence-transformers/all-MiniLM-L6-v2",    max_tokens=256,        # 每个块最多256 tokens    merge_peers=True,      # 合并语义相似的相邻块    headings=True          # 包含标题作为上下文)
# Step 3: 执行分块chunks = list(chunker.chunk(doc))
print(f"Total chunks: {len(chunks)}")
# Step 4: 分析分块结果token_sizes = [len(chunk.text.split()) for chunk in chunks]print(f"Token distribution:")print(f"  Min: {min(token_sizes)}")print(f"  Max: {max(token_sizes)}")print(f"  Average: {np.mean(token_sizes):.1f}")
# Step 5: 导出分块结果for i, chunk in enumerate(chunks):    print(f"\n=== Chunk {i+1} ===")    print(f"Heading: {chunk.meta.headings}")    print(f"Text: {chunk.text[:200]}...")  # 显示前200字符    print(f"Tokens: {len(chunk.text.split())}")

ounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineTotal chunks: 23
Token distribution:  Min: 87  Max: 256  Average: 184.3
=== Chunk 1 ===Heading: ['Product Overview', 'Introduction']Text: This manual provides comprehensive information about the XYZ-2000 industrial controller. The device features advanced automation capabilities...Tokens: 128
=== Chunk 2 ===Heading: ['Installation', 'Hardware Requirements']Text: Before installing the XYZ-2000, ensure your system meets the following requirements: Power supply: 24V DC, 2A minimum...Tokens: 156

import psycopg2from docling.document_converter import DocumentConverterfrom docling_core.transforms.chunker import HybridChunkerfrom sentence_transformers import SentenceTransformerfrom openai import OpenAI
# ===== 配置 =====DB_CONFIG = {    "host": "localhost",    "database": "rag_db",    "user": "postgres",    "password": "your_password"}
embedding_model = SentenceTransformer("all-MiniLM-L6-v2")openai_client = OpenAI(api_key="your_openai_key")
# ===== Step 1: 数据摄入 =====def ingest_documents(doc_paths):    conn = psycopg2.connect(**DB_CONFIG)    cursor = conn.cursor()
    converter = DocumentConverter()    chunker = HybridChunker(max_tokens=256)
    for doc_path in doc_paths:        print(f"Processing: {doc_path}")
        # 转换文档        result = converter.convert(doc_path)        doc = result.document
        # 分块        chunks = list(chunker.chunk(doc))
        # 向量化并存储        for chunk in chunks:            # 生成embedding            embedding = embedding_model.encode(chunk.text).tolist()
            # 插入数据库            cursor.execute("""                INSERT INTO document_chunks (                    document_path, chunk_text, embedding, metadata                ) VALUES (%s, %s, %s, %s)            """, (                doc_path,                chunk.text,                embedding,                {"headings": chunk.meta.headings}            ))
        conn.commit()        print(f"  Ingested {len(chunks)} chunks")
    cursor.close()    conn.close()
# ===== Step 2: 检索函数 =====def retrieve_context(query, top_k=5):    conn = psycopg2.connect(**DB_CONFIG)    cursor = conn.cursor()
    # 查询向量化    query_embedding = embedding_model.encode(query).tolist()
    # 向量检索（使用cosine similarity）    cursor.execute("""        SELECT chunk_text, metadata,                1 - (embedding <=> %s::vector) AS similarity        FROM document_chunks        ORDER BY similarity DESC        LIMIT %s    """, (query_embedding, top_k))
    results = cursor.fetchall()    cursor.close()    conn.close()
    return [        {            "text": row[0],            "metadata": row[1],            "similarity": row[2]        }        for row in results    ]
# ===== Step 3: RAG查询 =====def rag_query(question):    # 检索相关上下文    contexts = retrieve_context(question, top_k=3)
    # 构建提示词    context_text = "\n\n".join([        f"Context {i+1} (Similarity: {ctx['similarity']:.3f}):\n{ctx['text']}"        for i, ctx in enumerate(contexts)    ])
    prompt = f"""Based on the following context, answer the question.
Context:{context_text}
Question: {question}
Answer:"""
    # 调用LLM生成回答    response = openai_client.chat.completions.create(        model="gpt-4",        messages=[            {"role": "system", "content": "You are a helpful assistant."},            {"role": "user", "content": prompt}        ]    )
    answer = response.choices[0].message.content
    return {        "answer": answer,        "contexts": contexts    }
# ===== 使用示例 =====if __name__ == "__main__":    # 摄入文档    documents = [        "data/product_manual.pdf",        "data/faq.docx",        "data/support_guide.pdf"    ]    ingest_documents(documents)
    # 执行查询    question = "How do I install the device?"    result = rag_query(question)
    print(f"Question: {question}")    print(f"Answer: {result['answer']}")    print(f"\nSources:")    for i, ctx in enumerate(result['contexts'], 1):        print(f"  {i}. {ctx['metadata']['headings']} (Score: {ctx['similarity']:.3f})")

┌─────────────────────────────────────────────┐│              用户查询                        │└─────────────────────────────────────────────┘                    ↓┌─────────────────────────────────────────────┐│         检索模块 (Retrieval)                 ││  - 查询向量化                                ││  - 向量相似度计算                            ││  - Top-K结果返回                             │└─────────────────────────────────────────────┘                    ↓┌─────────────────────────────────────────────┐│         生成模块 (Generation)                ││  - 上下文组装                                ││  - LLM推理                                   ││  - 答案生成                                  │└─────────────────────────────────────────────┘                    ↓┌─────────────────────────────────────────────┐│              返回结果                        ││  - 答案文本                                  ││  - 来源引用                                  ││  - 置信度评分                                │└─────────────────────────────────────────────┘

import osfrom pathlib import Pathfrom docling.document_converter import DocumentConverterfrom docling_core.transforms.chunker import HybridChunker
def build_knowledge_base(root_dir, output_db):    converter = DocumentConverter()    chunker = HybridChunker(max_tokens=256)
    # 遍历所有文档    all_files = []    for ext in ['*.pdf', '*.docx', '*.pptx', '*.mp3']:        all_files.extend(Path(root_dir).rglob(ext))
    print(f"Found {len(all_files)} documents")
    total_chunks = 0    for file_path in all_files:        try:            # 转换并分块            result = converter.convert(str(file_path))            chunks = list(chunker.chunk(result.document))
            # 存入数据库（简化示例）            store_to_database(                file_path=str(file_path),                chunks=[chunk.text for chunk in chunks],                metadata=result.document.metadata            )
            total_chunks += len(chunks)            print(f"✓ {file_path.name}: {len(chunks)} chunks")
        except Exception as e:            print(f"✗ {file_path.name}: {str(e)}")
    print(f"\nTotal: {total_chunks} chunks from {len(all_files)} documents")
# 执行构建build_knowledge_base("company_docs/", "knowledge_base.db")

用户提问 → 意图识别 → Docling知识库检索 → 答案生成（模板/LLM） → 人工复核（高风险） → 回复用户

def customer_service_bot(question):    # 检索相关知识    contexts = retrieve_from_docling_kb(question, top_k=3)
    # 判断置信度    if contexts[0]['similarity'] > 0.9:        # 高置信度：使用模板答案        return contexts[0]['text']    elif contexts[0]['similarity'] > 0.7:        # 中置信度：LLM改写        return llm_rewrite(contexts[0]['text'], question)    else:        # 低置信度：转人工        return "您的问题已转接人工客服，请稍候..."

from docling.document_converter import DocumentConverterfrom docling_core.transforms.chunker import HybridChunker
def analyze_legal_document(contract_pdf):    # Step 1: 提取文档结构    converter = DocumentConverter()    result = converter.convert(contract_pdf)    doc = result.document
    # Step 2: 识别条款    clauses = []    for item in doc.body.texts:        if item.type == "heading":            clauses.append({                "title": item.content,                "level": item.level,                "content": []            })        elif item.type == "paragraph" and clauses:            clauses[-1]["content"].append(item.content)
    # Step 3: LLM分析风险    risk_analysis = []    for clause in clauses:        prompt = f"Analyze legal risks in this clause:\n{clause}"        risk = llm_analyze(prompt)        risk_analysis.append({            "clause": clause['title'],            "risk_level": risk['level'],            "details": risk['description']        })
    return {        "document": contract_pdf,        "total_clauses": len(clauses),        "risk_analysis": risk_analysis    }
# 批量分析contracts = ["contract_1.pdf", "contract_2.pdf", "contract_3.pdf"]for contract in contracts:    analysis = analyze_legal_document(contract)    print(f"Contract: {contract}")    print(f"  High-risk clauses: {sum(1 for r in analysis['risk_analysis'] if r['risk_level'] == 'high')}")

import arxivfrom docling.document_converter import DocumentConverterfrom docling_core.transforms.chunker import HybridChunker
def research_assistant(topic, num_papers=10):    # Step 1: 搜索arxiv论文    search = arxiv.Search(        query=topic,        max_results=num_papers,        sort_by=arxiv.SortCriterion.SubmittedDate    )
    papers = []    converter = DocumentConverter()
    for result in search.results():        # Step 2: 下载并解析论文        pdf_url = result.pdf_url        doc_result = converter.convert(pdf_url)
        # Step 3: 提取关键信息        abstract = ""        conclusions = ""
        for item in doc_result.document.body.texts:            if "abstract" in item.content.lower()[:50]:                abstract = item.content            if "conclusion" in item.content.lower()[:50]:                conclusions = item.content
        papers.append({            "title": result.title,            "authors": [a.name for a in result.authors],            "date": result.published,            "abstract": abstract,            "conclusions": conclusions,            "url": result.pdf_url        })
    # Step 4: LLM生成综述    summaries = [f"Paper: {p['title']}\nKey findings: {p['conclusions']}" for p in papers]    synthesis = llm_synthesize("\n\n".join(summaries))
    return {        "topic": topic,        "num_papers": len(papers),        "synthesis": synthesis,        "papers": papers    }
# 使用示例result = research_assistant("Retrieval Augmented Generation", num_papers=20)print(result['synthesis'])

        需要处理文档类型？                │      ┌─────────┴─────────┐     多种格式              单一格式（如仅PDF）      │                   │   是否有合规要求？     性能要求高？      │                   │  ┌───┴───┐          ┌───┴───┐ 数据敏感  可上云      是      否  │        │         │        │Docling  Textract  pdfplumber PyPDF2(本地)   (云服务)   (快速)    (简单)

from langchain_community.document_loaders import DoclingLoaderfrom langchain.text_splitter import DoclingChunker
# 加载文档loader = DoclingLoader(file_path="document.pdf")documents = loader.load()
# 使用Docling分块器chunker = DoclingChunker(max_tokens=256)chunks = chunker.split_documents(documents)
# 后续流程（向量化、存储）与LangChain标准流程一致

from llama_index.readers.docling import DoclingReaderfrom llama_index.core import VectorStoreIndex
# 读取文档reader = DoclingReader()documents = reader.load_data(file="document.pdf")
# 构建索引index = VectorStoreIndex.from_documents(documents)
# 查询query_engine = index.as_query_engine()response = query_engine.query("What is the installation process?")

from haystack.nodes import DoclingConverterfrom haystack.pipelines import Pipeline
# 创建转换节点converter = DoclingConverter()
# 构建Pipelinepipeline = Pipeline()pipeline.add_node(component=converter, name="Converter", inputs=["File"])pipeline.add_node(component=retriever, name="Retriever", inputs=["Converter"])pipeline.add_node(component=generator, name="Generator", inputs=["Retriever"])
# 运行result = pipeline.run(file_paths=["document.pdf"])

# 启动Docling MCP Serverdocling-mcp-server --host 0.0.0.0 --port 8080
# AI Agent可以通过MCP协议调用Docling# 支持的工具：# - parse_document: 解析文档# - chunk_document: 分块文档# - extract_tables: 提取表格# - transcribe_audio: 转录音频

{  "mcpServers": {    "docling": {      "url": "http://localhost:8080",      "description": "Document processing for RAG"    }  }}

User: 帮我分析这份PDF合同Claude: [调用 docling.parse_document]         [分析文档内容]        [生成分析报告]

┌─────────────────────────────────────────┐│          前端应用层                      ││  - Web UI / ChatBot / API Gateway       │└─────────────────────────────────────────┘                  ↓┌─────────────────────────────────────────┐│          业务逻辑层                      ││  - RAG Orchestrator                     ││  - Query Router                         ││  - Answer Generator                     │└─────────────────────────────────────────┘                  ↓┌─────────────────────────────────────────┐│       文档处理层 (Docling Cluster)       ││  ┌──────────┐  ┌──────────┐  ┌────────┐││  │ Worker 1 │  │ Worker 2 │  │Worker N│││  │ (PDF)    │  │ (Audio)  │  │(Excel) │││  └──────────┘  └──────────┘  └────────┘│└─────────────────────────────────────────┘                  ↓┌─────────────────────────────────────────┐│         数据存储层                       ││  - Vector DB (Pinecone/Weaviate)        ││  - Document Store (S3/MinIO)            ││  - Metadata DB (PostgreSQL)             │└─────────────────────────────────────────┘