利用Python与LLM提取财报电话会议中的AI生产力信号

📅 2026/8/27 22:58:51
利用Python与LLM提取财报电话会议中的AI生产力信号
企业财报电话会议是观察AI落地的重要窗口。越来越多的公司在股东电话会议中提到AI但高层说“AI提升生产力”和现实中AI真正影响运营之间往往隔着一段需要定量分析的证据链。这篇文章从一个可以落地的分析视角来拆解这件事用 Python 和大型语言模型把财报电话会议文本转化为可追踪、可验证的 AI 生产力信号。你会看到一张完整的分析管线包括数据准备、文本清洗、AI 相关语句召回、生产力维度分类、指标设计、结果验证和生产化改造整个过程可以在本地复现也可以扩展成自动化分析服务。1. 财报电话会议中的AI叙事为什么值得做技术分析1.1 从“提到AI”到“AI影响生产力”的语义断层财报电话会议Earnings Call是上市公司管理层与分析师之间的公开对话通常包含 CEO、CFO 对经营成果的陈述和分析师提问。近几年“AI”这个词在电话会议中被频繁提及但这并不等于公司已经在用 AI 改善生产力。这里存在一个语义断层管理层可以说“我们正在利用 AI 提升客户满意度”也可能说“AI 相关云服务收入同比增长 30%”。前者是模糊的愿景叙述后者是相对可验证的经营结果。如果只统计“AI”出现次数会把两类完全不同的信息混在一起。技术分析要做的事情就是把“提到 AI”的句子进一步拆分成几个维度是否涉及具体的生产力场景例如自动化流程、降低客服成本、辅助编程、供应链预测。是否提到可以观察的证据例如节省工时、缩短交付周期、人员结构调整、收入增量。是否包含时间、部门、产品或财务指标还是停留在“战略布局”层面。这一层拆解本质上是一次文本分类和信息抽取任务。输入是电话会议的逐字稿文本输出是结构化的“AI 生产力信号”片段方便后续按公司、按季度、按行业做趋势分析。1.2 分析目标把叙事文本转成可验证信号如果把财报电话会议当成一个非结构化数据集每一句话都是一个样本。分析管线的整体目标可以定义成输入一段会议文本输出该文本中与“AI 生产力”相关的关键证据列表每条证据至少包含原始语句、所属公司、会议日期、涉及的生产力维度、证据强度、情感倾向。这样做的价值在于分析师不必再人工翻阅几十份 PDF 逐字稿投资研究团队可以批量跟踪“AI 从战略叙事走向实际运营”的季度变化产品团队也可以透过公开信息判断竞品在 AI 自动化上的进展。从技术实现上这条管线并不需要一开始就做得非常复杂。先用规则做候选句召回再用大模型做细粒度判断最后用人工抽样评估效果是起步阶段最稳妥的方案。下面用一个最小可运行案例来搭建这条管线。为了让读者能直接看到输入输出本文示例会使用公开的会议记录片段和模拟数据实际项目中请替换成你自己获取的授权数据。2. 搭建文本分析环境依赖、模型与数据来源2.1 Python环境与核心依赖建议使用 Python 3.10 或更高版本。核心依赖分成四类数据处理、NLP 基础能力、大模型接口、可视化。依赖库用途安装方式pandas表格数据处理pip install pandasnumpy数值计算pip install numpyjieba中文分词可选pip install jiebaopenai兼容大模型 API 调用pip install openaimatplotlib指标趋势图pip install matplotlibpyyaml配置文件读取pip install pyyaml建议先创建一个独立虚拟环境避免和系统 Python 环境相互污染。python -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate pip install pandas numpy openai matplotlib pyyaml jieba这里有一个容易忽略的点openai库本身并不局限于官方接口很多国产大模型服务商都提供“OpenAI 兼容”的 HTTP 接口只需要修改base_url和api_key。示例代码会统一使用这种兼容模式方便切换模型供应商。2.2 大模型接入方式与成本控制大模型在本项目中负责两个任务语义判断和信息抽取。考虑到财报电话会议是英文会议记录占多数但中文媒体转写的会议摘要也大量存在至少要让模型同时支持中英文。接入方式建议用环境变量保存密钥不要在代码里写死export LLM_API_KEYyour_api_key_here export LLM_BASE_URLhttps://your-llm-endpoint.example.com/v1 export LLM_MODELyour-model-name在 Python 中读取import os api_key os.getenv(LLM_API_KEY) base_url os.getenv(LLM_BASE_URL) model os.getenv(LLM_MODEL, gpt-4o-mini) from openai import OpenAI client OpenAI(api_keyapi_key, base_urlbase_url)成本控制的核心是减少 Token 消耗。财报电话会议全文可能长达数万词如果每个句子都调用一次大模型费用和耗时都不可控。推荐做法是先做规则召回只把“疑似涉及 AI 的句子”送入大模型把调用量压缩一个数量级。后续代码中会体现这一策略。2.3 数据获取的合规路径公开渠道可以获取财报电话会议逐字稿的来源包括美国 SEC EDGAR 系统中部分公司会提交包含电话会议内容的 8-K 附件。财经数据供应商提供的电话会议文本数据集。公司投资者关系网站公开的财报电话会议记录 PDF。学术或研究机构发布的金融文本数据集。需要注意这些来源的许可证与使用限制不完全相同。用于学习研究时尽量选择明确授权公开的数据。如果是商业用途要核对服务条款。原始文本既可以是 CSV 文件也可以是 JSON 文件下面是两种常见结构。CSV 示例company,ticker,date,quarter,speaker,text Acme Corp,ACM,2024-11-01,Q3 2024,CEO,Our AI-driven recommendation engine has reduced response time by 30%. Acme Corp,ACM,2024-11-01,Q3 2024,CFO,We expect automation to lower operating costs in the next fiscal year.JSON 示例[ { company: Acme Corp, ticker: ACM, date: 2024-11-01, quarter: Q3 2024, speaker: CEO, text: Our AI-driven recommendation engine has reduced response time by 30%. } ]进入代码之前需要先确认字段完整性。项目代码可以加一个简单校验函数import pandas as pd required_cols [company, ticker, date, quarter, speaker, text] def load_transcripts(path): df pd.read_csv(path) missing [col for col in required_cols if col not in df.columns] if missing: raise ValueError(f缺少必要字段: {missing}) df[text] df[text].astype(str) return df这样在数据加载阶段就能发现字段缺失问题而不是等到分析时才发现某条记录是空字符串。3. 预处理财报电话会议文本从原始文本到可分析片段3.1 原始文本的特点财报电话会议的原始文本通常来自自动语音识别或人工转写存在几个典型问题包含大量填充词例如 “um”、“uh”、“well”、“you know”。有主持人提示语例如 “Operator: Your next question comes from...”。一句话可能被换行或标点切割成多段尤其是英文逗号、分号混在一起。中英文混排时标点和空格不规范。如果直接把这整段文本送入模型会让召回结果噪声很大。所以预处理要按“会议 ID - 发言片段 - 句子”的层级拆分。3.2 分句与片段化策略可以用re.split按句号、问号、感叹号分句但需要注意小数点、缩写词造成的误切。英文中 “Mr.”、“e.g.” 这类缩写如果直接按句号切分会把一句完整的话拆成两段。一个简单策略是先用正则做保守切分再做合并。示例代码import re def split_sentences(text: str) - list[str]: # 保护常见缩写 text re.sub(r\b(Mr|Ms|Dr|e\.g|i\.e|vs)\., r\1DOT, text) parts re.split(r(?[.!?。])\s, text.strip()) cleaned [] for part in parts: part part.replace(DOT, .) if part.strip(): cleaned.append(part.strip()) return cleaned这里用DOT占位符保留缩写中的句点避免把一句话切碎。对于中文文本如果原始语料没有明确标点可以先使用jieba分词或模型做断句但生产项目中更推荐使用专门的句子分割模型或高质量转写工具。3.3 数据清洗规则数据清洗按顺序执行删除主持人操作语例如 “Your line is open.”、“Please go ahead.”。删除纯数字、时间戳、音频转写标识。合并由于换行导致的半句话。统一全角和半角标点。去除多余空格但保留句子内部换行信息用于后续定位。清洗函数示例def clean_text(text: str) - str: text re.sub(r\s, , text) text text.replace(’, ).replace(“, ).replace(”, ) text re.sub(r\[.*?\], , text) # 删除转写方括号内容 text re.sub(r\(.*?\), , text) # 根据场景决定是否删除括号 return text.strip()需要说明删除括号内容并不总是安全的。如果括号中是分析师姓名或职位可以删如果是财务数据如 “(GAAP EPS: $1.20)”删掉会丢失关键证据。建议把这一步作为可配置项不要写死。预处理完成后把每条句子打上会议和发言者信息生成一个长表供下一步召回使用。def build_sentence_df(df: pd.DataFrame) - pd.DataFrame: rows [] for _, row in df.iterrows(): for sent in split_sentences(clean_text(row[text])): rows.append({ company: row[company], ticker: row[ticker], date: row[date], quarter: row[quarter], speaker: row[speaker], sentence: sent }) return pd.DataFrame(rows)4. 设计AI生产力信号提取管线4.1 第一步规则召回AI相关语句为了控制大模型调用量先用关键词规则把候选句召回。这里需要覆盖英文和中文常见表达。英文关键词可以包括ai,artificial intelligence,machine learning,deep learning,automation,intelligent,copilot,agent,generative ai,nlp,computer vision。中文关键词包括人工智能、AI、大模型、机器学习、自动化、智能体、算法、模型。注意大小写问题。AI是全大写词但ai也可能是某些单词的一部分例如said中并不包含ai但email包含ai。召回时需要按词边界匹配。import re AI_KEYWORDS [ r\bai\b, r\bartificial intelligence\b, r\bmachine learning\b, r\bdeep learning\b, r\bautomation\b, r\bintelligent\b, r\bcopilot\b, r\bagent\b, r\bgenerative ai\b, r\bnlp\b, r\bcomputer vision\b, r人工智能, r大模型, r机器学习, r自动化, r智能体, r算法, r智能 ] def recall_candidates(sentence: str) - bool: lower sentence.lower() for pattern in AI_KEYWORDS: if re.search(pattern, lower): return True return False召回阶段宁可多召回也不要漏掉关键句。因为后面有大模型做细粒度判断召回多一些只会增加少量成本漏掉则会直接丢失信号。4.2 第二步LLM判断生产力维度候选句进入大模型后需要让它输出结构化 JSON。这里的关键是定义清楚分类维度。可以把“生产力信号”分成五个维度维度含义示例句效率提升明确提到速度、响应时间、人均产出变化AI客服降低了平均响应时间成本优化提到成本下降、运营费用减少、资源利用率提升自动化帮助我们减少外包支出收入贡献提到AI直接或间接带来收入AI相关订阅收入同比增长20%人员与组织提到编制、岗位、技能变化我们重新培训客服团队使用AI助手战略与研发提到研发投入、产品路线、合作投资公司计划在2025年扩大AI研发预算大模型只需要判断“这句话是否包含上述维度的生产力证据”并输出证据强度等级high、medium、low、none。none表示提到了 AI 但只是愿景或泛泛而谈没有生产力证据。Prompt 设计示例SYSTEM_PROMPT 你是一个严谨的财报文本分析助手。 你的任务是从一句话中判断它是否包含 AI 或自动化对生产力影响的证据。 判断维度 - efficiency: 效率提升 - cost: 成本优化 - revenue: 收入贡献 - people: 人员与组织 - strategy: 战略与研发 输出 JSON 格式 { contains_evidence: true, dimension: efficiency, strength: high, reason: 简要说明判断依据 } 规则 1. 如果只是提到 AI 但没有具体影响strength 为 lowcontains_evidence 为 false。 2. 如果提到了时间、比例、金额、人员数量等可验证信息strength 可设为 high。 3. 不要自行补充原文不存在的信息。 调用模型时把候选句放入 user 消息。为了提升稳定性可以设置temperature0并输出约束到 JSON。def analyze_sentence(sentence: str) - dict: completion client.chat.completions.create( modelmodel, temperature0, response_format{type: json_object}, messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: fSentence: {sentence}} ] ) content completion.choices[0].message.content import json return json.loads(content)4.3 第三步防止AI幻觉的验证机制大模型在分类时可能产生幻觉常见有两种模型把原文没有的数据补充到理由里。模型因为过度拟合训练数据把 “AI strategy” 错误分类成 “high impact”。在提示词中约束只能引用原文是第一步。第二步是在后处理中做校验检查strength字段是否在合法枚举中。如果strength high要求理由里必须出现原文中的数字、百分比、时间词否则降级为medium。如果contains_evidencefalse但是该句命中多个生产力关键词标记为待人工复核。示例校验函数def validate_result(original_sentence: str, parsed: dict) - dict: allowed_strength {high, medium, low, none} strength parsed.get(strength, none) if strength not in allowed_strength: parsed[strength] none if strength in {high, medium}: # 关键字校验high 强度必须包含数字证据 has_number bool(re.search(r\d(\.\d)?%?, original_sentence)) has_evidence_word any(w in original_sentence.lower() for w in [ increase, reduce, decrease, improve, saving, time, cost, revenue, 提升, 降低, 减少, 收入, 成本 ]) if not (has_number or has_evidence_word): parsed[strength] low parsed[flag] weak_evidence return parsed4.4 核心代码实现下面这段主流程把召回、分析、校验串起来。实际项目中建议增加缓存避免重复调用 API。def extract_ai_productivity_signals(sentence_df: pd.DataFrame, batch_size: int 10) - pd.DataFrame: results [] candidate_count 0 for _, row in sentence_df.iterrows(): sentence row[sentence] if not recall_candidates(sentence): continue candidate_count 1 parsed {} try: parsed analyze_sentence(sentence) except Exception as exc: parsed { contains_evidence: None, dimension: unknown, strength: none, reason: fAPI_ERROR: {exc} } parsed validate_result(sentence, parsed) results.append({**row, **parsed}) if len(results) % batch_size 0: print(f已处理 {len(results)} 条候选句) print(f候选句总数: {candidate_count}) return pd.DataFrame(results)这一段逻辑并不复杂但已经能形成一个最小闭环。你只需要准备一个 CSV 文件运行后就能得到包含dimension和strength的 DataFrame。5. 构建可量化的AI生产力指标5.1 指标定义与计算逻辑提取出的结构化结果可以聚合成若干指标。本文定义四个核心指标。首先是 AI 提及密度。它计算每篇电话会议中“提到 AI 的句子数”占“总句子数”的比例用来衡量管理层讨论 AI 的活跃程度。def ai_mention_density(group: pd.DataFrame) - float: total len(group) if total 0: return 0.0 ai_sentences group[sentence].apply(recall_candidates).sum() return round(ai_sentences / total, 4)其次是生产力信号占比。它只统计被模型标记为contains_evidenceTrue的句子数占 AI 相关句子数的比例。这个指标比单纯提 AI 更接近“AI 是否真实作用于运营”。def productivity_signal_ratio(group: pd.DataFrame) - float: ai_candidates group[group[sentence].apply(recall_candidates)] if len(ai_candidates) 0: return 0.0 evidence_count ai_candidates[ai_candidates[contains_evidence] True].shape[0] return round(evidence_count / len(ai_candidates), 4)第三是证据强度得分。对strength做加权打分high3medium2low1none0。可以看出一家公司从“口头提AI”到“给出数据证据”的进阶程度。第四是维度分布。统计五大维度分别出现多少次用于判断公司当前 AI 落地的侧重点是降本、增效还是创收。def dimension_distribution(df: pd.DataFrame) - pd.Series: evidence_df df[df[contains_evidence] True] return evidence_df[dimension].value_counts()5.2 指标解释和阈值选择指标不能只给数值还要解释业务含义。指标计算口径数值偏低数值偏高AI提及密度AI句子数 / 总句子数业务与AI关联弱叙事驱动明显需要进一步看证据生产力信号占比有效性证据句 / AI候选句多数只是战略口号AI已进入具体业务环节证据强度得分high3, medium2, low1缺少量化描述存在时间、金额、比例等可验证数据阈值没有统一标准。用于横向比较时建议分行业分季度看相对变化。一家传统零售公司的 AI 提及密度可能一直低于科技公司但若它连续两个季度从 0.5% 升到 1.5%同时生产力信号占比同步提升就是一个值得追踪的信号。5.3 可视化输出把时间序列绘制成折线图方便观察变化趋势。import matplotlib.pyplot as plt def plot_trend(summary_df: pd.DataFrame, metric: str, title: str): plt.figure(figsize(10, 5)) plt.plot(summary_df[date], summary_df[metric], markero) plt.title(title) plt.xlabel(会议日期) plt.ylabel(metric) plt.xticks(rotation45) plt.tight_layout() plt.savefig(f{metric}_trend.png) plt.show()这里summary_df是按季度汇总后的指标表。如果公司数量较多可以先筛选 top N 公司再绘图避免图形过于拥挤。6. 运行验证与结果解读6.1 小样本人工标注在把自动分析结果用于任何正式研究前必须做人工抽样验证。建议从输出结果中随机抽取 50 到 100 条记录由两个人分别标注最后计算一致率。标注界面可以是一张简单的表格包含原始句子、模型判断维度、证据强度、人工判断。人工标注主要看三个问题这句话是否真的涉及 AI 生产力。模型判断的维度是否正确。证据强度是否和原文描述的量化程度匹配。计算准确率时把“模型判断维度与人工一致”记为正确。如果两轮标注结果不一致需要重新定义 prompt 或规则。6.2 输出示例与口径说明假设输入文本是Our AI-powered support bot has reduced average response time from 4 hours to 30 minutes.模型可能输出{ contains_evidence: true, dimension: efficiency, strength: high, reason: 原文提到了响应时间从4小时下降到30分钟量化证据充分。 }这里strengthhigh是合理的因为有明确数字和时间单位。但需要注意这只能说明管理层提到了结果并不代表财务数据已经验证。分析口径要写成“电话会议中披露的 AI 生产力声明”而不是“AI 真实生产力提升”。在结果报告中必须区分这两个口径否则会把叙事当成事实。6.3 常见偏差和修正方案自动分析容易出现系统性偏差。偏差一是语言偏差。英文语料中AI匹配规则较稳定但中文语料中“智能”一词使用范围很广例如“智能制造能力”“智能化转型”这些词不一定代表 AI 生产力证据。修正方法是增加否定词或场景词过滤或单独训练一个二分类器。偏差二是行业偏差。银行和制造商会用“算法”“模型”指代传统统计模型而不是生成式 AI。可以在召回阶段加入llm、generative ai、大模型等更明确的词并允许人工配置行业词典。偏差三是时间偏差。一段话可能在回顾过去的 AI 投入效果也可能在描述未来计划。模型默认不会区分时态所以 prompt 应加入一个time_horizon字段输出past、present、future之一。这在投资分析中尤其重要因为未来规划不能算已实现的生产力。def classify_time(sentence: str) - str: past_words [已, 完成, 降低, 提升, 节省, reduced, improved, saved] future_words [计划, 预计, 将, will, plan, expect, goal] lower sentence.lower() if any(w in lower for w in future_words): return future if any(w in lower for w in past_words): return past return present这个简单函数可以作为兜底更精确的做法是让模型输出time_horizon字段。7. 常见问题排查从API报错到信号误判下面这张表总结了运行本管线时最常遇到的问题按“现象 - 原因 - 检查方式 - 处理建议”列出。问题现象常见原因检查方式处理建议API 返回 401 或 403API Key 错误、没有读取环境变量检查环境变量是否生效打印 key 前缀确认 key 写入正确的 shell 环境不要在代码中硬编码输出不是合法 JSON模型返回了 Markdown 或额外解释打印模型原始返回内容在 prompt 中增加response_format后处理中增加 JSON 解析兜底候选句数量极其少关键词规则太严格或文本为空打印sentence列统计放宽关键词确认清洗没有误删全部内容高比例contains_evidencetrue模型过于宽松喜欢给出肯定判断抽样人工标注调整 prompt 规则增加none示例高比例strengthlow原文多是战略表述缺少量化证据检查原始语料是否只包含管理层主旨发言同时检查分析师问答部分里面经常有更具体的数字中文文本召回效果差关键词没有覆盖“大模型”“智能体”检查分词和关键词列表增加中文领域词表必要时增加同义词映射结果不稳定重复运行不同temperature 不为 0 或模型非确定性固定 temperature0固定随机种子对同一批数据跑多次并做多数投票费用增长过快每句话都调用 API没有做召回统计调用次数先做规则召回再对候选句调用模型使用缓存排查顺序可以按“数据 - 规则 - 模型 - 校验 - 聚合”推进。不要先怀疑模型先确认数据清洗是否正确再确认召回规则是否覆盖了常见表达最后才调整模型参数。8. 从分析原型到生产级应用8.1 学习环境与生产环境的差距本地跑通原型只需要一个 CSV 和 Jupyter Notebook但生产环境需要另外考虑四件事。第一是调度。财报电话会议通常按季度集中发布需要定时任务自动抓取新文本并触发分析流程。可以使用 Airflow、Dagster 或最简单的事件触发函数甚至 GitHub Actions 定时运行脚本。第二是存储。原始文本、模型输出、人工标注结果要分别落库。推荐使用 PostgreSQL 或更轻量的 DuckDB。表结构至少包含transcript_id、sentence_id、analysis_id、model_version、raw_output。第三是监控。每天记录 API 调用量、失败数、Token 消耗、人工复核队列长度。大模型输出质量会随模型版本变化需要记录模型版本号否则无法定位结果漂移。第四是安全合规。不要把会议文本直接发送到不受控的外部模型除非已经获得数据公开授权。企业内部私有化部署模型时还要做好访问控制和日志审计。8.2 引入AI Agent后的自动化扩展当分析需求变多后可以考虑把流程拆成多个 Agent 任务数据采集 Agent定时访问数据源下载新增电话会议 PDF 或文本。文本解析 Agent把 PDF 转成结构化文本并按发言者分句。信号提取 Agent调用大模型输出证据列表。人工复核 Agent把低置信度结果推送给审核队列。报表 Agent生成季度趋势报告并推送到企业微信、钉钉或邮件。每个 Agent 本质上是原来的一个小函数只是加上了状态管理、重试和任务队列。这样做的收益是让不同岗位的人只关注自己负责的环节降低大模型调用出问题时的排查范围。8.3 回归测试与模型更新生产环境不能每次改 prompt 后就直接上线。要把一批已经人工标注好的句子作为回归测试集例如 200 条覆盖五个维度的样本。改 prompt 后先用这批样本跑指标对比只有当准确率和召回率没有显著下降时才进入正式流程。最好把回归测试集成到 CI/CD 中。每次更新 prompt 或切换模型时自动执行一遍回归测试并输出一个对比报告。这个机制对于依赖大模型的分析系统尤其重要因为模型行为并不完全可控。8.4 实践建议清单环境检查Python 版本、依赖包版本、API 地址和模型名称都记录在一个配置文件中。数据检查加载后执行空值、重复项、编码格式检查中文编码统一为 UTF-8。召回检查统计每篇电话会议召回候选句数量并设定上下界低于阈值要告警。模型检查先跑 5 条样例对比输出结构和 prompt 定义是否一致。人工复核对强度为 high 的记录优先复核因为这类记录对结论影响最大。版本记录每次运行都保存模型版本、prompt 版本、运行时间、原始输出。成本优化对同一会议多次分析使用缓存按会议 ID 和模型版本建立唯一键。结果解释所有指标都要附带口径说明避免把“管理层披露”写成“实际提升”。这篇文章的核心思路是把 AI 叙事拆成可以反复计算和验证的信号。从关键词召回到大模型分类再到强度校验每一步都服务于一个目标降低把“嘴上的 AI”误当成“业务里的 AI”的风险。下一步你可以根据自己的数据源继续扩展比如加入股票收益分析、同行业横向对比、管理层语气分析或者用更细粒度的事件抽取模型识别 AI 投资金额和产品发布时间。对刚入门的读者建议先用一两家公司的小数据集跑通完整流程再逐步扩大范围。