RAG 在投研报告生成中的应用：多源研报的检索与融合

一、一份投研报告需要参考 20 份券商研报——信息过载如何解决？

投研分析师在撰写一份行业报告时，通常需要阅读：

• 5-10 份券商深度研报
• 3-5 份行业白皮书
• 若干公司财报和公告

传统方式是人工逐份阅读 + 手动摘录关键数据。这个过程的效率瓶颈不在于"写作"，而在于"找到需要的信息"——在 20 份报告里找到 5 个关键数字。

RAG（检索增强生成）可以将这个过程变为：提问 → 检索相关文献段落 → 融合多维信息 → 生成报告草稿。但投研场景对 RAG 有两个特殊要求：信息的权威性排序（头部券商 > 中小券商 > 自媒体的主观分析）和多源信息的交叉验证（不同券商的预测差异需要被标注）。

二、投研 RAG 架构

三、Python 实现投研 RAG

研报知识库与权威度加权

from typing import List, Dict, Optional, Tuple from dataclasses import dataclass, field from enum import Enum import numpy as np class SourceAuthority(Enum): """数据源权威等级""" TIER_1 = 5 # 头部券商、官方数据（中信/中金/华泰/交易所公告） TIER_2 = 4 # 中大型券商、行业协会报告 TIER_3 = 3 # 中小券商、咨询公司报告 TIER_4 = 2 # 媒体分析、专家观点 TIER_5 = 1 # 自媒体、论坛分析 @dataclass class ResearchChunk: """研报知识块""" chunk_id: str text: str source_type: str # brokerage / industry_report / financial_report / news source_name: str # 券商名 / 媒体名 authority: SourceAuthority report_date: str # 报告发布日期 stock_codes: List[str] # 涉及的股票代码 industry: str # 所属行业 embedding: Optional[np.ndarray] = None class ResearchKnowledgeBase: """投研知识库——研报检索系统""" def __init__(self, embedding_model): self.encoder = embedding_model self.chunks: List[ResearchChunk] = [] self.embeddings: Optional[np.ndarray] = None def add_chunks(self, chunks: List[ResearchChunk]): """批量添加知识块""" self.chunks.extend(chunks) # 增量计算 Embedding new_embeddings = self.encoder.encode([c.text for c in chunks]) if self.embeddings is None: self.embeddings = new_embeddings else: self.embeddings = np.vstack([self.embeddings, new_embeddings]) def search( self, query: str, top_k: int = 10, filters: Optional[Dict] = None, ) -> List[Tuple[ResearchChunk, float]]: """ 检索研报——多维排序：向量相似度 × 权威度权重 × 时效性衰减 """ query_embedding = self.encoder.encode([query])[0] # 向量相似度 similarities = np.dot(self.embeddings, query_embedding) # 综合评分 scores = [] for i, chunk in enumerate(self.chunks): # 基础过滤 if filters: if "min_authority" in filters: if chunk.authority.value < filters["min_authority"]: continue if "source_names" in filters: if chunk.source_name not in filters["source_names"]: continue # 向量相似度得分（0-1） sim_score = float(similarities[i]) # 权威度加权（0-1） authority_weight = chunk.authority.value / 5.0 # 时效性衰减（近 3 个月的不衰减，超过 6 个月的衰减到 0.5） date_weight = self._calculate_date_weight(chunk.report_date) # 综合评分：相似度 * 权威度 * 时效性 final_score = sim_score * (0.5 + 0.5 * authority_weight) * date_weight scores.append((chunk, final_score)) # 按综合评分降序排列 scores.sort(key=lambda x: x[1], reverse=True) return scores[:top_k] def _calculate_date_weight(self, report_date: str) -> float: """时效性衰减计算""" from datetime import datetime, timedelta try: report_dt = datetime.strptime(report_date, "%Y-%m-%d") days_ago = (datetime.now() - report_dt).days if days_ago <= 90: # 3 个月内：不衰减 return 1.0 elif days_ago <= 180: # 3-6 个月：线性衰减到 0.7 return 1.0 - (days_ago - 90) * 0.3 / 90 else: # 6 个月以上：衰减到 0.5 return max(0.3, 0.7 - (days_ago - 180) * 0.4 / 180) except: return 0.5

多源信息融合与交叉验证

class MultiSourceFusion: """多源信息融合——去重 + 交叉验证""" def __init__(self, llm_client): self.llm = llm_client def fuse(self, query: str, retrieved: List[Tuple[ResearchChunk, float]]) -> Dict: """ 融合多源检索结果 1. 去重——合并多家券商的相同观点 2. 提取——抽取结构化关键数据 3. 验证——标注数据矛盾和分歧 """ # 按主题聚类 clusters = self._cluster_by_topic(retrieved) # 提取关键数据 key_data = self._extract_key_data(query, clusters) # 交叉验证 contradictions = self._detect_contradictions(key_data) return { "clusters": clusters, "key_data": key_data, "contradictions": contradictions, "source_count": len(retrieved), } def _cluster_by_topic(self, retrieved): """按子主题聚类——例如：销量预测、政策影响、竞争格局""" # 简化实现：使用 LLM 做主题分类 chunks_with_scores = retrieved texts = [f"[来源:{c.source_name} 权威度:{c.authority.name}]\n{c.text}" for c, _ in chunks_with_scores] prompt = f"""请将以下研报段落按子主题聚类（如：销量预测、价格趋势、政策影响、竞争格局等）。 返回 JSON 格式: {{"主题1": [段落索引列表], "主题2": [...]}}""" # LLM 聚类（生产环境可用更轻量的方法） return {"聚类结果": [texts]} def _extract_key_data(self, query, clusters): """提取关键数据——结构化的数据点""" prompt = f"""从以下研报内容中提取与 "{query}" 相关的关键数据点。 要求： 1. 每个数据点包含：指标名称、数值、单位、时间、来源 2. 如果多家券商对同一指标有不同预测，全部列出 3. 标注数据是"实际数据"还是"预测数据" 返回 JSON 格式数组。""" return [] def _detect_contradictions(self, key_data): """检测矛盾——不同券商对同一数据的分歧""" contradictions = [] # 简化版：分组对比 # 对于同一指标，如果最大值和最小值差异超过 20%，标记为存在分歧 # 生产环境用更完善的统计方法 return contradictions

研报草稿生成

class ReportDraftGenerator: """研报草稿生成器""" def __init__(self, llm_client): self.llm = llm_client def generate_draft( self, topic: str, fused_data: Dict, template: str = "standard", ) -> str: """ 生成研报草稿 核心原则：每个数据点都必须有来源引用 """ # 构造上下文（关键：包含数据来源） context = self._build_context(fused_data) # 标注数据分歧 contradictions_note = self._format_contradictions( fused_data.get("contradictions", []) ) prompt = f"""你是一位资深行业分析师。请基于以下数据撰写一份 {topic} 的行业分析报告草稿。 【写作要求】 1. 每个数据点必须注明来源（券商名 + 报告日期） 2. 如果多家券商数据不一致，在文中说明分歧 3. 格式：标题 + 摘要 + 正文（分点）+ 数据来源附录 4. 语言客观中立，不给出投资建议 5. 字数 800-1500 字 【可用数据】 {context} 【数据分歧说明】 {contradictions_note} 【报告草稿】""" draft = self.llm.generate(prompt, max_tokens=3000) # 添加引用附录 draft += "\n\n---\n【数据来源】\n" sources_seen = set() for source in fused_data.get("sources", []): key = f"{source['name']}_{source['date']}" if key not in sources_seen: draft += f"- {source['name']}, {source['date']}\n" sources_seen.add(key) draft += "\n【免责声明】\n" draft += "本报告由 AI 辅助生成，数据来源于公开研报。" draft += "报告内容不构成投资建议，投资决策请基于专业判断。" return draft

四、边界分析与 Trade-offs

权威度的主观性：

• 哪些券商是 Tier-1 需要行业经验判断
• 权威度不是绝对的——小券商在某些细分行业可能分析更深入
• 建议：对于特定行业，允许覆盖默认的权威度权重

交叉验证的精度：

• 不同券商对"2026 年新能源汽车销量"的预测可能本质上是不同口径
• 简单的数值对比会产生误报
• 复杂的交叉验证需要 NLP 模型理解语义层面的口径差异

引用准确性：

• RAG 检索到的段落可能被 LLM 错误归因到错误的来源
• 报告生成后的来源核验是必须的步骤

时效性权重：

• 研报观点可能过时，但数据（如历史财务数据）仍然有效
• 观点和数据需要不同的时效性衰减策略

五、总结

RAG 在投研报告生成中的应用核心要点：

1. 权威度加权检索——头部券商优先，提高检索结果的质量基线
2. 多源融合——合并相似观点，减少信息冗余
3. 交叉验证——标注数据分歧，让分析师自主判断
4. 强制引用——每个数据点都必须注出来源

RAG 投研的目的不是替代分析师，而是把"找信息"的时间从 2 小时压缩到 5 分钟，让分析师专注于"分析信息"。