智能投顾 AgentRAG 如何让大模型读懂财报、研报和行情数据一、深度引言与场景痛点去年尝试用 GPT-4 直接做投顾问答时结果惨不忍睹。问宁德时代 2023 年 Q3 的毛利率变化趋势GPT-4 可以给你一个措辞流畅但对数字全是编造的回答——它把 2022 年的数据和 2023 年的趋势混杂在一起还自信满满地说毛利率从 21.3% 提升到 25.8%实际上 Q3 毛利率才 22.4%。这是 LLM 做金融场景的致命问题金融对准确性的要求是二进制的——数字对就是对错就是错没有差不多。用户追问这数据来源是哪里GPT-4 要么给不出 source要么给出一个不存在的链接幻觉。更深层的痛点是金融数据的异构性。一份完整的基本面分析需要三类数据结构化数据财报表格里的数字如营收、利润、ROE、半结构化数据研报里的分析结论和图表、非结构化数据新闻、公告、社交媒体情绪。RAG 要同时处理这三类数据并能交叉引用这个难度远超一般的文档问答。还有一个时效性问题。财报季发布后新旧数据并存——同一个指标在 Q2 报和 Q3 报里是不同的值。RAG 需要知道用户问的是最新值还是历史趋势而不是把所有版本的财报都混在一起回复。二、底层机制与原理深度剖析智能投顾 RAG 的架构需要在标准文档 RAG 之上叠加三层金融特有的处理能力三个独特的处理层表格结构化提取——财报里的数字不是普通文本必须保持行列关系才能做同比环比计算数据交叉验证——同一个指标可能在多个来源有不同的值财报 vs 研报引用需要标注差异而非隐去意图路由——茅台现在多少钱是查行情茅台值得投资吗是问分析检索策略完全不同。三、生产级代码实现import asyncio import json import logging import re from dataclasses import dataclass from datetime import datetime, timedelta from enum import Enum from typing import Any, Optional import numpy as np import pandas as pd from pydantic import BaseModel, Field, ValidationError, field_validator from sentence_transformers import SentenceTransformer logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) # ── 金融数据模型 ───────────────────────────────────────── class DataSource(str, Enum): FINANCIAL_REPORT financial_report RESEARCH_NOTE research_note MARKET_DATA market_data NEWS news ANNOUNCEMENT announcement class FinancialChunk(BaseModel): 金融数据块带有元数据标记 chunk_id: str text: str source_type: DataSource stock_code: str stock_name: str report_date: str # 财报日期 YYYY-MM-DD metric_name: str # 财务指标名如果是表格数据 metric_value: Optional[float] None unit: str # 亿/万元/% embedding: Optional[list[float]] None confidence: float 1.0 # 数据可信度 class FinancialQuery(BaseModel): 用户提问的意图解析结果 raw_query: str intent: str # data_lookup / analysis / comparison / market stock_codes: list[str] Field(default_factorylist) metrics: list[str] Field(default_factorylist) time_range: Optional[str] None require_source: bool True # ── 财报表格解析 ───────────────────────────────────────── class FinancialReportParser: 财报解析器把 PDF/Excel 财报转为结构化数据 staticmethod async def parse_excel(file_path: str) - list[FinancialChunk]: 解析财报 Excel 中的表格数据 try: dfs await asyncio.to_thread(pd.read_excel, file_path, sheet_nameNone) except Exception as e: logger.error(fExcel 解析失败 {file_path}: {e}) raise ValueError(f财报文件解析失败: {e}) chunks [] for sheet_name, df in dfs.items(): if df.empty: continue # 从文件路径提取股票代码和日期 stock_code FinancialReportParser._extract_stock_code(file_path) report_date FinancialReportParser._extract_date(file_path) for _, row in df.iterrows(): # 逐行构建金融数据块 metric_name str(row.iloc[0]) if len(row) 0 else metric_value None for col_idx in range(1, min(len(row), 4)): val row.iloc[col_idx] if pd.notna(val) and isinstance(val, (int, float)): metric_value float(val) break if not metric_name or metric_name nan: continue text_desc f{stock_code} {report_date} {metric_name}: {metric_value} chunks.append(FinancialChunk( chunk_idffin-{stock_code}-{report_date}-{metric_name[:10]}, texttext_desc, source_typeDataSource.FINANCIAL_REPORT, stock_codestock_code, report_datereport_date, metric_namemetric_name, metric_valuemetric_value, unit万元, )) logger.info(f解析财报 {file_path}: {len(chunks)} 条指标) return chunks staticmethod def _extract_stock_code(path: str) - str: match re.search(r(\d{6}), path) return match.group(1) if match else UNKNOWN staticmethod def _extract_date(path: str) - str: match re.search(r(\d{4})[_-]?(\d{2})[_-]?(\d{2}), path) if match: return f{match.group(1)}-{match.group(2)}-{match.group(3)} return datetime.now().strftime(%Y-%m-%d) # ── 数据交叉验证 ───────────────────────────────────────── class DataValidator: 金融数据交叉验证器 def __init__(self, tolerance: float 0.05): self.tolerance tolerance # 5% 容差 def validate(self, chunks: list[FinancialChunk]) - list[dict]: 检测同一指标在不同来源中的数值差异 metric_groups: dict[str, list[FinancialChunk]] {} for c in chunks: if c.metric_name and c.metric_value is not None: key f{c.stock_code}|{c.metric_name}|{c.report_date[:7]} metric_groups.setdefault(key, []).append(c) discrepancies [] for key, group in metric_groups.items(): if len(group) 2: continue values [c.metric_value for c in group if c.metric_value is not None] if len(values) 2: continue v_min, v_max min(values), max(values) if v_min 0: continue deviation abs(v_max - v_min) / abs(v_min) if deviation self.tolerance: discrepancies.append({ key: key, values: [ {source: c.chunk_id, value: c.metric_value, source_type: c.source_type.value} for c in group ], deviation: round(deviation * 100, 1), severity: high if deviation 0.2 else medium, }) logger.warning( f数据不一致 {key}: 差异 {deviation*100:.1f}%, f值范围 [{v_min}, {v_max}] ) return discrepancies # ── 智能投顾检索引擎 ───────────────────────────────────── class InvestmentAdvisorRAG: 智能投顾 RAG 引擎 def __init__(self): self.encoder SentenceTransformer(BAAI/bge-large-zh-v1.5) self.chunks: list[FinancialChunk] [] self.validator DataValidator() async def index(self, chunks: list[FinancialChunk]): 索引金融数据块 texts [c.text for c in chunks] if not texts: logger.warning(无数据可索引) return try: embeddings await asyncio.to_thread( self.encoder.encode, texts, normalize_embeddingsTrue ) except RuntimeError as e: logger.error(fEmbedding 编码失败: {e}) raise for chunk, emb in zip(chunks, embeddings): chunk.embedding emb.tolist() self.chunks.extend(chunks) logger.info(f索引完成: {len(chunks)} 条数据) async def search( self, query: FinancialQuery, top_k: int 10 ) - list[dict]: 检索相关金融数据 if not self.chunks: return [] try: query_emb await asyncio.to_thread( self.encoder.encode, [query.raw_query], normalize_embeddingsTrue ) except RuntimeError as e: logger.error(fQuery 编码失败: {e}) raise # 向量相似度 股票代码过滤 时间过滤 scores [] for i, chunk in enumerate(self.chunks): if query.stock_codes and chunk.stock_code not in query.stock_codes: continue if query.time_range and chunk.report_date query.time_range: continue chunk_emb np.array(chunk.embedding) similarity float(np.dot(query_emb[0], chunk_emb)) # 结构化数据有数字的给予额外加权 if chunk.metric_value is not None: similarity * 1.1 scores.append((i, similarity)) scores.sort(keylambda x: x[1], reverseTrue) top_indices [idx for idx, _ in scores[:top_k]] results [] for idx in top_indices: chunk self.chunks[idx] results.append({ chunk_id: chunk.chunk_id, text: chunk.text, stock_code: chunk.stock_code, metric_name: chunk.metric_name, metric_value: chunk.metric_value, report_date: chunk.report_date, source_type: chunk.source_type.value, score: scores[[s[0] for s in scores].index(idx)][1], }) # 交叉验证 relevant_chunks [self.chunks[idx] for idx in top_indices] discrepancies self.validator.validate(relevant_chunks) return { results: results, data_discrepancies: discrepancies, has_discrepancy: len(discrepancies) 0, } async def main(): parser FinancialReportParser() advisor InvestmentAdvisorRAG() try: # 模拟财报数据 chunks await parser.parse_excel(./samples/300750_2024Q3.xlsx) if not chunks: # 如果文件不存在构造模拟数据 chunks [ FinancialChunk( chunk_idfin-300750-2024Q3-revenue, text宁德时代 2024Q3 营业收入 922.78亿元, source_typeDataSource.FINANCIAL_REPORT, stock_code300750, stock_name宁德时代, report_date2024-09-30, metric_name营业收入, metric_value922.78, unit亿元, ), FinancialChunk( chunk_idfin-300750-2024Q3-gross_margin, text宁德时代 2024Q3 毛利率 22.4%, source_typeDataSource.FINANCIAL_REPORT, stock_code300750, stock_name宁德时代, report_date2024-09-30, metric_name毛利率, metric_value22.4, unit%, ), FinancialChunk( chunk_idresearch-300750-2024Q3-gm_note, text研究机构测算宁德时代 2024Q3 毛利率约 22.8%含非经常性损益调整, source_typeDataSource.RESEARCH_NOTE, stock_code300750, stock_name宁德时代, report_date2024-10-15, metric_name毛利率, metric_value22.8, unit%, confidence0.85, ), ] await advisor.index(chunks) query FinancialQuery( raw_query宁德时代2024年三季度毛利率是多少和之前比有什么趋势, intentdata_lookup, stock_codes[300750], metrics[毛利率], time_range2024-07-01, ) result await advisor.search(query, top_k5) logger.info(f检索到 {len(result[results])} 条数据) for i, r in enumerate(result[results]): logger.info(f#{i1} {r[metric_name]}: {r[metric_value]}{r.get(unit,)} (score{r[score]:.3f})) if result[has_discrepancy]: for d in result[data_discrepancies]: logger.warning(f⚠ 数据不一致: {d[key]}, 偏差{d[deviation]}%) except (ValueError, ValidationError) as e: logger.error(f处理失败: {e}) except Exception as e: logger.exception(f未预期错误: {e}) if __name__ __main__: asyncio.run(main())四、边界分析与架构权衡精确检索 vs 语义检索对于宁德时代 Q3 营收多少这种确定性问题RAG 的语义检索不如 SQL 精确查询。更好的方案是意图路由——识别出查数据类意图时走结构化查询直接查表格识别出问分析类意图时才走向量检索 LLM 生成。模型搞混这两种意图是金融场景下最大的用户体验杀手。数据时效性冲突Q2 报和 Q3 报同时索引在向量库里用户问毛利率时应该返回 Q3 的。单纯靠相似度排序不够两份报告可能分数接近需要叠加时间衰减因子——越新的数据权重越高但保留旧数据做同比/环比时能被检索到。幻觉的防御层次金融场景不允许幻觉。除了常规的 source 标注外建议加两层防御一是 rule-based 的数字提取校验如果 LLM 输出的数字在源数据中找不到直接拦截二是回复中的数字全部用[来源: xxx]标注让用户能溯源验证。合规风险在中国提供投资建议需要牌照。Agent 生成的内容必须在末尾加免责声明且不能出现推荐买入强烈建议持有等引导性表述——可以把 LLM 的 output 做一个合规检测 filter。五、总结RAG 做智能投顾的关键不是检索技术本身有多高级而是让模型只说它从数据里读到的东西不编造它不知道的东西。三层设计表格解析保证结构化数据的准确性、交叉验证暴露数据源之间的差异不给用户虚假一致性、意图路由让精确查询和语义分析各走各路。金融场景没有什么差不多的空间数字对错了用户信任就没有了——而信任是投顾产品的唯一资产。
