简介:这份PPT资料聚焦证券知识库的构建与应用,面向金融科技从业者、大模型应用开发者及对智能投研感兴趣的技术人员,系统讲解如何将大模型与知识库结合,解决证券信息自动化管理与智能决策支持问题。资源包内含1个pptx文件,大小约14.69MB,以演示文稿形式呈现完整技术脉络。内容覆盖文档结构化解析、知识库构建与问答、应用案例及思考展望等模块,具体涉及跨页段落合并、无框线表格还原、扫描件签章覆盖文字识别、用户意图识别、文本切片策略、向量化与Embedding微调等关键环节,并给出BGE-M3模型微调后recall@10由62.7%提升至73.3%的实测数据。目前已有52人学习,适合希望掌握证券领域知识库落地路径、了解大模型与知识库协同范式的读者参考借鉴。
1. 证券知识库构建和应用:从散落文档到可检索问答的落地路径
券商研究所、财富管理部和投顾团队每天都在生产研报、公告解读、合规问答和产品说明书,这些内容散落在共享盘、邮件附件和 IM 聊天记录里。想查一条“某类资管产品适当性匹配要求”,往往要翻三四个系统。证券知识库构建和应用,本质是把这些非结构化文本变成可检索、可引用、可更新的结构化资产,再挂到大模型问答或内部搜索上。它适合有大量合规文档、研报沉淀,又想让投顾和客服快速拿到准确答案的团队。难点不在模型,而在文档切分、元数据设计和答案溯源——这三件事决定了知识库是“能用”还是“看着热闹”。
2. 证券文档的切分与元数据设计:为什么按段落切会翻车
证券类文档和通用百科不一样。一份研报里有标题、摘要、正文、图表说明、免责声明;一份产品合同里有条款编号、嵌套子条款、附件表格。如果直接按固定字数切,很容易把“适当性匹配意见”和“风险揭示”切到两个块里,检索时召回一半、答案残缺。我一般会先做文档类型识别,再按语义结构切。
2.1 先分清四类文档,再决定切分粒度
证券知识库常见的文档可以粗分为四类,每类的切分策略不同:
| 文档类型 | 典型来源 | 切分粒度 | 关键元数据 |
|---|---|---|---|
| 研报/策略报告 | 研究所 PDF | 按章节 + 段落,保留标题层级 | 行业、评级、发布日期、分析师 |
| 产品合同/招募书 | 资管、基金 | 按条款编号切,条款内再按段 | 产品代码、风险等级、条款号 |
| 合规问答/制度 | 内部制度库 | 一问一答为一块 | 制度名称、生效日期、适用范围 |
| 公告/新闻 | 交易所、资讯 | 按事件切,标题+正文合并 | 证券代码、公告类型、日期 |
这张表不是拍脑袋来的。研报如果按 500 字硬切,图表说明和正文混在一起,检索“某行业毛利率变化”时会把免责声明也召回。产品合同如果按段落切,一条包含多个子项的适当性条款会被拆散,用户问“R3 产品能不能卖给保守型客户”时,答案可能只召回半句。
2.2 用 Python 做结构化切分的最小实现
下面这段代码演示如何按标题层级和条款编号切分,并给每个块打上元数据。实际项目中我会把split_by_structure替换成更细的规则,但骨架不变。
import re from dataclasses import dataclass, field @dataclass class Chunk: text: str metadata: dict = field(default_factory=dict) def split_by_structure(raw_text: str, doc_type: str, base_meta: dict) -> list: """ 按文档类型做结构化切分。 doc_type: 'research' | 'contract' | 'qa' | 'announcement' base_meta: 文档级元数据,如 {'doc_id': 'R001', 'date': '2025-01-01'} """ chunks = [] if doc_type == 'contract': # 按“第X条”或“X.X”条款编号切 pattern = r'(第[一二三四五六七八九十百]+条|\d+\.\d+)' parts = re.split(pattern, raw_text) # re.split 会把分隔符单独保留,需要两两合并 for i in range(1, len(parts), 2): clause_no = parts[i] clause_text = parts[i+1] if i+1 < len(parts) else '' if len(clause_text.strip()) < 20: continue meta = {**base_meta, 'clause': clause_no, 'type': 'contract'} chunks.append(Chunk(text=f"{clause_no} {clause_text.strip()}", metadata=meta)) elif doc_type == 'research': # 按 Markdown 标题或 PDF 提取后的标题行切 sections = re.split(r'\n(?=#{1,3}\s)', raw_text) for sec in sections: if len(sec.strip()) < 50: continue title_match = re.match(r'#{1,3}\s*(.+)', sec) title = title_match.group(1) if title_match else '正文' meta = {**base_meta, 'section': title, 'type': 'research'} chunks.append(Chunk(text=sec.strip(), metadata=meta)) else: # 问答和公告按空行或固定分隔符切 for part in re.split(r'\n{2,}', raw_text): if len(part.strip()) < 30: continue chunks.append(Chunk(text=part.strip(), metadata={**base_meta, 'type': doc_type})) return chunks逻辑说明:split_by_structure先根据doc_type走不同分支。合同类用正则匹配“第X条”或“X.X”编号,把每个条款单独成块,并保留条款号到元数据;研报类按 Markdown 标题切,保留章节名;问答和公告按空行切。每个块都继承文档级元数据,再附加自身结构信息。
参数说明:base_meta至少包含doc_id、date、source,方便后续按来源过滤。len(clause_text.strip()) < 20这个阈值用来过滤空条款或只有编号的行,实际可以调到 30。研报的len(sec.strip()) < 50是为了避免把只有标题没有内容的块入库。
提示:切分后的块不要直接丢进向量库,先落一张关系表或 JSONL,人工抽检 20 条,确认没有把关键条款切碎再继续。
2.3 元数据字段怎么定:检索过滤比向量相似度更管用
很多团队一上来就调 embedding 模型,结果用户问“2024 年新能源行业研报里关于补贴退坡的说法”,向量检索把 2022 年的旧研报也召回。问题不在模型,在元数据没建好。我一般要求每个块至少带这几个字段:doc_type、date、industry、risk_level、source、clause(合同类)。检索时先用date >= '2024-01-01'和industry = '新能源'做过滤,再在候选集里做向量排序。这样召回准确率会明显提升,而且答案可以标注“来自 2024 年某研报第 3 章”,投顾才敢用。
3. 向量化与检索层搭建:embedding 选型、索引参数和混合检索
切分和元数据做完,下一步是把文本块变成向量,并搭一个能按元数据过滤的检索层。证券知识库对“查得准”的要求高于“查得全”,所以混合检索(关键词 + 向量)比纯向量更稳。下面按选型、建库、查询三步走。
3.1 embedding 模型怎么选:中文金融语料上的实际表现
通用中文 embedding 在证券文本上会遇到两个问题:一是专业术语多,比如“摊余成本法”“侧袋机制”,通用模型可能把它们映射到相近但不准确的向量;二是长条款多,超过模型最大长度会被截断。我一般会优先选支持 512 或 1024 token 的中文模型,并在自己的语料上做一次小规模评测:拿 50 个真实问题,看 Top5 召回里有没有正确答案。如果通用模型召回率低于 70%,就考虑用金融语料微调过的版本,或者至少加一层关键词检索兜底。
选型时看三个参数:最大输入长度、向量维度、是否支持中文金融领域。维度不是越高越好,1024 维在百万级块上检索延迟已经明显,768 维往往够用。如果团队没有 GPU,优先选能 CPU 推理的模型,批量编码时用batch_size=32左右,再大内存容易爆。
3.2 用 FAISS 建索引并挂元数据过滤
下面代码演示用 FAISS 建向量索引,并把元数据存在 SQLite 里,查询时先过滤再检索。实际生产可以用 Milvus 或 Qdrant,但本地验证用 FAISS + SQLite 最快。
import sqlite3 import numpy as np import faiss # 假设 embeddings 是 list[np.ndarray],chunks 是 list[Chunk] def build_index(embeddings: list, chunks: list, db_path='kb.db', index_path='kb.faiss'): dim = embeddings[0].shape[0] index = faiss.IndexFlatIP(dim) # 内积,需先归一化 mat = np.vstack(embeddings).astype('float32') faiss.normalize_L2(mat) index.add(mat) faiss.write_index(index, index_path) conn = sqlite3.connect(db_path) conn.execute('''CREATE TABLE IF NOT EXISTS chunks (id INTEGER PRIMARY KEY, text TEXT, doc_type TEXT, date TEXT, industry TEXT, risk_level TEXT, clause TEXT)''') for i, c in enumerate(chunks): m = c.metadata conn.execute('INSERT INTO chunks VALUES (?,?,?,?,?,?,?)', (i, c.text, m.get('type'), m.get('date'), m.get('industry'), m.get('risk_level'), m.get('clause'))) conn.commit() conn.close() return index def search(query_vec, index, db_path, filters: dict, top_k=5): # 先按元数据过滤出候选 id conn = sqlite3.connect(db_path) where = ' AND '.join([f"{k} = ?" for k in filters]) cur = conn.execute(f'SELECT id FROM chunks WHERE {where}', list(filters.values())) candidate_ids = {row[0] for row in cur.fetchall()} conn.close() q = np.array([query_vec]).astype('float32') faiss.normalize_L2(q) scores, ids = index.search(q, top_k * 10) # 多取一些再过滤 results = [] for score, idx in zip(scores[0], ids[0]): if idx in candidate_ids: results.append((idx, float(score))) if len(results) >= top_k: break return results逻辑说明:build_index先把所有向量归一化,用内积索引等价于余弦相似度;元数据写入 SQLite,每个块一行。search先按filters查出候选 id 集合,再在 FAISS 返回的 Top50 里筛出属于候选集的块,取前top_k。这样既利用了向量语义匹配,又保证了元数据约束。
参数说明:top_k * 10是过度召回倍数,因为过滤会淘汰一部分,实际可以调到 20 倍。filters的键必须是 SQLite 表里存在的列,比如{'doc_type': 'contract', 'risk_level': 'R3'}。如果候选集很小,可以直接用暴力检索,不必走 FAISS。
注意:FAISS 的
IndexFlatIP在百万级向量上内存占用约 4GB(1024 维 float32),如果块数超过 500 万,建议换 IVF 或 HNSW 索引,并接受一定精度损失。
3.3 混合检索:关键词兜底和重排序
纯向量检索在证券场景下有两个典型翻车:一是用户问“某代码 600XXX 的公告”,向量模型对数字不敏感,可能召回其他代码;二是问“侧袋机制触发条件”,如果 embedding 没学好,会召回“侧袋估值”而不是“触发条件”。我一般会加一路 BM25 关键词检索,把两路结果合并后用一个轻量重排序模型(比如 cross-encoder)打分。BM25 可以用rank_bm25库,几行代码就能跑。合并时按0.6 * 向量分 + 0.4 * BM25 分加权,再取 Top5。重排序模型如果资源不够,可以省略,但关键词那一路必须保留,它是数字和专有名词的后悔药。
4. 问答层与答案溯源:怎么让投顾敢用生成的答案
检索层给出 Top5 块之后,问答层要做两件事:把块拼成 prompt 让大模型生成答案,以及把答案和来源块绑定。证券行业对合规要求高,答案必须能点回原文,否则投顾不敢直接引用。这一章讲 prompt 设计、引用格式和拒答策略。
4.1 prompt 模板:约束模型只基于检索块回答
我一般用下面这个模板,核心是“只允许使用给定材料”和“必须标注来源编号”。
PROMPT_TEMPLATE = """你是一名证券合规助手。请仅根据以下材料回答问题。 如果材料中没有答案,直接回答“根据现有资料无法确认”,不要编造。 材料: {context} 问题:{question} 要求: 1. 答案中每个关键结论后用 [编号] 标注来源,如 [1]。 2. 不要引用材料之外的知识。 3. 如果涉及风险等级、适当性匹配,必须原文引用相关条款。 """逻辑说明:context是把 Top5 块按[1] 文本... [2] 文本...拼起来的字符串。要求模型标注来源编号,是为了后续把编号映射回块 id 和原文位置。不要引用材料之外的知识这句必须写,否则模型会用自己的预训练知识补全,在证券场景下这是合规风险。
参数说明:temperature设 0.1 到 0.3,越低越稳定。max_tokens根据答案长度设 512 到 1024。如果模型支持stop参数,可以加stop=["\n\n问题"]防止它自问自答。
4.2 答案溯源:把编号映射回文档和条款
生成答案后,用正则提取[1]、[2]这样的编号,再查回块 id 和元数据,拼成引用列表返回给前端。前端展示时,用户点编号就能看到原文片段和文档名。这一步看起来简单,但很多团队漏掉,导致答案和来源对不上。我一般会在返回结构里加citations字段:
import re def attach_citations(answer: str, retrieved_chunks: list) -> dict: cited_ids = set(int(x) for x in re.findall(r'\[(\d+)\]', answer)) citations = [] for cid in cited_ids: if 1 <= cid <= len(retrieved_chunks): chunk = retrieved_chunks[cid - 1] citations.append({ 'index': cid, 'text': chunk.text[:200], 'metadata': chunk.metadata }) return {'answer': answer, 'citations': citations}逻辑说明:re.findall提取答案里的编号,retrieved_chunks是传给模型的块列表,编号从 1 开始对应列表下标 0。text[:200]只截取前 200 字做预览,完整原文通过元数据里的doc_id和clause去文档库取。
参数说明:如果模型输出的编号格式不稳定,可以在 prompt 里强制要求[1]这种方括号数字格式,并在后处理时兼容[1,2]这种写法,用re.findall(r'\[(\d+)\]', answer)也能提取出来。
4.3 拒答策略:什么时候必须说“不知道”
证券知识库里最危险的不是答错,而是答得太像真的。我一般设三条拒答规则:检索 Top1 相似度低于阈值(比如 0.65)时拒答;问题里包含“预测”“保证收益”“推荐买入”等词时拒答并提示合规;检索块里没有明确条款支持时,要求模型输出“根据现有资料无法确认”。这三条规则写进 prompt 和后处理逻辑,能挡掉大部分合规风险。阈值不要设太高,否则正常问题也拒答;0.6 到 0.7 之间根据实际语料调。
5. 避坑与排查:证券知识库落地时最容易翻车的五件事
这一章记录我在实际项目里踩过的坑,每条按现象、原因、解决写。有些坑在通用 RAG 里也会遇到,但证券场景下后果更严重。
5.1 现象:用户问“R3 产品能不能卖给保守型客户”,答案引用了旧版适当性办法
原因:元数据里没有effective_date字段,检索时把已废止的制度也召回了。证券制度更新频繁,旧版和新版可能同时存在知识库里。
解决:每个制度块必须带effective_date和status(有效/废止)。检索时默认加status = '有效',除非用户明确要查历史版本。如果制度有修订,旧版块不要删,把status改成废止,并在答案里提示“该条款已于某日期废止”。
5.2 现象:研报里的表格数据检索不到,用户问“某公司毛利率”时召回的是正文描述
原因:PDF 提取时表格被转成纯文本,行列关系丢失,切分后表格数据变成一串数字,embedding 无法理解。很多 PDF 解析库对合并单元格和跨页表格支持不好。
解决:表格单独处理,用pdfplumber或camelot提取成结构化数据,存成“表名 + 列名 + 行值”的文本块,比如“表:某公司毛利率(2023):营业收入 100 亿,毛利率 35%”。这样 embedding 能捕捉到“毛利率”和数值的关系。如果表格太复杂,至少把表头和第一列拼进每个数据块。
5.3 现象:问答响应慢,用户等 5 秒以上
原因:每次查询都重新编码 query 向量,且 FAISS 索引没做量化,百万级向量检索耗时高。另外如果重排序模型跑在 CPU 上,也会拖慢。
解决:query 编码用缓存,相同问题直接返回;FAISS 换IndexIVFFlat,nlist设为sqrt(N)左右,查询时nprobe设 10 到 20;重排序模型如果非必须,先去掉,用加权分数代替。实测这些改完,P99 能从 5 秒降到 1 秒以内。
5.4 现象:答案里出现“根据相关法规”但没给具体条款
原因:prompt 里虽然要求标注来源,但模型有时会偷懒,用模糊表述。后处理也没有强制校验。
解决:在后处理里检查答案是否包含[数字]格式的引用,如果没有,就重新生成或直接返回检索块原文。另外 prompt 里加一句“禁止使用‘相关法规’‘有关规定’等模糊表述,必须引用具体条款编号”。这条规则对合规场景很关键。
5.5 现象:知识库更新后,旧答案还能被检索到
原因:向量索引和元数据库没有同步更新,或者更新时只加了新块没删旧块。证券公告和研报每天新增,如果更新流程是手动的,很容易漏。
解决:建一个增量更新脚本,每天定时跑:先解析新文档,切分后写入 SQLite,再重新编码新增块并index.add()。删除旧版本时,从 SQLite 删行,FAISS 不支持按 id 删除,所以要么定期重建索引,要么用支持删除的向量库(如 Milvus)。我一般每周重建一次全量索引,增量部分先放内存索引,查询时合并结果。
6. 进阶技巧:用问题日志反哺知识库,把召回率从 70% 拉到 90%
知识库上线不是终点。我习惯在问答接口里记录每次查询的问题、召回块 id、用户是否点击了引用、是否追问。这些日志是优化知识库的金矿。具体做法:每周导出问题日志,人工标注 100 条,看哪些问题召回失败。失败原因通常三类:切分太碎、元数据缺失、embedding 不匹配。针对第一类,调整切分规则;第二类补字段;第三类把失败问题作为难例,微调 embedding 或加关键词规则。
下面是一个简单的日志表结构和分析脚本,用来找出“高频率但低点击”的问题。
CREATE TABLE qa_log ( id INTEGER PRIMARY KEY, question TEXT, retrieved_ids TEXT, -- 逗号分隔的块 id cited_ids TEXT, -- 用户点击的引用 id created_at TEXT );import sqlite3 from collections import Counter def find_low_click_questions(db_path, min_count=5): conn = sqlite3.connect(db_path) rows = conn.execute('SELECT question, retrieved_ids, cited_ids FROM qa_log').fetchall() conn.close() counter = Counter() for q, ret, cited in rows: if not cited or cited.strip() == '': counter[q] += 1 return [q for q, c in counter.items() if c >= min_count]逻辑说明:find_low_click_questions统计那些被问过多次但用户从未点击引用的 question,这些大概率是召回不相关或答案没帮助。拿到列表后,人工看这些问题的检索块,判断是切分问题还是模型问题。
参数说明:min_count=5是阈值,低于 5 次的问题样本太少,不急着优化。cited_ids为空表示用户没点任何引用,可能是答案直接满足了,也可能是答案没用。需要结合追问率一起看。
我自己的习惯是每两周跑一次这个脚本,把 Top20 低点击问题拿出来,逐条看检索块。有一次发现“某类产品赎回规则”连续三周排第一,查下来是合同切分时把“赎回”条款和“申购”条款切到同一个块,检索时总是召回申购内容。把切分规则改成按条款标题切后,这个问题就消失了。知识库优化没有一劳永逸,但问题日志能让每次调整都有依据。希望帮到你。
本文还有配套的精品资源,点击获取