AI赋能金融行业的工程复盘智能研报系统的RAG架构踩坑全记录一、项目背景从全文喂给LLM到RAG的必然选择2025年Q1受某券商委托开发一个智能研报摘要系统。需求是输入一家上市公司名称系统自动检索近3年的相关研报用通俗语言输出核心要点和投资逻辑。初版方案最直接把相关研报的全部文本拼接起来一次性发给GPT-4。问题很明显——一份研报PDF平均15页约2万字撮合3年数据轻松突破15万字符。即使GPT-4 Turbo的128K上下文窗口能装下长上下文中的中间内容遗忘问题——Lost-in-the-Middle效应——导致关键信息提取完整率仅58%。转向RAG架构势在必行。但金融场景的RAG有独特挑战数据高度结构化表格数据占比30%术语专业夏普比率、杜邦分析时效性敏感季度财报的窗口期不超过72小时。二、金融场景RAG的五个特殊挑战挑战一表格数据的向量化困境。研报中的财务报表三张表是结构化表格常规Embedding处理效果很差。营业收入2024Q3: 23.5亿被分割后失去表格上下文检索时几乎不可被匹配。解决方案提取表格后不参与语义切分而是存储为TableMetadata对象在检索阶段通过SQL联合查询。表格的检索走结构化查询文本的检索走向量检索。挑战二行业术语的语义对齐。杜邦分析这个术语在不同语境下指代不同东西——有时指分析框架有时指具体指标。解决建立行业术语向量索引用户查询时先做术语扩展。ROE下降原因→净资产收益率下降原因 OR 杜邦分析显示。术语扩展不是盲目的synonym替换而是基于向量相似度业务规则的组合。挑战三时效性的多层控制。一份2022年的年报在分析当下时应该权重降低但在做历史趋势对比时又很重要。解决不只在检索时降权而是在metadata中加入report_date字段由LLM在生成时根据上下文判断是否参考。时效性过滤规则对于当前估值类问题只检索6个月内的数据对于历史趋势类问题全时段开放。挑战四幻觉在金融场景的代价极高。建议买入和建议卖出一字之差在金融场景是无法接受的错误。解决方案在LLM输出后加入事实核查层——要求生成的每条数据结论附带引用来源source chunk。同时用正则匹配检测关键数据的变化如数字、方向性词汇与source对比。挑战五长文档的语义丢失。一份研报的开篇行业概览和结尾风险提示在语义上相差很远但被切割在不同chunk中后检索行业风险时只能命中结尾部分。解决引入父文档引用——每个chunk保存其所在的父文档ID和位置序号检索时不仅返回匹配chunk还返回其前后各1个chunk作为上下文补充。三、关键实现混合检索表格处理from typing import List, Dict, Optional from pymilvus import Collection import numpy as np class FinancialRAGPipeline: def __init__(self, vector_store: Collection, bm25_index, table_store, # PostgreSQL存储表格 term_expander): self.vector_store vector_store self.bm25 bm25_index self.table_store table_store self.term_expander term_expander async def search(self, query: str, top_k: int 10) - List[Document]: # 1. 术语扩展 expanded_terms await self.term_expander.expand(query) # 2. 并行检索向量 关键词 结构化 vector_results await self._vector_search(query, top_k * 2) keyword_results self._bm25_search(expanded_terms, top_k) table_results await self._table_search(expanded_terms, top_k) # 3. 融合RRF (Reciprocal Rank Fusion) fused self._reciprocal_rank_fusion([ vector_results, keyword_results, table_results, ]) # 4. 重排序——金融专用Reranker reranked await self._financial_reranker.rerank(query, fused) # 5. 添加父文档上下文 enriched self._add_parent_context(reranked) return enriched[:top_k] def _reciprocal_rank_fusion( self, result_lists: List[List[Document]], k: int 60 ) - List[Document]: scores {} for results in result_lists: for rank, doc in enumerate(results): doc_id doc.metadata[doc_id] scores[doc_id] scores.get(doc_id, 0) 1.0 / (k rank 1) # 排序并返回 sorted_docs sorted(scores.items(), keylambda x: x[1], reverseTrue) return [self._get_doc(doc_id) for doc_id, _ in sorted_docs] async def _table_search(self, query: str, top_k: int) - List[Document]: 结构化表格检索——用SQL而非向量 # 提取查询中的实体和指标 entities await self._extract_financial_entities(query) sql_parts [] params [] if entities.get(company): sql_parts.append(company_name ILIKE %s) params.append(f%{entities[company]}%) if entities.get(metric): sql_parts.append(metric_name ILIKE %s) params.append(f%{entities[metric]}%) if entities.get(period): sql_parts.append(report_period %s) params.append(entities[period]) where_clause AND .join(sql_parts) if sql_parts else 11 rows await self.table_store.query( fSELECT * FROM financial_tables WHERE {where_clause} LIMIT %s, params [top_k] ) return [self._table_row_to_doc(row) for row in rows]四、效果评估与边界基于100条测试查询的评估结果指标基础方案(全文)RAG方案混合检索RAG信息完整率58%76%89%数据准确率71%84%93%平均检索延迟0ms(无检索)1.8s2.3s每次查询成本$0.15$0.06$0.08RAG的成本优势全文方案每次塞入15万字符的prompttoken消耗巨大。RAG方案精选top 10 chunk约8000字符显著降低成本。额外开销混合检索的2.3秒延迟对于实时问答偏慢。优化方向embedding缓存重复查询直接返回已计算的向量 检索结果缓存同义查询返回相同结果。Reranker的ROI重排序模型占用约15%的延迟和10%的成本但贡献了13个百分点的信息完整率提升。在金融场景的高准确性要求下这个ROI是正面的。五、总结金融场景的RAG落地核心经验表格数据不能走常规语义检索需要结构化SQL查询作为补充术语扩展是金融领域的必备步骤——ROE有6种不同的中文表达时效性过滤规则必须按问题类型分级不能一刀切每条结论必须有source引用——这是金融合规的基本要求混合检索向量BM25结构化的融合策略RRF简单有效项目的错误一开始低估了表格处理的难度。PDF中财务报表的跨页合并、合并单元格的语义识别、数字格式的标准化——这些脏活占了总开发时间的40%。如果重来一次会优先调研专业的表格提取库如Camelot、Tabula而不是自研解析器。