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(keylambda x: x[1], reverseTrue) 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_tokens3000) # 添加引用附录 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 在投研报告生成中的应用核心要点权威度加权检索——头部券商优先提高检索结果的质量基线多源融合——合并相似观点减少信息冗余交叉验证——标注数据分歧让分析师自主判断强制引用——每个数据点都必须注出来源RAG 投研的目的不是替代分析师而是把找信息的时间从 2 小时压缩到 5 分钟让分析师专注于分析信息。