从 PDF 到 MCP Server：手搓一套财报智能问答系统的全流程实录（混合检索 + 手写 ReAct Agent + 踩坑合集）

2301_81970386

844人浏览 · 2026-09-09 16:06:38

2301_81970386 · 2026-09-09 16:06:38 发布

从 PDF 到 MCP Server：手搓一套财报智能问答系统的全流程实录（混合检索 + 手写 ReAct Agent + 踩坑合集）

关键词：RAG、混合检索、FAISS、BM25、RRF、BGE-M3、ReAct Agent、MCP、qwen-max

适用读者：有一定 Python 基础、了解大模型基本用法，想动手搭一个"能跑起来、能拿得出手"的 RAG 项目的同学。全文基于我真实落地的一个项目写成，所有代码和踩坑都是实录，不是纸上谈兵。

前言：为什么财报问答必须自己做
一、系统整体架构与技术选型
二、环境准备与项目结构
三、数据流水线：从 PDF 到可检索的知识库
四、检索层核心：混合检索三件套
五、生成层：让 LLM 可靠地回答
六、手写 ReAct Agent：一次检索答不了的题怎么办
七、封装 MCP Server：让任何 Agent 都能调用你的知识库
八、踩坑实录：这 6 个坑我替你踩过了
九、局限与迭代方向
十、总结

前言：为什么财报问答必须自己做

如果你拿"中芯国际 2025 年一季度毛利率是多少"去问通用大模型，大概率会得到两种结果：要么一本正经地编一个数，要么老实说不知道。原因很简单——券商研报、公司年报、机构调研纪要都是私有数据，模型训练时根本没见过。

而这些恰恰是金融领域最有价值的数据。想让 LLM 回答这类问题，RAG（Retrieval-Augmented Generation，检索增强生成）是标准解法：先从私有文档里检索相关片段，再把片段塞进 prompt 让 LLM 基于事实作答。

但财报场景和普通 RAG Demo（比如"读一个 PDF 聊天"）相比，有三个实打实的难点：

数值必须精确。“目标价 63 港币"检索成"53 港币”，差一个数字就是事故。这对检索的关键词精确匹配能力要求极高，纯向量检索撑不住。
黑话和缩写极多。研报里写 1Q25、capex、归母净利润、FY2025E，用户提问却可能说"2025 年第一季度"“资本开支”“全年预测”。同一个意思，字面完全对不上，关键词检索又撑不住了。
很多问题一次检索答不了。“A 券商和 B 券商对毛利率的预测有什么分歧？”——答案分散在两份不同文档里，单轮检索的上下文根本凑不齐。

这篇文章记录我是怎么一步步解决这三个问题的。先看全貌。

一、系统整体架构与技术选型

整个系统的数据流如下：

知识库是中芯国际 2024-2025 年的 9 份文档：7 份券商研报（华泰、光大、东方、国信、中原、上海证券、兴证国际）+ 2024 年年度报告 + 机构调研纪要。

技术选型一览，以及每一项"为什么这么选"：

组件	选型	为什么
PDF 解析	MinerU	研报里全是表格，MinerU 对表格的 Markdown 还原效果最好
Embedding	BGE-M3 本地 CPU 推理	私有数据不出域 + 零 API 成本；1024 维稠密向量，中文效果强
向量库	FAISS `IndexFlatIP`	9 份文档量级用暴力检索即可，精确且无需调参
关键词检索	BM25（rank_bm25）	股票代码、精确数字、术语缩写必须靠字面匹配
双路融合	RRF	两路分数量纲不可比，排名融合天然免疫
LLM	qwen-max（DashScope）	中文金融语料理解好，结构化输出稳定，价格友好
Agent	手写 ReAct（无框架）	逻辑完全可控，出问题能定位到每一轮
对外服务	FastMCP 封装成 MCP Server	任何支持 MCP 的 Agent（Claude Desktop、GPT Researcher）即插即用

二、环境准备与项目结构

2.1 环境搭建

推荐用 conda 独立环境，Python 3.10+：

conda create -n rag_fin python=3.10 -y
conda activate rag_fin

# 核心依赖（faiss-cpu 建议用 conda 装，pip 版偶有 DLL 问题）
pip install -r requirements.txt
pip install dashscope modelscope FlagEmbedding streamlit

# MCP 封装需要
pip install fastmcp mcp

2.2 API Key 配置

项目根目录放一个 .env：

DASHSCOPE_API_KEY=sk-你的密钥

代码里统一用 python-dotenv 读取，密钥永远不要硬编码进代码（发博客、传 GitHub 前记得检查）。

2.3 项目结构

RAG-cy/
├── src/                      # 核心模块
│   ├── ingestion.py          #   BGE-M3 加载 + FAISS/BM25 索引构建
│   ├── text_splitter.py      #   分块 + 噪声过滤
│   ├── query_expansion.py    #   查询扩展（正则）
│   ├── query_router.py       #   题型路由（LLM）
│   ├── retrieval.py          #   双路检索器
│   ├── rrf_fusion.py         #   RRF 融合 + 后处理
│   ├── agent.py              #   手写 ReAct Agent
│   ├── questions_processing.py # 问答主逻辑
│   └── pipeline.py           #   流程调度
├── mcp_adapter.py            # MCP Server 封装
├── app_streamlit.py          # 问答界面
└── data/stock_data/
    ├── pdf_reports/          # 9 份原始 PDF
    └── databases/            # 三套索引产物
        ├── chunked_reports/  #   分块后的 JSON
        ├── vector_dbs/       #   FAISS 索引
        └── bm25_dbs/         #   BM25 索引

三、数据流水线：从 PDF 到可检索的知识库

3.1 PDF → Markdown

用 MinerU 把每份 PDF 解析成 Markdown。这一步最关键的不是正文，是表格——研报里的核心数据（营收预测表、财务摘要表）几乎全在表格里，解析工具对表格的还原质量直接决定后面问答的上限。

3.2 分块策略：按"行"而不是按"字符"

我没有用常见的"固定 512 token 切一刀"，而是按 Markdown 的行来分块：每 30 行一块，相邻块重叠 5 行。

为什么？财报 Markdown 里，一张表格动辄十几行、跨页表格更长。按 token 数硬切会把一张表拦腰斩断——上半截在块 A、下半截在块 B，两块单独看都是废数据。按行分块 + 重叠，配合对表格起止的识别，可以基本保证表格的完整性。同时我用 tiktoken（o200k_base）对每个块统计真实 token 数存进元数据，供后面做上下文预算。

分块时还做了一层噪声过滤：研报目录页、页眉页脚、"请务必阅读正文之后的免责条款"这类对检索毫无价值的块，直接在入库前丢弃。

3.3 构建双索引

向量索引：用 BGE-M3 把每个 chunk 编码成 1024 维稠密向量，存入 FAISS。这里有个非常容易踩的坑，单独强调：

FAISS 内积检索 ≠ 余弦相似度，除非你先做 L2 归一化。

BGE-M3 输出的向量不是单位向量，直接用 IndexFlatIP（内积）算出来的分数会被向量长度污染，排序结果是错的。正确做法是入库前 faiss.normalize_L2(embeddings)，查询向量同样归一化，此时内积严格等价余弦相似度。一行代码的事，但没有它整个检索层的排序都是乱的。

关键词索引：BM25（rank_bm25 库的 BM25Okapi），对每个 chunk 的文本建倒排索引。它的分词细节里藏着一个大坑，放第八章细说。

BGE-M3 是本地推理的：从 ModelScope 下载约 2.2GB 权重，CPU 上 fp16 跑，max_length=8192。首次加载十几秒，所以做了全局懒加载单例——全项目只加载一次，所有模块共享。

四、检索层核心：混合检索三件套

现在回到前言里的三个难点。难点 1（数值精确）和难点 2（术语对不上）本质是矛盾的：前者要字面匹配，后者要语义理解。我的解法是双路检索 + 融合，但在此之前，先做两件更便宜的事。

4.1 查询扩展：纯正则，零成本解决"术语对不上"

用 LLM 改写 query 也可以，但每次提问多一次 API 调用，延迟和成本都上去了。而财报术语的互译其实是封闭集合，用正则就能覆盖：

时间表达双向互译：2025年第一季度 ↔ 1Q25 / 25Q1 / Q1 2025；2025年全年 ↔ FY2025；2025年预测 ↔ 2025E
财务术语中英互译：毛利率 ↔ gross margin、资本支出 ↔ capex、归母净利润、EBITDA、产能利用率 ↔ capacity utilization 等十组高频词

一个查询进来，扩展出若干变体，每个变体分别送进两路检索器，结果合并。因为研报里中文缩写、英文缩写、全称混着用，变体越多，命中率越高——而且纯正则，零 API 调用，零延迟。

4.2 题型路由：不同的问题，不同的检索配方

我用 qwen-max 做了一个轻量分类器，把问题分成 7 类，每类给不同的检索参数：

题型	例子	向量权重	BM25 权重	特殊处理
数值查询	“毛利率是多少”	0.3	0.7	—
排名占比	“中国区收入占比”	0.3	0.7	—
时间比较	“同比增长多少”	0.5	0.5	—
跨文档比较	“A 券商 vs B 券商预测”	0.5	0.5	召回量×2，强制多文档
因果解释	“为什么下降”	0.7	0.3	—
预测指引	“目标价是多少”	0.7	0.3	—
异常识别	“增收不增利怎么回事”	0.5	0.5	召回量×1.5，按时间排序

直觉很容易理解：问精确数值的，BM25 权重高（数字和术语必须字面命中）；问原因和预测的，向量权重高（同样的意思说法五花八门，靠语义）。分类器只有一次轻量调用（temperature=0.1），失败时静默回退到默认配置，不阻塞主链路。

4.3 RRF 融合：为什么不用分数加权

两路检索器各自返回 top N 后，怎么合并？最直觉的想法是分数加权求和，但这里有个隐蔽的坑：两路的分数量纲完全不可比。余弦相似度在 0~1 之间，BM25 分数无上界（可以到几十），直接加权等于让 BM25 一票独大。

我用的是 RRF（Reciprocal Rank Fusion），只看排名不看分数：

score(doc) = Σ  weight_i × 1 / (k + rank_i(doc) + 1)    # k = 60

一个文档如果在向量路排第 1、BM25 路排第 3，它的 RRF 分就是 0.5×(1/61) + 0.5×(1/64)。哪路分数通胀、哪路分数保守，统统不影响——排名是公平货币。k=60 是业界经验值，作用是平滑头部排名的差距。

4.4 后处理四板斧

融合之后还有四道清洗：

按文本去重：查询变体可能检索回同一块，保留分数最高的；
封面元数据过滤：券商研报的封面/首页全是分析师邮箱、执业证书编号、"股票与沪深 300 对比图"这类元数据，一通正则识别，纯元数据且无财务数字的块直接丢弃（有财务特征的豁免，防止误杀）；
多文档强制（跨文档比较题专用）：确保最终上下文至少来自 2 份不同文档，否则"比较"就无从谈起；
时间排序（异常识别题专用）：按文本中的年份/季度信息排序，最新的排前面。

五、生成层：让 LLM 可靠地回答

检索质量决定下限，生成层的工程决定上限。三个关键设计：

结构化输出。我不用自由文本，而是让 qwen-max 按 JSON schema 输出四个字段：step_by_step_analysis（分步推理）、reasoning_summary（一句话结论）、relevant_sources（引用了哪些来源，按编号）、final_answer（最终答案）。结构化之后，前端可以分区渲染，评测可以自动比对，引用可以校验。

上下文预算控制。检索回来的 chunk 拼成 RAG 上下文时，设了硬性预算：总量 20000 字符封顶，单块最多 3000 字符。不设限的话，一次塞十几块长文直接顶到 API 的 token 上限，而且长上下文会稀释关键数据的注意力。

引用校验，防"幻觉来源"。LLM 有时会声称"根据来源 7"——但本次只检索回了 5 块。我对 relevant_sources 里的编号做真实性校验：越界的剔除，一个都不填时用检索排名兜底补齐，再映射回真实的文件名和行号。这样界面上展示的每条引用都是可回溯到原文的。

六、手写 ReAct Agent：一次检索答不了的题

难点 3 登场：“光大证券和兴证国际对 2Q25 毛利率指引的预测有什么分歧？”——答案一半在光大、一半在兴证，而单轮检索的 query 只有一个重心，几乎不可能同时召回两份文档的关键段落。

我的方案是手写一个 ReAct Agent（约 300 行，零框架依赖）：让 LLM 进入 Thought → Action → Observation → … → Final Answer 的循环，自主决定检索几次、每次检索什么。

几个真正决定效果的设计细节：

工具带 source_hint 参数。rag_search(query, source_hint) 里 source_hint 可以指定"东方证券""机构调研"等来源，在 RRF 融合之前就按来源过滤——保证 Agent 指名要某家券商时，一定拿到那家的内容，而不是被高分的无关块挤掉。

Prompt 里写死防编造规则：第一轮必须先检索、不能直接作答；跨文档题必须分别检索每个来源；检索不到就明说"该来源未提供此数据"，严禁估算和推测；Final Answer 必须注明来源；最多 5 轮，超过强制收尾。金融场景里，"诚实地说没找到"远比"自信地编一个"值钱。

为什么不用 LangChain 的 Agent？ 手写的循环逻辑透明：每一轮的 Thought、Action、Observation 都是普通 Python 对象，打印出来就能调试；换框架反而要适应它的抽象。当然，代价是解析 LLM 输出、处理格式异常这些脏活得自己写——这是笔划算的交易，但你可以有自己的选择。

路由层会把"跨文档比较"和"因果解释"两类问题自动分流给 Agent，其他五类走单轮 RAG，各得其所。

七、封装 MCP Server：让任何 Agent 都能调用你的知识库

做完以上，系统已经是一个能用的问答应用了。但我想更进一步：把检索能力变成一个标准工具，让任何 AI Agent 都能直接调用——这就是 MCP（Model Context Protocol）。

7.1 30 秒理解 MCP

MCP 是 Anthropic 推出的开放协议，可以理解为 AI 应用的 USB-C 接口：你的程序实现一个 MCP Server，暴露若干工具（tool）；任何 MCP Client（Claude Desktop、GPT Researcher、Cursor……）连上之后，它的 LLM 就能自主决定何时调用你的工具。对我们来说，价值在于：GPT Researcher 这类研究 Agent 原本不知道中芯国际研报里写了什么，接上我的 MCP Server 之后，它就能按需检索我的私有知识库了。

7.2 用 FastMCP 封装

核心代码出乎意料地短——把 @mcp.tool() 装饰器往检索函数上一挂，一个工具就注册好了。我的工具签名：

@mcp.tool()
def search_semiconductor_reports(
    query: str,                       # 检索文本
    top_k: int = 5,                   # 返回条数, 1-20
    company_name: str = "中芯国际",
    enable_query_expansion: bool = True,
) -> list[dict]:                      # [{"content": ..., "source": ...}]

工具的 docstring 不是写给人看的，是写给调用方的 LLM 看的。 这一点被很多人忽视。我在 docstring 里写了两段关键内容：

适用范围：中芯国际财务数据、2024-2025 财报研报内容、晶圆代工行业分析、营收/毛利率/capex 等指标查询……
不适用范围：其他半导体公司、2023 年及更早的数据、实时行情、闲聊……

调用方的 Agent 就是靠这段描述来决定"这个问题该不该调这个工具"的。描述写得越清楚，Agent 的工具选择就越准，你的知识库被乱用的概率就越小。

7.3 两个致命坑（MCP 封装的核心难点）

坑 1：print 污染 stdio 协议流。 MCP 的 stdio 传输是：JSON-RPC 消息走进程的 stdout。而我的检索链路里到处是 print() 调试日志（ingestion、retrieval、计时信息），它们一执行，stdout 里就混进了非协议文本，客户端直接解析失败，表现为工具调用卡死或莫名超时。解法简单粗暴但有效——启动时把全局 print 劫持，重定向到 stderr（stderr 不参与协议传输）：

_real_print = builtins.print
def _print_to_stderr(*args, **kwargs):
    kwargs.setdefault("file", sys.stderr)
    _real_print(*args, **kwargs)
builtins.print = _print_to_stderr

坑 2：首次调用超时。 BGE-M3 模型加载要 10-15 秒，而 MCP 客户端（如 MCP Inspector）的默认请求超时等不了这么久——第一次调用工具必然超时。解法是 Server 启动时立刻用后台守护线程预热模型，等客户端真正发起首次调用时，模型早已在内存里待命；再加一把锁保证并发场景下模型只加载一次。

7.4 调试与接入

# 本地调试（自带可视化的 MCP Inspector）
mcp dev mcp_adapter.py

# 接入 Claude Desktop：在配置文件里加一段
{
  "mcpServers": {
    "semi-financial-reports": {
      "command": "python",
      "args": ["C:/path/to/mcp_adapter.py"]
    }
  }
}

GPT Researcher 等支持 MCP 的框架同理，配置好 command 指向脚本即可，传输协议细节 FastMCP 全部代劳。

【截图占位：MCP Inspector 调用 search_semiconductor_reports 的界面，可看到工具参数表单和返回的 content/source】

八、踩坑实录：这 6 个坑我替你踩过了

除了 MCP 的两坑，再把整个项目里其他值得记录的坑一次性放送。每个都是真实发生、排查过、验证过解法的。

#	现象	根因	解法
1	FAISS 和 PyTorch 一起 import 就崩溃	两个库各自捆绑了 OpenMP 运行时，双重加载冲突	启动前设 `KMP_DUPLICATE_LIB_OK=TRUE`
2	BGE-M3 CPU 推理偶发段错误	多线程内存竞争	`torch.set_num_threads(4)` 限线程 + encode 失败自动重试 3 次
3	BM25 对中文查询经常检索不中	代码用 `query.split()` 按空格分词，中文句子整句成一个 token，倒排索引根本匹配不上	靠查询扩展把查询改写出带空格/英文的变体兜底；根治要换 jieba 分词（见迭代方向）
4	批量跑题时 API 报限流错误	qwen-turbo 限流 500 次/分钟、50 万 token/分钟，并发一高就触顶	并发参数可配置，批量评测时压到 4 并发
5	评测结果"惨不忍睹"，人工抽查发现不少答案其实是对的	评测脚本的严格匹配太死板：标准答案 `HKD63`，系统答 `63港币`，判错；LLM 答案带 ```json 围栏没清洗就比对，判错	评测脚本本身也要 review：清洗 markdown 围栏、加单位换算和同义匹配
6	Streamlit 界面每次点击按钮都卡十几秒	每次交互 rerun 都重新加载 2.2GB 的 BGE-M3 模型和全部索引	`@st.cache_resource` 缓存 Pipeline 对象，只在首次加载

第 5 条特别多说一句，是我这个项目里最有"元认知"价值的教训：当你对系统表现不满意时，先怀疑评测，再怀疑系统。评测器也是代码，也会有 bug；一个有缺陷的评测器会把你所有的迭代方向带偏。

九、局限与迭代方向

老实交代目前的不足，也算给想接着做的同学留个地图：

知识库只有一家公司。按公司名检索的设计天然支持多公司，扩容只需补充数据、重建索引。
BM25 分词是硬伤。空格分词对中文本质是失效的，目前靠查询扩展兜底。计划引入 jieba 分词，预计关键词路对中文长查询的召回会有明显改善。
表格数值的精确问答还有提升空间。研报的财务预测表是核心资产，值得做表格结构化序列化（把表格转成"指标-期间-数值"三元组入库），这也是原始开源框架里预留但我尚未启用的能力。
评测体系要自动化。目前已构建 35 题、7 题型的评测集并跑通了评测脚本，但比对逻辑（单位换算、同义匹配、围栏清洗）需要继续加固，让每次迭代都能拿数据说话。