简介:本资源面向经济学、统计学与数据科学方向的研究者及高年级学生,提供一套将RAG技术与大语言模型应用于A股上市公司年报分析的完整统计建模方案,核心议题是评估人工智能对企业绿色全要素生产率(GTFP)的影响,并纳入企业融资约束的异质性分析与稳健性检验。压缩包共20个文件,约2.29MB,以7个py脚本和7个ipynb笔记本为主体,覆盖年报爬取、文本转换、embedding生成、GPT评分与LightGBM建模等环节,另含2个txt说明、2个json配置、1个docx附赠材料与1个md说明,便于按流程复现。已有67人学习下载。读者可据此掌握从年报数据采集、RAG检索增强到大模型文本打分、计量建模与稳健性检验的完整链路,并获得可复用的脚本与配置模板,为绿色经济与AI应用交叉研究提供参考。
1. 年报里挖出绿色生产率:这套 RAG 加 LLM 的统计建模到底在算什么
一份 A 股上市公司年报动辄两三百页,里面藏着董事会报告、经营讨论、社会责任、研发投入、环保处罚、政府补助等几十个章节。想从几千份年报里把“人工智能技术应用”和“企业绿色全要素生产率”这两件事的关系量化出来,靠人工翻 PDF 基本等于自虐。这套方案的核心思路是:先用 RAG(检索增强生成)把年报里跟 AI 投入、绿色生产、融资约束相关的段落精准捞出来,再用大语言模型做结构化抽取,最后落到面板回归模型上,做异质性分析和稳健性检验。它适合谁?适合手里有年报文本数据、想跑实证但被非结构化文本卡住的研究生和量化从业者。整条链路里,RAG 负责“找得准”,LLM 负责“读得懂”,统计建模负责“算得清”,三者缺一不可。
2. 从 PDF 到面板数据:RAG 检索层怎么搭才不丢关键信息
年报文本处理的第一个翻车点,往往不是模型不够强,而是 PDF 解析阶段就把表格和正文搅成了一锅粥。很多开源解析器对双栏排版、跨页表格、页眉页脚的容忍度极低,解析出来的文本里混着大量乱码和重复行。我一般会先用版面分析工具把年报切成“正文区”和“表格区”,正文区走文本抽取,表格区单独走 OCR 加结构还原,最后再合并。这一步做不好,后面 RAG 检索再准也是垃圾进垃圾出。
2.1 年报 PDF 解析与分块策略
分块是 RAG 检索层最容易被低估的环节。年报的章节结构天然适合做语义分块,但很多教程直接按固定 token 数硬切,结果一个完整的“环保投入”段落被切成两半,检索时两边都召回不全。我的做法是按标题层级做递归分块:先按一级标题切大块,再按二级标题切中块,最后对超过 800 token 的中块按段落边界二次切分。每个块保留其所属的章节路径作为元数据,检索时可以按章节过滤。
import re from typing import List, Dict def recursive_chunk(text: str, max_tokens: int = 800) -> List[Dict]: """ 按标题层级递归分块,保留章节路径元数据。 max_tokens 控制单块上限,超过则按段落边界二次切分。 """ # 匹配中文年报常见的标题格式:第X节、一、二、(一)(二) heading_pattern = re.compile(r'^(第[一二三四五六七八九十]+节|[一二三四五六七八九十]+、|([一二三四五六七八九十]+))') lines = text.split('\n') chunks = [] current_path = [] buffer = [] buffer_len = 0 for line in lines: if heading_pattern.match(line.strip()): # 遇到新标题,先冲刷缓冲区 if buffer: chunks.append({ 'text': '\n'.join(buffer), 'path': ' > '.join(current_path), 'token_count': buffer_len }) buffer, buffer_len = [], 0 # 更新章节路径:根据标题层级截断 level = 1 if line.startswith('第') else (2 if '、' in line[:3] else 3) current_path = current_path[:level-1] + [line.strip()] else: buffer.append(line) buffer_len += len(line) # 粗略按字符数估算 token if buffer_len >= max_tokens: chunks.append({ 'text': '\n'.join(buffer), 'path': ' > '.join(current_path), 'token_count': buffer_len }) buffer, buffer_len = [], 0 if buffer: chunks.append({ 'text': '\n'.join(buffer), 'path': ' > '.join(current_path), 'token_count': buffer_len }) return chunks这段代码的关键参数是max_tokens,设太小会导致语义碎片化,设太大则检索精度下降。根据我处理年报的经验,600 到 900 之间比较平衡。path字段记录了每个块所属的章节路径,后续检索时可以用它做元数据过滤,比如只在“董事会报告”和“社会责任”章节里搜“人工智能”和“绿色生产”。
2.2 向量化与混合检索:为什么纯向量搜不准年报
纯向量检索在年报场景下有个致命问题:它擅长语义相似,但对精确术语和数字不敏感。比如你搜“研发投入占营业收入比例”,向量模型可能召回一堆讲“研发战略”的段落,但真正包含具体比例数字的表格块反而排到后面。所以我的方案是混合检索:BM25 负责关键词精确匹配,向量检索负责语义泛化,两路结果用 RRF(倒数排名融合)合并。
from rank_bm25 import BM25Okapi import numpy as np def hybrid_retrieve(query: str, chunks: List[Dict], top_k: int = 10): """ 混合检索:BM25 + 向量相似度,RRF 融合。 query 为检索词,chunks 为分块后的年报文本列表。 """ # BM25 路 tokenized_corpus = [list(c['text']) for c in chunks] # 中文按字切分,简单但有效 bm25 = BM25Okapi(tokenized_corpus) bm25_scores = bm25.get_scores(list(query)) bm25_ranks = np.argsort(bm25_scores)[::-1] # 向量路(此处用伪代码表示,实际替换为你的 embedding 模型) # vector_scores = embedding_model.encode([c['text'] for c in chunks]) @ embedding_model.encode(query) # vector_ranks = np.argsort(vector_scores)[::-1] vector_ranks = bm25_ranks # 占位,实际替换 # RRF 融合 rrf_scores = {} for rank, idx in enumerate(bm25_ranks[:top_k*2]): rrf_scores[idx] = rrf_scores.get(idx, 0) + 1 / (60 + rank) for rank, idx in enumerate(vector_ranks[:top_k*2]): rrf_scores[idx] = rrf_scores.get(idx, 0) + 1 / (60 + rank) sorted_items = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True) return [chunks[i] for i, _ in sorted_items[:top_k]]RRF 里的常数 60 是经验值,来自信息检索领域的标准做法,作用是平滑排名靠前和靠后文档的权重差距。中文 BM25 的分词我直接按字切,虽然粗糙,但在年报这种术语密集的场景下反而比某些分词器更稳,因为不会把“绿色全要素生产率”切成奇怪的碎片。如果你有领域词典,换成 jieba 自定义分词会更好。
2.3 检索质量评估:召回率不够时先查这三个地方
检索层搭好后,别急着上 LLM。先做一轮人工评估:从年报里挑 20 个你确定包含关键信息的段落,看它们能不能被检索到前 10 结果里。如果召回率低于 70%,按顺序查这三个地方:第一,分块是不是把关键段落切碎了,看path元数据是否完整;第二,BM25 的分词粒度是不是太粗或太细,试着调整分词策略;第三,查询词本身是不是太宽泛,比如“人工智能”不如“人工智能 研发 投入”这种组合词精准。我踩过的坑是:一开始用固定 512 token 分块,结果“环保投入”和“政府补助”经常被切到同一个块里,检索时互相干扰,后来改成按章节递归分块才解决。
3. 让 LLM 从段落里抽变量:Prompt 设计与结构化输出
检索层把相关段落捞出来后,下一步是用大语言模型把这些非结构化文本转成结构化变量。这一步的核心挑战是:LLM 容易“过度解读”,把一段泛泛而谈的“公司重视环保”硬生生抽成“有环保投入”,导致变量失真。我的策略是设计一个带“证据锚定”的 Prompt,要求模型在抽取每个变量时,必须引用原文中的具体句子作为依据,没有明确依据就填“未提及”。
3.1 变量定义表与 Prompt 模板
在写 Prompt 之前,先把你要抽取的变量定义清楚。这套方案里核心变量分三组:AI 应用强度(是否有 AI 相关研发、AI 在产品中的应用描述)、绿色全要素生产率相关(环保投入、绿色专利、能耗强度、碳排放描述)、融资约束相关(现金流紧张、融资难度、信贷获取描述)。每个变量给一个明确的判定标准,避免模型自由发挥。
| 变量名 | 类型 | 判定标准 | 取值 |
|---|---|---|---|
| AI_RD | 二值 | 年报中明确提及人工智能研发投入或 AI 技术研发项目 | 0/1 |
| AI_Product | 二值 | 年报中描述 AI 技术已应用于产品或服务 | 0/1 |
| Green_Invest | 连续 | 环保投入金额(万元),取年报中明确披露的数值 | 数值 |
| Green_Patent | 计数 | 绿色专利数量,取年报中明确披露的数值 | 整数 |
| Energy_Intensity | 连续 | 单位营收能耗,若披露则抽取 | 数值 |
| Finance_Constraint | 二值 | 年报中提及融资困难、现金流紧张、信贷受限等表述 | 0/1 |
Prompt 模板的设计要点是:先给变量定义表,再给待抽取文本,最后要求模型以 JSON 格式输出,每个字段包含value和evidence两个子字段。evidence必须是原文中的直接引用,不能是模型转述。
EXTRACT_PROMPT = """你是一个财务年报信息抽取助手。请根据以下变量定义,从文本中抽取对应信息。 变量定义: - AI_RD:年报中是否明确提及人工智能研发投入或AI技术研发项目,是填1,否填0 - AI_Product:年报中是否描述AI技术已应用于产品或服务,是填1,否填0 - Green_Invest:环保投入金额(万元),只填数字,未披露填null - Green_Patent:绿色专利数量,只填数字,未披露填null - Finance_Constraint:年报中是否提及融资困难、现金流紧张、信贷受限,是填1,否填0 要求: 1. 每个变量必须给出原文证据句子,不能转述。 2. 没有明确证据的变量填null,不要猜测。 3. 输出JSON格式,每个字段包含value和evidence。 待抽取文本: {text} 输出:"""这个 Prompt 的关键在于“没有明确证据就填 null”这条约束。我试过不加这条,模型会把“公司积极推进智能化转型”这种模糊表述也标成 AI_RD=1,导致变量虚高。加上证据锚定后,抽取准确率明显提升,代价是召回率会降一些,但对于统计建模来说,宁可漏掉也不能错标。
3.2 批量抽取与结果校验
几千份年报不可能一条条手动跑,需要批量处理。批量抽取的工程要点是:控制并发数避免 API 限流,对失败请求做重试,对输出做 JSON 解析容错。我一般用 5 到 10 的并发,每个请求超时设 30 秒,失败重试 3 次。解析时如果 JSON 格式不对,先用正则把 JSON 部分抠出来再解析,实在不行就记录原始输出人工复核。
import json import re from concurrent.futures import ThreadPoolExecutor, as_completed def parse_llm_output(raw: str) -> dict: """从 LLM 原始输出中解析 JSON,容忍前后有解释文字的情况。""" match = re.search(r'\{.*\}', raw, re.DOTALL) if not match: return {'error': 'no_json_found', 'raw': raw[:200]} try: return json.loads(match.group()) except json.JSONDecodeError: return {'error': 'json_decode_failed', 'raw': match.group()[:200]} def batch_extract(texts: list, prompt_template: str, max_workers: int = 8): """批量抽取,控制并发数,返回结果列表。""" results = [None] * len(texts) with ThreadPoolExecutor(max_workers=max_workers) as executor: future_map = { executor.submit(call_llm, prompt_template.format(text=t)): i for i, t in enumerate(texts) } for future in as_completed(future_map): idx = future_map[future] try: raw = future.result(timeout=30) results[idx] = parse_llm_output(raw) except Exception as e: results[idx] = {'error': str(e)} return resultsmax_workers设 8 是我在多个 API 上试出来的平衡点,再高容易触发限流,再低则跑几千份年报太慢。解析容错那一步很重要,LLM 有时候会在 JSON 前后加一句“以下是抽取结果”,直接json.loads会报错,用正则抠出花括号部分再解析就稳了。
3.3 抽取结果的人工抽检与修正
批量跑完后,别直接拿去做回归。先随机抽 50 到 100 条结果,人工核对evidence字段是否真的支持value。我一般重点看两类:一类是value为 1 但evidence很模糊的,另一类是value为 null 但原文明显有相关信息的。抽检发现系统性偏差时,回去改 Prompt 重新跑,而不是手动改几条了事。这一步花的时间,比后面回归跑出来结果不对再回头查要少得多。
4. 绿色全要素生产率的测算与面板回归建模
变量抽出来后,就进入统计建模阶段。绿色全要素生产率(Green TFP)不能直接从年报文本里读出来,需要用投入产出数据测算。常见做法是先把劳动、资本、能源作为投入,把考虑环境产出的增加值作为产出,用 SBM-GML 指数或 EBM 方法测算。年报文本抽取的 AI 应用变量和融资约束变量,则作为解释变量和调节变量进入面板回归。
4.1 绿色 TFP 的测算:投入产出指标怎么选
测算绿色 TFP 的第一步是定指标。劳动投入用员工人数,资本投入用固定资产净值,能源投入用年报里披露的能耗数据或代理变量。产出端分两部分:期望产出用营业收入,非期望产出用碳排放或污染物排放。如果年报里没有直接披露碳排放,可以用能耗乘以排放因子估算,但要在稳健性检验里换一种估算方式验证结果是否敏感。
| 指标类型 | 具体指标 | 数据来源 | 处理方式 |
|---|---|---|---|
| 劳动投入 | 员工总数 | 年报 | 取年初年末均值 |
| 资本投入 | 固定资产净值 | 年报 | 取年初年末均值 |
| 能源投入 | 综合能耗 | 年报/ESG报告 | 缺失时用营业成本代理 |
| 期望产出 | 营业收入 | 年报 | 直接取 |
| 非期望产出 | 碳排放量 | 估算 | 能耗×排放因子 |
测算工具方面,Python 里可以用pulp或scipy自己写 SBM 模型,也可以用现成的 DEA 包。我一般自己写,因为年报数据的缺失情况复杂,现成包对缺失值的处理不够灵活。测算出来的绿色 TFP 是一个逐年逐公司的面板数据,后续回归直接用它做被解释变量。
4.2 基准回归与异质性分析:融资约束怎么当调节变量
基准回归的设定是:绿色 TFP 对 AI 应用变量回归,控制企业规模、资产负债率、ROA、股权集中度等常见控制变量,固定企业个体效应和年份效应。融资约束的异质性分析,则是把样本按融资约束程度分组,或者引入交互项。我的做法是两者都做:先按融资约束中位数分组回归,看系数差异;再在全样本里加 AI×Finance_Constraint 交互项,看交互项显著性。
import statsmodels.formula.api as smf # 基准回归:绿色TFP ~ AI应用 + 控制变量 + 个体/年份固定效应 model_base = smf.ols( 'Green_TFP ~ AI_RD + AI_Product + Size + Lev + ROA + Top1 + C(Industry) + C(Year)', data=df ).fit(cov_type='cluster', cov_kwds={'groups': df['Firm_ID']}) # 异质性分析:按融资约束分组 median_fc = df['Finance_Constraint'].median() df_high = df[df['Finance_Constraint'] > median_fc] df_low = df[df['Finance_Constraint'] <= median_fc] model_high = smf.ols( 'Green_TFP ~ AI_RD + AI_Product + Size + Lev + ROA + Top1 + C(Industry) + C(Year)', data=df_high ).fit(cov_type='cluster', cov_kwds={'groups': df_high['Firm_ID']}) model_low = smf.ols( 'Green_TFP ~ AI_RD + AI_Product + Size + Lev + ROA + Top1 + C(Industry) + C(Year)', data=df_low ).fit(cov_type='cluster', cov_kwds={'groups': df_low['Firm_ID']})这里用cov_type='cluster'按企业聚类标准误,是因为同一家公司的年度观测值之间不独立,不聚类的话标准误会偏小,容易把不显著的结果误判为显著。分组回归后比较AI_RD和AI_Product的系数大小和显著性,如果高融资约束组的系数明显更小或不显著,就说明融资约束确实抑制了 AI 对绿色 TFP 的促进作用。
4.3 稳健性检验:换变量、换样本、换模型
稳健性检验是让审稿人信服的关键。我一般做三组:第一,替换被解释变量的测算方法,比如把 SBM-GML 换成 EBM 再算一遍绿色 TFP;第二,替换核心解释变量,把 AI_RD 从二值变量换成 AI 相关词频占比;第三,缩尾处理,对连续变量做 1% 和 99% 分位的缩尾,排除极端值影响。三组结果方向一致,才能说结论是稳的。
# 稳健性检验1:替换被解释变量 model_robust1 = smf.ols( 'Green_TFP_EBM ~ AI_RD + AI_Product + Size + Lev + ROA + Top1 + C(Industry) + C(Year)', data=df ).fit(cov_type='cluster', cov_kwds={'groups': df['Firm_ID']}) # 稳健性检验2:替换核心解释变量为词频占比 model_robust2 = smf.ols( 'Green_TFP ~ AI_WordFreq + Size + Lev + ROA + Top1 + C(Industry) + C(Year)', data=df ).fit(cov_type='cluster', cov_kwds={'groups': df['Firm_ID']}) # 稳健性检验3:缩尾处理 for col in ['Green_TFP', 'Size', 'Lev', 'ROA']: lower, upper = df[col].quantile([0.01, 0.99]) df[col + '_winsor'] = df[col].clip(lower, upper) model_robust3 = smf.ols( 'Green_TFP_winsor ~ AI_RD + AI_Product + Size_winsor + Lev_winsor + ROA_winsor + Top1 + C(Industry) + C(Year)', data=df ).fit(cov_type='cluster', cov_kwds={'groups': df['Firm_ID']})三组稳健性检验里,替换被解释变量那组最重要,因为绿色 TFP 的测算方法本身就有多种流派,换一种方法结论不变才说明不是测算方法带来的伪相关。缩尾处理那组最容易被忽略,但年报数据里经常有极端值,不缩尾的话回归结果可能被一两家公司带偏。
5. 避坑与排查:这套链路里最容易翻车的五个地方
5.1 检索召回率低,LLM 拿不到关键段落
现象:抽取结果里大量变量为 null,人工翻原文却发现信息明明存在。原因通常是分块把关键段落切碎了,或者 BM25 分词粒度不对导致检索不到。解决:先检查分块的path元数据是否完整,再试着把 BM25 换成 jieba 自定义分词,加入领域词典。如果还不行,把查询词从单词改成组合词,比如“人工智能”改成“人工智能 研发 投入”。
5.2 LLM 过度抽取,把模糊表述标成确定变量
现象:AI_RD 变量为 1 的比例异常高,人工核对发现很多证据只是“公司关注智能化趋势”这种泛泛表述。原因是 Prompt 里没有强制证据锚定。解决:在 Prompt 里加“没有明确证据填 null”的约束,并要求evidence必须是原文直接引用。改完后召回率会降,但准确率提升,对回归来说更安全。
5.3 面板回归标准误不聚类,显著性虚高
现象:回归结果里核心变量系数显著,但换成聚类标准误后不显著了。原因是同一家公司的年度观测值之间不独立,普通标准误偏小。解决:所有面板回归都加cov_type='cluster',按企业聚类。如果企业数量太少,可以考虑按行业和年份双重聚类,但要注意聚类数不能太少,否则聚类标准误本身也不可靠。
5.4 绿色 TFP 测算时投入产出指标口径不一致
现象:不同年份的绿色 TFP 波动异常大,或者某些公司数值明显不合理。原因是投入产出指标的口径在不同年份或不同公司之间不一致,比如有的年份能耗数据来自年报,有的年份来自 ESG 报告,统计范围不同。解决:统一数据来源优先级,优先用年报正文披露的数据,缺失时用同行业同年份中位数填充,并在稳健性检验里换一种填充方式验证。
5.5 稳健性检验只换模型不换变量,审稿人不买账
现象:稳健性检验做了好几组,但都是换回归模型(比如从 OLS 换到固定效应),核心变量和被解释变量的定义没变。审稿人会觉得这不算真正的稳健性检验。解决:稳健性检验要动核心变量的定义或测算方法,比如换绿色 TFP 的测算方法、换 AI 变量的度量方式、换样本期间。只换模型不换变量,说服力有限。
6. 进阶技巧:把 RAG 检索层做成可复用的年报分析流水线
这套方案跑通一次后,最值得做的事是把检索层和抽取层做成可复用的流水线,而不是每次换研究问题就从头搭一遍。我的做法是把年报解析、分块、向量化、混合检索、LLM 抽取这五步封装成独立的模块,每个模块的输入输出用标准格式对接。这样下次想研究“数字化转型对碳排放的影响”,只需要换一套变量定义和 Prompt,检索层和解析层直接复用。
具体来说,解析层输出统一的 JSON 格式,每个块包含text、path、source_file、page_num四个字段。检索层接收查询词和可选的章节过滤条件,返回带相似度分数的块列表。抽取层接收块列表和 Prompt 模板,返回结构化变量表。三层之间用文件系统或轻量数据库对接,不搞复杂的服务化,因为年报分析通常是离线批处理,没必要上实时接口。
# 流水线入口示例:从年报目录到变量表 import os from pathlib import Path def run_pipeline(pdf_dir: str, output_csv: str): """年报分析流水线:解析 -> 分块 -> 检索 -> 抽取 -> 输出变量表。""" all_chunks = [] for pdf_file in Path(pdf_dir).glob('*.pdf'): text = parse_pdf(str(pdf_file)) # 解析层 chunks = recursive_chunk(text) # 分块层 for c in chunks: c['source_file'] = pdf_file.name all_chunks.extend(chunks) # 对每个研究问题,换查询词和 Prompt 即可 queries = ['人工智能 研发 投入', '环保 投入 绿色 生产', '融资 约束 现金流'] retrieved = [] for q in queries: retrieved.extend(hybrid_retrieve(q, all_chunks, top_k=20)) # 去重后送入 LLM 抽取 seen = set() unique_chunks = [] for c in retrieved: key = (c['source_file'], c['path'], c['text'][:50]) if key not in seen: seen.add(key) unique_chunks.append(c) results = batch_extract([c['text'] for c in unique_chunks], EXTRACT_PROMPT) # 后续合并、清洗、导出 CSV 略 return results这个流水线的关键设计是“检索与抽取解耦”:检索层不关心你要抽什么变量,只负责按查询词召回相关块;抽取层不关心块是怎么来的,只负责按 Prompt 抽变量。这样换研究问题时,只需要改查询词列表和 Prompt 模板,解析和分块完全不用动。我自己的习惯是每跑完一个研究问题,就把当时的查询词和 Prompt 存成一个配置文件,下次做类似问题时先翻旧配置,能省不少调参时间。
最后一个血泪经验:别在流水线里硬编码任何文件路径和 API 密钥,用环境变量或配置文件管理。我早期图省事把密钥写在脚本里,结果换机器跑的时候忘了改,白白浪费一晚上排查为什么请求全部失败。希望帮到你。
本文还有配套的精品资源,点击获取