RAG如何赋能金融研报的智能分析

可视化演示

动态演示RAG如何赋能金融研报的智能的核心机制，通过输入/处理/存储/输出四种模式展示关键概念与交互流程。

知识库构建：给研报安个家

小叶最先要解决的是“书找不到”的问题。研报堆在角落，PDF 里有的能复制文字、有的整页是扫描图，还有的连发布日期都丢了。老周说：“档案室第一步不是写报告，而是把每份研报登记入库——像图书管理员给新书盖馆藏章一样，每份研报要打上标题、机构、日期、板块标签，再按标签摆上书架。”小叶这才明白，检索再快，前提也得是资料有家可归。

这个“档案室”在 RAG 里叫知识库。构建金融研报知识库的第一步是格式解析：用 pdfplumber、PyMuPDF 这类工具把 PDF 的文字层抽取出来，遇到纯扫描件就动用 OCR（光学字符识别）把图像变成文字。接着做元数据管理，用 date、institution、industry 等字段描述每份研报，就像盖馆藏章。最后还有权限管理：投研报告可能来自付费数据源、内部研究或公开报告，不同角色只能访问对应书架。

工程上，元数据会写入 Elasticsearch 这类索引库，配合向量库一起使用，让系统能先按条件筛再按语义找，例如“只查 2024 年以后的券商电子行业研报”。知识库还需要增量更新：每天收盘后自动入库新报告，数据源掉线要能告警。权限模型通常对接企业统一登录系统，保证研究员能看付费报告，而普通客服看不到。

档案室安顿好后，小叶再面对陈总的问题，至少知道去哪儿找、按什么条件筛。但老周又提醒她：“档案室还有第二件事——研报太厚，怎样让人快速定位到具体段落？”

分块策略：把厚研报切成索引卡

老周从柜子里拿出一叠发票大小的索引卡：“档案室的第二步，是把厚研报裁成能单手翻阅的卡片，每张卡要保留页码和上下文提示。”他说，卡片太大，一次翻开的内容太多、检索不精准；卡片太小，单独一张卡就看不懂前后的逻辑。小叶记下了：好卡片的关键，是大小适中、边界整齐。

在 RAG 中，这个操作叫文本分块（chunking）。大模型有上下文窗口限制，不能把整本研报塞进去，所以要把研报拆成若干小段；每段独立编码成向量、独立参与检索。分块质量直接决定回答质量：切得好，“宁德时代毛利率提升”会被完整保留；切得差，一句话被拦腰斩断，检索到也没法用。

常见方案有三种：固定长度切分，按 token 或字符数切成等长片段，最省事但可能切断句子；语义分块，按段落、小标题、章节边界切，能保住完整语义，工程上常用标题层级做递归切分；重叠窗口，在两块之间保留 overlap 个字符的重叠，避免关键信息恰好落在切缝处。给定研报长度 L、块大小 chunk_size、重叠 overlap，需要的分块数量为：

分块之后，小叶的档案柜变成了一盒盒带编号的索引卡，查起来快多了。但新的问题来了：研报里的表格和图表怎么办？

表格与图表处理：数字不能睁眼瞎

陈总问“各家电池企业的毛利率对比”，小叶翻到一张大表格，密密麻麻的数字，直接复制出来却是一串乱码。老周说：“表格和图片是研报的'重武器'，不能当成普通文字处理，得像给菜谱配'用料表'一样，单独做一份结构化卡片，把用料名称、用量、单位都列清楚。”

研报的关键信息大量藏在表格和图表里：财务三张表、盈利预测、渗透率趋势图。PDF 解析出的表格常常丢失行列结构，变成一行行错位的字符；图表则完全无法被文本解析器读取。如果知识库只收正文，用户问“盈利预测表里 2024 年营收是多少”，系统就抓瞎了。

处理思路有三步：第一，用 pdfplumber 的 extract_table 提取表格，转成 Markdown 或 JSON 结构体，保留表头和行列关系；第二，图片型图表用图表识别模型（如 Table Transformer）还原其中的数字，再配合原图标题和章节结论，生成一段“看图说话”的摘要；第三，把“表格标题＋表格摘要＋关键数值”作为检索单位存放，同时保留单元格位置引用，例如“第 3 节第二行第三列”，方便后续逐字溯源。

图表被翻译成“数字卡片”后，小叶在回答估值问题时，终于能说出“根据第 12 页盈利预测表，目标价 85 元”这样的句子。老周满意地点点头：“现在，档案室里的资料终于'能读懂了'。”

向量检索：不仅是找关键词

有一天陈总问：“电动车卖得怎么样？”小叶用关键词搜“新能源汽车”，只搜到两篇。老周提醒她：“档案室真正厉害的地方是按意思找。'电动车'和'新能源汽车'字面上完全不同，但意思相近，靠逐字匹配就漏了。”小叶恍然大悟：找资料不能只看字面，还得看语义。

向量检索的核心是 Embedding——把文本映射成一组数字（向量），让语义相近的文本向量方向也相似。把查询向量与知识库中所有分块的向量逐一计算余弦相似度，就能把最相关的段落捞出来。向量检索讲究的不是“字长得一样”，而是“意思方向一致”。

余弦相似度的公式如下：

实际工程里只有向量检索还不够，常用混合检索：BM25 负责精确关键词匹配（比如“ROE 12.5%“这类数字），向量检索负责语义泛化（”电动车“命中”新能源汽车”），两者结果再经过重排序模型融合。加权得分公式为：

向量检索让小叶真正“懂”了问题背后的意思，陈总再问“电动车卖得好不好”，她也能立刻定位到新能源汽车研报。但老周又抛来一个新要求：“找到只是第一步，你写答案时，每个数字都得能说出处。”

答案溯源：每个数字都有来处

小叶第一次给陈总交报告，写“新能源汽车渗透率 35.6%”，陈总问：“哪份研报？哪一页？”小叶答不上来。老周说：“金融汇报的铁律是结论必须带出处，就像法庭上指证要有证据编号。没有页码的数字，在金融圈里等于没说。”

RAG 的溯源机制把“证据”与“回答”绑定在一起：检索阶段保留每个分块的 doc_id 和页码，相当于给每张索引卡盖上来源章；生成阶段让模型只在证据范围内作答，并在句子末尾输出引用角标，例如“[1]（来源：XX证券《新能源汽车行业深度报告》P12）”；最终答案后面附上完整来源列表，用户可以一键跳回原文位置。

为了让“不确定的答案”不被当成事实，系统还要计算置信度。设检索到 k 个证据片段，每个片段的相似度为 sim_i，则答案置信度定义为：

有了溯源和置信度，小叶的报告终于经得起陈总追问。老周又提醒她最后一个环节：“别忘了，档案室外面还有一位'门神'。”

合规与风控：档案室主任把门

研报库不是谁都能碰的。老周说，档案室主任老王负责把关：外人不许看未公开研报、客户姓名要打码、每次借阅都要登记。小叶这才意识到，金融 RAG 系统比普通问答系统多了一道“合规关卡”，没有这道门，技术再先进也上不了线。

金融合规最看重两件事：不准把错误信息给客户、不准泄露未公开信息。具体到 RAG 系统，就要加三重保险：第一层是准确性校验，把模型生成的答案与检索到的原文片段做一致性比对，防止模型自由发挥“编数字”；第二层是脱敏，检索结果中隐藏客户名称、内部评级等敏感字段；第三层是审计追踪，记录谁在何时问了什么、模型引用了哪些文档，做到全程留痕可追溯。

落地时，工程团队通常会在 LLM 输出后加一道“合规过滤器”：用正则匹配替换身份证号、手机号等个人隐私；用相似度判断输出是否偏离原文主旨；所有问答日志写入审计系统，按监管要求保留 180 天备查。监管强调的“适当性”也会写进提示词——系统不会对个人投资者给出“买/卖”等明确指令，而是给出证据和风险提示，把决定权还给客户。

有老王把关，小叶才能放心地把答案交给陈总。老周笑着说：“技术上的活你都会了，接下来咱看看这三间办公室怎么用你。”

应用落地：三间办公室的智能助理

培训结束后，小叶有三项任务：帮投研部整理“今天该看什么”、帮财富中心回答客户疑问、帮风控部门盯研报里的异常信号。她发现，同一个“档案室＋检索＋溯源”工作流，换不同的入口和话术，就是三个完全不同的产品。

投研问答是主力场景：研究员输入“梳理比亚迪近三年毛利率变化及原因”，系统从产业研报库中检索出相关段落，生成带页码引用的摘要，单次查询从人工翻报告的 30 分钟缩到 10 秒。智能投顾把产品 FAQ 和宏观研报结合，回答“养老基金新规对我有什么影响”这类问题，自动附带风险揭示。风控分析则检索负面舆情和财务预警研报，当发现“质押爆仓”“商誉减值”等信号时，匹配受影响个股的历史研报，生成风险提示单推送给交易员。

三者的 RAG 架构相似但护栏不同：投研强调召回率和溯源深链，希望答得全、引得准；投顾强调提示词里的风险揭示模板和脱敏规则，一句违规话术都不能出；风控强调时效性，当天新发的研报和新闻必须在小时内入库，同时要有监控告警。评估指标也各有所长：投研看重答案相关性，投顾看重合规通过率，风控看重预警准确率。

小叶的成长故事，正是 RAG 从“实验产品”走向“金融生产力”的过程——先有档案室（知识库），再会裁卡（分块）、读表（表格处理）、快速定位（检索）、写带引用的报告（生成与溯源），最后通过合规审查。这七块拼图组合起来，就是 RAG 赋能金融研报智能分析的完整答案。

AI 应用场景

Python 演示

# -*- coding: utf-8 -*-"""Mini RAG demo for financial research report analysis.""" import mathimport refrom collections import Counter REPORTS = {    "REPORT-2023-001": "2023年新能源汽车渗透率达35.6%, 全年销量突破900万辆. 头部车企毛利率持续改善, 电池成本下降是主要驱动.",    "REPORT-2023-002": "稳增长政策持续发力, 基建投资增速回升至8.2%. 消费复苏偏弱, 社零同比增长3.1%.",    "REPORT-2024-001": "2024年光模块行业受AI算力拉动, 出货量同比增长40%. 海外云厂商资本开支超预期.",    "REPORT-2024-002": "银行板块净息差收窄至1.54%, 资产质量整体稳定. 高股息策略仍具备配置价值."} def chunk_text(text, size=80, overlap=16):    chunks = []    start = 0    total = len(text)    while start < total:        end = min(start + size, total)        chunks.append({"text": text[start:end], "start": start, "end": end})        if end == total:            break        start = max(start + size - overlap, start + 1)    return chunks def tokenize(text):    return re.findall(r"[\u4e00-\u9fa5a-zA-Z0-9]+", text.lower()) def vectorize(text):    return dict(Counter(tokenize(text))) def cosine_sim(vec_a, vec_b):    common = set(vec_a.keys()) & set(vec_b.keys())    dot = sum(vec_a[w] * vec_b[w] for w in common)    norm_a = math.sqrt(sum(v * v for v in vec_a.values()))    norm_b = math.sqrt(sum(v * v for v in vec_b.values()))    if norm_a == 0 or norm_b == 0:        return 0.0    return dot / (norm_a * norm_b) def build_knowledge_base():    kb = []    for doc_id, text in REPORTS.items():        for chunk in chunk_text(text):            chunk["doc_id"] = doc_id            chunk["vec"] = vectorize(chunk["text"])            kb.append(chunk)    return kb def retrieve(kb, query, top_k=2):    qvec = vectorize(query)    scored = []    for c in kb:        score = cosine_sim(qvec, c["vec"])        scored.append({"score": score, "doc_id": c["doc_id"], "text": c["text"]})    scored.sort(key=lambda x: x["score"], reverse=True)    return scored[:top_k] def answer(kb, query, top_k=2):    hits = retrieve(kb, query, top_k)    if not hits or hits[0]["score"] < 0.1:        return "未在研报库中找到足够证据, 无法给出答案."    lines = []    for h in hits:        lines.append("[%s] (置信度: %.2f) %s" % (h["doc_id"], h["score"], h["text"]))    return "\n".join(lines) if __name__ == "__main__":    kb = build_knowledge_base()    print("研报知识库分块数量:", len(kb))    for q in ("新能源汽车渗透率是多少", "基建投资增速如何"):        print("问题:", q)        print(answer(kb, q))        print()

1. def chunk_text(text, size=80, overlap=16):把研报切成带重叠的片段，保证查任意段落都能搜到上下文。
2. return re.findall(r"[\u4e00-\u9fa5a-zA-Z0-9]+", text.lower())把中文和英文都切成词，作为最简“向量”的维度。
3. return dot / (norm_a * norm_b)余弦相似度核心运算，衡量两个词频向量方向是否一致。
4. chunk["doc_id"] = doc_id给每个分块打上研报编号，这是溯源的基础。
5. scored.sort(key=lambda x: x["score"], reverse=True)按相似度降序排列，取前 k 个作为证据。
6. if not hits or hits[0]["score"] < 0.1:分数过低的查询拒绝回答，模拟真实 RAG 的置信度门槛。

常见误区

面试题