智能投顾 Agent：RAG 如何让大模型读懂财报、研报和行情数据

一、深度引言与场景痛点

去年尝试用 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(level=logging.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_factory=list)
    metrics: list[str] = Field(default_factory=list)
    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_name=None)
        except Exception as e:
            logger.error(f"Excel 解析失败 {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_id=f"fin-{stock_code}-{report_date}-{metric_name[:10]}",
                    text=text_desc,
                    source_type=DataSource.FINANCIAL_REPORT,
                    stock_code=stock_code,
                    report_date=report_date,
                    metric_name=metric_name,
                    metric_value=metric_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_embeddings=True
            )
        except RuntimeError as e:
            logger.error(f"Embedding 编码失败: {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_embeddings=True
            )
        except RuntimeError as e:
            logger.error(f"Query 编码失败: {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(key=lambda x: x[1], reverse=True)
        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_id="fin-300750-2024Q3-revenue",
                    text="宁德时代 2024Q3 营业收入 922.78亿元",
                    source_type=DataSource.FINANCIAL_REPORT,
                    stock_code="300750", stock_name="宁德时代",
                    report_date="2024-09-30",
                    metric_name="营业收入", metric_value=922.78, unit="亿元",
                ),
                FinancialChunk(
                    chunk_id="fin-300750-2024Q3-gross_margin",
                    text="宁德时代 2024Q3 毛利率 22.4%",
                    source_type=DataSource.FINANCIAL_REPORT,
                    stock_code="300750", stock_name="宁德时代",
                    report_date="2024-09-30",
                    metric_name="毛利率", metric_value=22.4, unit="%",
                ),
                FinancialChunk(
                    chunk_id="research-300750-2024Q3-gm_note",
                    text="研究机构测算宁德时代 2024Q3 毛利率约 22.8%（含非经常性损益调整）",
                    source_type=DataSource.RESEARCH_NOTE,
                    stock_code="300750", stock_name="宁德时代",
                    report_date="2024-10-15",
                    metric_name="毛利率", metric_value=22.8, unit="%",
                    confidence=0.85,
                ),
            ]
        
        await advisor.index(chunks)
        
        query = FinancialQuery(
            raw_query="宁德时代2024年三季度毛利率是多少？和之前比有什么趋势？",
            intent="data_lookup",
            stock_codes=["300750"],
            metrics=["毛利率"],
            time_range="2024-07-01",
        )
        
        result = await advisor.search(query, top_k=5)
        
        logger.info(f"检索到 {len(result['results'])} 条数据")
        for i, r in enumerate(result["results"]):
            logger.info(f"#{i+1} {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 做智能投顾的关键不是检索技术本身有多高级，而是让模型只说它从数据里读到的东西，不编造它不知道的东西。三层设计：表格解析保证结构化数据的准确性、交叉验证暴露数据源之间的差异不给用户虚假一致性、意图路由让精确查询和语义分析各走各路。金融场景没有什么"差不多"的空间，数字对错了，用户信任就没有了——而信任是投顾产品的唯一资产。