05 项目实战：企业知识库

项目概述：企业知识库RAG系统实战

本项目以RAG-Challenge-2（处理公司年度报告）冠军方案为参考，旨在搭建一个高性能的企业知识库RAG系统。该系统在比赛中展现了强大的问题回答能力，其核心特点包括：定制PDF解析（Docling）、向量搜索与父文档检索、LLM重排序、链式推理与结构化输出提示以及多公司比较的查询路由功能。

企业RAG挑战赛任务要求：在2.5小时内解析100份PDF年度报告，构建数据库，并快速回答100个随机问题。答案必须准确并注明引用的页码，以避免模型幻觉。

基础RAG系统流程

一个基础的RAG系统包含以下核心开发流程：

解析 (Parsing)：准备知识库数据，包括文档收集、格式转换和信息清理。
内容提取 (Ingestion)：创建并载入知识库，通常涉及分块和向量化。
检索 (Retrieval)：根据用户查询查找并返回相关数据，通常通过向量数据库进行语义搜索。
回答 (Answering)：将检索到的数据和用户提示词发送给LLM，生成最终答案。

解析模块（Parsing）

PDF解析是挑战性任务，需处理表格结构、格式元素（标题、列表）、多栏文本、图表等复杂情况。作者尝试了20多种解析器，最终选择了Docling。

Docling优化

尽管Docling表现优秀，但作者通过重写其部分方法，使其能够：

生成包含必要元数据的JSON文件。
利用JSON构建Markdown文档，将表格从PDF转换为Markdown或HTML。
使用正则表达式处理解析错误，清理语法问题。

表格序列化

表格序列化旨在解决大型表格中度量名称与表头语义连贯性差、LLM理解困难的问题。然而，在实际测试中，表格序列化功能反而略微降低了系统有效性，因此未采用。

关于表格格式：以HTML格式输入表格给LLM，因为LLM对HTML的理解程度更高，能描述更复杂的表格结构（如合并单元格）。

内容提取（Ingestion）

分块 (Chunking)

为了精确引用页码并提高相关性得分，将文档的整页切割为300个token（约15个句子）的块。每个块存储其ID及其在元数据中的父页面编号。

向量化 (Vectorization)

针对100家公司的文档，采用为每家公司创建一个独立的Faiss数据库的方式，而非混合存储。这使得检索更加清晰高效。

Faiss数据库：使用IndexFlatIP方法创建，优点是精度高，缺点是计算和内存消耗高。
嵌入模型：text-embedding-3-large（可根据需求替换）。

Python

# 示例：Faiss数据库创建（概念性代码，非完整实现）
import faiss
import numpy as np

def create_faiss_db(embeddings: np.ndarray) -> faiss.Index:
    # d是向量维度
    d = embeddings.shape[1]
    # IndexFlatIP表示使用内积作为相似度度量
    index = faiss.IndexFlatIP(d)
    index.add(embeddings)
    return index

# 假设embeddings_company_A是某公司文档块的嵌入向量
# embeddings_company_A = np.random.rand(1000, 1536).astype('float32') # 1000个块，维度1536
# faiss_db_company_A = create_faiss_db(embeddings_company_A)

检索（Retrieval）

RAG系统中的检索是关键部分。虽然混合搜索 (vDB + BM25) 理论上能结合语义和关键词匹配提高准确性，但在比赛的基础实现中，它通常会降低检索质量，因此需要谨慎评估。

重排序（Reranking）

重排序用于提高检索结果的上下文相关性。Jina Reranker是流行的选择，但比赛中作者使用了LLM重排序。

LLM重排序 (LLM Reranking)

LLM重排序通过让LLM对检索到的文本块进行格式化输出，包含reasoning和relevance_score。最终相关性得分通过加权平均计算（例如vector_weight = 0.3, llm_weight = 0.7）。这种方式在成本效益上优于直接将每一页传递给LLM。

Python

# system_prompt_rerank_single_block 示例
system_prompt_rerank_single_block = """
你是一个RAG检索重排专家。
你将收到一个查询和一个检索到的文本块，请根据其与查询的相关
性进行评分。
评分说明：
1. 推理：分析文本块与查询的关系，简要说明理由。
2. 相关性分数（0-1，步长0.1）：
0 = 完全无关
0.1 = 极弱相关
...
1 = 完全匹配
3. 只基于内容客观评价，不做假设。
"""

父页面检索 (Parent Page Retrieval)

尽管问题核心信息常在小块中，但同一页面的其他部分可能包含重要细节。因此，会先检索出Top N最相关的文本块，这些块作为“指针”来定位对应的完整页面，随后将整个页面的内容纳入上下文进行分析。每个块的元数据中记录了所属页码。

整合后的检索器 (Assembled Retriever)

最终的检索器步骤：

查询向量化。
找到Top 30相关块。
提取并去重对应的页面。
通过LLM重排序处理这些页面，调整相关性得分。
返回得分最高的Top 10页，并合并成带页码的字符串。

增强 (Augmentation)

这部分主要涉及提示词管理。项目将提示词存储在专门的prompts.py文件中，并分割成逻辑块：

核心系统指令：定义LLM角色和遵循规则。
Pydantic schema：定义LLM返回响应的JSON格式，强制结构化输出。
单次/少次示例：通过问答对示例指导LLM输出格式和风格。
上下文和查询模板：用于插入动态内容。

Python

# 1. 核心系统指令示例
class AnswerWithRAGContextSharedPrompt:
    instruction = """
    你是一个RAG（检索增强生成）问答系统。
    你的任务是仅基于公司年报中RAG检索到的相关页面内容，回答给定问题。
    ...
    """

# 2. Pydantic schema 示例
from pydantic import BaseModel, Field
from typing import List, Union, Literal

class ComparativeAnswerPrompt:
    class AnswerSchema(BaseModel):
        step_by_step_analysis: str = Field(description="详细分步推理过程，至少5步，150字以上。")
        reasoning_summary: str = Field(description="简要总结推理过程，约50字。")
        relevant_pages: List[int] = Field(description="保持为空列表。")
        final_answer: Union[str, Literal["N/A"]] = Field(description="公司名称需与问题中完全一致。答案只能是单个公司名或'N/A'。")

# 3. 示例问答对
example = r"""示例：
问题：
"下列公司中，哪家2022年总资产最低："A公司", "B公司", "C公司"？若无数据则排除。"
答案：
{
"step_by_by_analysis": "...",
"reasoning_summary": "...",
"relevant_pages": [],
"final_answer": "C公司"
}
"""

# 4. 上下文和查询模板
user_prompt = """
以下是上下文:
\"\"\"
{context}
\"\"\"
---
以下是问题：
"{question}"
"""

生成 (Generation)

生成阶段需要巧妙运用多种技术，其中查询路由是关键。

技术1：将查询路由到数据库

通过提取问题中的公司名，将查询路由到对应的公司向量数据库，将搜索范围缩小。

技术2：将查询路由到提示词

根据问题预期的答案类型（如int/float/bool/str/list[str]），将复杂查询拆解为多个简单步骤，或选择专门设计的提示词模板，以减少LLM的认知负荷，确保输出格式的精确性。

技术3：复合查询路由

处理多公司比较问题时，采用分步拆解：

问题拆解：LLM将原始比较问题拆解为多个独立子问题（如“苹果营收是多少？”）。
并行查询：对每个子问题独立处理，获取各公司指标数据。
综合判断：将收集到的数据作为上下文，由LLM进行最终比较并生成答案。

思维链 (Chain of Thoughts, CoT)

CoT通过让模型“出声思考”（如“一步步思考”）来提高答案质量。为了确保CoT有效，需要清晰引导模型推理步骤、目标并提供示例。

Plain

# 思维链推理示例（截取自原文，非代码）
模型的推理步骤：
1. 问题询问Ritter Pharmaceuticals Inc. 的“研发设备，按成本计”。这表明需要从资产负债表中提取一个特定数值，代表
专门用于研发的设备的原始购置成本，不包含任何累计折旧。
...
5. 由于上下文没有提供仅用于研发设备的原始成本，且我们无法进行假设、计算，因此答案是“N/A”。

结构化输出 (Structured Outputs, SO)

SO通过API传递Pydantic schema，强制模型返回标准化格式（如JSON），确保输出符合预定结构，便于后续解析和校验。

Python

# 结构化输出Pydantic schema 示例
class RetrievalRankingSingleBlock(BaseModel):
    """Rank retrieved text block relevance to a query."""
    reasoning: str = Field(
        description=(
            "Analysis of the block, identifying key information and how it"
            "relates to the query"
        )
    )
    relevance_score: float = Field(
        description=(
            "Relevance score from 0 to 1, where 0 is Completely Irrelevant "
            "and 1 is Perfectly Relevant"
        )
    )

CoT + SO （思维链+结构化输出）

将CoT和SO结合，模型在生成阶段有一个专门用于推理的字段（step_by_step_analysis），和一个用于最终答案的字段（final_answer）。这使得答案可以直接提取，且推理过程可追溯。

指令细化指判断“答案的灵活范围”和处理“未找到答案”的情况。这需要与客户提前确定规则、收集边缘案例进行测试，并进行大量迭代调试和手动分析。

提示词创建与 `Prompt.py` 实现

通过创建验证集并手动回答问题，团队量化了系统改进并发现了隐性规则，最终将这些洞察整合到提示词设计中。

prompts.py文件根据问题类型（kind字段）定义了多种提示词模板：

boolean：输出是/否。
number：输出数值。
name：输出实体名（公司、人名、产品名）。
names：输出实体名列表。
string：开放性问题，回答一段文本（需要手动添加实现）。

Python

# prompts.py 中 AnswerWithRAGContextBooleanPrompt 示例
import inspect
import re

class AnswerWithRAGContextBooleanPrompt:
    instruction = AnswerWithRAGContextSharedPrompt.instruction
    user_prompt = AnswerWithRAGContextSharedPrompt.user_prompt
    class AnswerSchema(BaseModel):
        step_by_step_analysis: str = Field(description="""
        详细分步推理过程，至少5步，150字以上。特别注意问题措辞，避
        免被迷惑。有时上下文中看似有答案，但可能并非所问内容，仅为
        相似项。
        """)
        reasoning_summary: str = Field(description="简要总结分步推理过
        程，约50字。")
        relevant_pages: List[int] = Field(description="""
        仅包含直接用于回答问题的信息页面编号。只包括：
        - 直接包含答案或明确陈述的页面
        - 强有力支持答案的关键信息页面
        不要包含仅与答案弱相关或间接相关的页面。
        列表中至少应有一个页面。
        """)        
        final_answer: Union[bool] = Field(description="""
        一个从上下文中精确提取的布尔值（True或False），直接回答问题。
        如果问题问某事是否发生，且上下文有相关信息但未发生，则返回
        False。
        """)
    pydantic_schema = re.sub(r"^ {4}", "", 
    inspect.getsource(AnswerSchema), flags=re.MULTILINE)
    example = r"""
    问题：
    "'万科企业股份有限公司'年报是否宣布了分红政策变更？"
    答案：
    ```
    {
    "step_by_step_analysis": "1. 问题询问是否有分红政策变更。\n2. 年
    报12、18页提到年度分红金额增加，但政策未变。\n3. 45页有分红细
    节。\n4. 持续小幅增长，符合既定政策。\n5. 问题问的是政策变更，
    非金额变化。",
    "reasoning_summary": "年报显示分红金额变化但政策未变，答案为
    False。",
    "relevant_pages": [12, 18, 45],
    "final_answer": false
    }
    ```
    """

RAG系统调参

通过配置化关键功能和超参数（如use_serialized_tables, parent_document_retrieval, llm_reranking等），可以衡量实际效果并进行微调。

Python

# RunConfig 示例
class RunConfig:
    # 运行流程参数配置
    use_serialized_tables: bool = False
    parent_document_retrieval: bool = False
    use_vector_dbs: bool = True
    use_bm25_db: bool = False
    llm_reranking: bool = False
    llm_reranking_sample_size: int = 30
    top_n_retrieval: int = 10
    parallel_requests: int = 1 # 并行的数量，需要限制，否则qwen-turbo会超出阈值
    pipeline_details: str = ""
    submission_file: bool = True
    full_context: bool = False
    api_provider: str = "dashscope" #openai
    answering_model: str = "qwen-turbo-latest" 
    config_suffix: str = ""

CASE：打造自己的RAG系统

Step1：运行RAG-Challenge-2

python -m src.pipeline

Step2：简化Markdown生成与MinerU解析

原有的pipeline处理过程复杂，可以简化为：解析PDF为JSON -> 序列化表格 -> JSON规整为每页Markdown -> 导出Markdown。

用MinerU替换Docling进行PDF解析，并上传文件到URL或使用本地MinerU。

Python

# pdf_mineru.py 示例
import requests
import time
import zipfile

api_key = '你的api_key'

def get_task_id(file_name):
    url='https://mineru.net/api/v4/extract/task'
    header = {
        'Content-Type':'application/json',
        "Authorization":f"Bearer {api_key}"
    }
    pdf_url = 'https://vl-image.oss-cn-shanghai.aliyuncs.com/pdf/' + file_name # 示例URL，实际可能需要上传或本地路径
    data = {
        'url':pdf_url,
        'is_ocr':True,
        'enable_formula': False,
    }
    res = requests.post(url,headers=header,json=data)
    print(res.status_code)
    print(res.json())
    task_id = res.json()["data"]['task_id']
    return task_id

def get_result(task_id):
    url = f'https://mineru.net/api/v4/extract/task/{task_id}'
    header = {
        'Content-Type':'application/json',
        "Authorization":f"Bearer {api_key}"
    }
    while True:
        res = requests.get(url, headers=header)
        result = res.json()["data"]
        print(result)
        state = result.get('state')
        if state in ['pending', 'running']:
            print("任务未完成，等待5秒后重试...")
            time.sleep(5)
            continue
        if state == 'done':
            full_zip_url = result.get('full_zip_url')
            if full_zip_url:
                local_filename = f"{task_id}.zip"
                r = requests.get(full_zip_url, stream=True)
                with open(local_filename, 'wb') as f:
                    for chunk in r.iter_content(chunk_size=8192):
                        if chunk:
                            f.write(chunk)
                print(f"下载完成，已保存到: {local_filename}")
                unzip_file(local_filename)
            return
        print(f"未知状态或错误: {state}, {result.get('err_msg', '')}")
        return

def unzip_file(zip_path, extract_dir=None):
    if extract_dir is None:
        extract_dir = zip_path.rstrip('.zip')
    with zipfile.ZipFile(zip_path, 'r') as zip_ref:
        zip_ref.extractall(extract_dir)
    print(f"已解压到: {extract_dir}")

if __name__ == "__main

Step3：对Markdown进行切分

改写text_splitter.py，添加split_markdown_reports和split_markdown_file函数，用于按行分割Markdown文件，生成适合向量化的分块JSON。

Python

# text_splitter.py 中添加的函数概念
from pathlib import Path
from typing import List, Dict, Optional

def split_markdown_reports(all_md_dir: Path, output_dir: Path, 
                             chunk_size: int = 30, chunk_overlap: int = 5, 
                             subset_csv: Optional[Path] = None):
    """
    批量处理目录下所有markdown文件，分块并输出为json文件到目标目录。
    """
    # 实现细节：遍历md_dir，对每个md文件调用 split_markdown_file
    pass

def split_markdown_file(md_path: Path, chunk_size: int = 30, 
                        chunk_overlap: int = 5) -> List[Dict]:
    """
    按行分割markdown文件，每个分块记录起止行号和内容。
    """
    # 实现细节：读取md文件，按行切割，根据chunk_size和chunk_overlap生成块
    pass

kind=string 的实现

为开放性问题添加AnswerWithRAGContextStringPrompt功能，在prompts.py中定义其Pydantic schema和示例，并在api_request.py中添加对kind="string"的处理逻辑。

搭建RAG系统界面

使用Streamlit或Gradio搭建可视化界面，用户输入问题后，后台调用类似pipeline.process_questions()，默认kind=string，并将结果显示在界面上。

关键要点

系统化而非单一“神奇方案”：RAG成功依赖于流程优化和多技术精细调整（解析、检索、路由、排序、提示词）。
数据预处理至关重要：高质量的PDF解析（如Docling优化、HTML表格）是RAG效果的基础。
精准检索与LLM重排序：通过细粒度分块、父页面检索及LLM重排序，显著提升相关性。
智能路由与结构化输出：查询路由到特定数据库或提示词，结合Pydantic schema实现高效、精确的结构化回答。
迭代优化与指令细化：通过验证集、手动分析和指令细化，不断提升LLM回答质量和符合业务规则。

embed

05 项目实战：企业知识库

项目概述：企业知识库RAG系统实战

基础RAG系统流程

解析模块（Parsing）

Docling优化

表格序列化

内容提取（Ingestion）

分块 (Chunking)

向量化 (Vectorization)

检索（Retrieval）

重排序（Reranking）

LLM重排序 (LLM Reranking)

父页面检索 (Parent Page Retrieval)

整合后的检索器 (Assembled Retriever)

增强 (Augmentation)

生成 (Generation)

技术1：将查询路由到数据库

技术2：将查询路由到提示词

技术3：复合查询路由

思维链 (Chain of Thoughts, CoT)

结构化输出 (Structured Outputs, SO)

CoT + SO （思维链+结构化输出）

指令细化 (Instruction Refinement)

提示词创建与 Prompt.py 实现

RAG系统调参

CASE：打造自己的RAG系统

Step1：运行RAG-Challenge-2

Step2：简化Markdown生成与MinerU解析

Step3：对Markdown进行切分

kind=string 的实现

搭建RAG系统界面

关键要点

提示词创建与 `Prompt.py` 实现