这类标题最近在各类信息流里越来越常见:用四位投资大师的AI工具搞了146个达不溜,听起来像是又一个“AI荐股神器”的流量包装。但如果愿意多停留半分钟,会发现这套玩法里真正值得拆解的,不是那个数字,而是一条很具体的技术链路:把格雷厄姆、巴菲特、芒格、费雪各自关心的财务问题,变成四个可以并行工作的Agent智能体,再通过多智能体协作输出一份结构化的财报研究纪要。
本文不打算讨论“146万”是真金白银还是模拟盘,也不打算给任何荐股建议。我想拆解的是一套可以用普通AI工具和开源项目手工复现的研究工作流:从一份年报PDF开始,经过文档解析、财务指标提取、多Agent视角评估,最终生成一份可审计、可追溯的投研笔记。读完这篇文章,你会得到四个可直接改造的Agent提示词模板、一套调用大模型接口的Python骨架,以及一组真实工程里很容易踩中的坑。
先说判断:这类AI投研工具真正降低的不是“预测涨跌”的成本,而是“把一份几十页财报读完、拆开、对照不同大师标准提问”的时间成本。如果方向搞反了,你会得到一个看起来什么都懂、但什么都证明不了的聊天机器人;如果方向对了,它就是一个随时能出工的初级分析师。
1. 这篇文章真正要解决的问题
做基本面和价值投资研究的人,大概率经历过这样的场景:从年报PDF里找净利润、毛利率、经营现金流,翻到附录的合并资产负债表,再去找上一年同期数据对比,最后还要手动整理一份Excel。一份50页以上的财报,完整读完并提炼关键指标,少说半天,多则一两天。
这还只是“把数据读出来”的阶段。真正难的是判断:格雷厄姆会关注安全边际和资产质量,巴菲特会关注护城河和自由现金流,芒格会反过来想“什么会让这家公司变差”,费雪会关心成长空间和管理层。一个人要同时切换这么多视角,很容易被自己的持仓偏见带走。
AI工具在这里解决的是流程问题,不是决策问题。通过Agent智能体,我们可以把“读财报—提取指标—多视角评估—生成纪要”压缩成一条流水线。四个Agent各领一个大师角色,基于同一份财报片段独立输出观点,最后由人来汇总和判断。多智能体在这里的定位不是玄学,而是对“投研团队分工”的模拟:每个角色只负责自己最擅长的问题,再回到同一个事实基础上讨论。
当然,这套方案并不适合所有人。如果你做的是短线交易,或者只看K线和技术指标,本文这套面向财报的Agent工作流对你帮助有限。如果你一没有数据源、二不愿意读输出、三想把AI评分直接变成买入信号,那更要谨慎——AI Agent可以做研究,但决策责任和资金风险最终还是在自己身上。
需要明确一个边界:本文所有代码和提示词,都定位为“研究辅助工具”,不是投资建议,也不构成自动交易信号。明白这一点之后,再看下面这些概念就顺了。
2. 基础概念:AI工具、Agent智能体与多智能体协作
2.1 什么是这里的AI工具
广义上的AI工具很多,但本文真正用到的只有三类。第一类是通用大模型接口,负责理解财报语义、抽取指标、生成分析意见;第二类是文档解析工具,比如微软开源的MarkItDown,可以把PDF、Word、Excel转成Markdown文本,方便后续处理;第三类是数据处理脚本,也就是我们写的Python代码,负责切片、调用、解析JSON、输出汇总表。
很多人以为AI投研是一套现成的神奇软件,装完就能自动赚钱。实际上更靠谱的形态是:把通用AI能力嵌入一条自己能控制的脚本链路。好处是每一层都看得见、改得动、可审计。
2.2 什么是Agent智能体
Agent智能体可以理解为“大模型 + 系统提示词 + 上下文 + 输出格式约束”。普通聊天机器人的使用方式是“你问我答”,而Agent有一个固定的任务边界。比如在本文示例中,格雷厄姆Agent永远只负责财务稳健性和安全边际,用户给它的财报片段就是它的全部世界。
一个完整的Agent,通常包含以下要素:
- 角色设定:告诉模型它是谁,比如“你是一位极度保守的价值投资分析师”。
- 任务目标:让它从财报片段里提取哪些指标、回答什么问题。
- 上下文:给它喂入的财报内容,也就是它做判断的依据。
- 输出约束:要求它返回结构化的JSON,而不是一段自由发挥的文字。
- 能力边界:明确告诉它“不要编造未提供的数据”,这个约束在金融场景里尤其重要。
2.3 多智能体协作为什么比单Agent更合适
单Agent也能读财报,但它容易陷入两种极端:要么面面俱到但深度不够,要么被一种投资风格带偏。多智能体的思路是把“多视角”拆成“多角色”,让每个Agent只回答自己最擅长的问题,最后再把结论拼起来。
| 对比维度 | 单 Agent | 多智能体协作 |
|---|---|---|
| 视角覆盖 | 依赖同一个系统提示词,视角容易单一 | 每个角色独立,覆盖不同分析框架 |
| 输出稳定性 | 问题一多,回答容易跑偏 | 每个Agent只专注一个职责,更容易输出结构化结果 |
| 人工复核 | 需要从头检查一整段长文 | 可以按Agent维度分别检查,定位更精准 |
| 成本 | 一次提问即可 | 多次调用,成本更高,需要缓存和并发优化 |
本文讨论的多智能体,不是多智能体强化学习里的“训练协作”,而是更工程化的提示词编排:把不同专家角色放进同一套研究流程。
2.4 四位投资大师的视角如何映射到Agent
四位投资大师的组合并不神秘,关键是每个角色关注的点要足够聚焦。
| 投资大师 | 代表理念 | 对财报的关注点 | Agent职责 |
|---|---|---|---|
| 格雷厄姆 | 安全边际、资产质量 | 资产负债率、有息负债、流动资产、经营现金流 | 评估公司财务稳健性,识别隐藏风险 |
| 巴菲特 | 护城河、自由现金流、ROE | 毛利率、ROE、资本开支、现金流质量 | 判断商业模式是否有持续回报能力 |
| 芒格 | 逆向思维、多元模型 | 盈利波动、管理层激励、行业竞争 | 列出可能导致公司变差的三个因素 |
| 费雪 | 成长股、管理层、市场空间 | 营收增速、研发投入、管理层战略 | 判断成长质量,而不只看当期利润 |
这套映射的价值在于,它把四位大师比较抽象的投资哲学,变成了可执行的提示词和结构化字段。后面写代码时,每个Agent都会遵循这个职责边界。
3. 环境准备与前置条件
3.1 本地环境要求
本文示例使用Python实现,建议使用Python 3.9以上版本,操作系统不限。核心依赖只有四个:requests、markitdown、pandas、python-dotenv。其中requests用于调用大模型接口,markitdown用于解析PDF,pandas用于最后整理评分表,python-dotenv用于读取环境变量。
不依赖特定深度学习框架,也不需要本地显卡。大模型推理发生在服务端,本地只做请求、解析和汇总。
3.2 创建虚拟环境并安装依赖
先创建一个干净的虚拟环境,避免和系统Python环境互相污染。
python -m venv .venv source .venv/bin/activate pip install --upgrade pip pip install requests markitdown pandas python-dotenv如果你的操作系统是Windows,激活虚拟环境的命令需要换成:
.venv\Scripts\activate依赖安装完成后,可以用一行命令验证MarkItDown是否可用。这里不写死具体版本,因为这类工具更新较快,当前安装版本以实际情况为准。
markitdown --help如果命令找不到,也可以直接在Python里验证:
from markitdown import MarkItDown print("markitdown import ok")3.3 准备大模型API密钥与数据源
大模型接口推荐使用OpenAI兼容格式的服务,这样可以比较方便地在不同供应商之间切换。配置项包括三部分:API地址、API Key、模型名称。
在项目根目录创建.env文件:
OPENAI_API_KEY=你的密钥 OPENAI_BASE_URL=https://api.openai.com/v1 MODEL_NAME=gpt-4o-mini安全提醒:密钥不要写死在Python代码里,也不要提交到Git仓库。.env文件要加入.gitignore,生产环境下密钥应该放在密钥管理系统中,并以最小权限原则配置。
财报数据建议从公开合规渠道获取。A股年报可以从巨潮资讯网、上交所或深交所官网下载,美股年报可以从SEC官网的EDGAR系统获取。本文不介绍任何破解、抓取反爬或绕过访问限制的方法,只建议下载官方公开PDF,并遵守平台规则。
3.4 准备一份测试财报
新手可以先拿一份自己熟悉公司的年报PDF做测试。文件不需要太大,建议先用“管理层讨论与分析”和“合并财务报表”这两个章节,因为它们包含最多关键财务信息。
把PDF放在项目目录下,命名为annual_report.pdf。后面的示例代码都默认读取这个文件。
4. 核心流程拆解:从PDF到多Agent投研纪要
整个流程可以拆成六步,每一步都有明确的输入和输出。
4.1 第一步:PDF转Markdown
为什么要先转Markdown?因为PDF格式里的表格、页眉、页脚对模型不友好。MarkItDown这类工具会把PDF转换成带层级结构的文本,后续切片和提问都更容易。
这一步常见的问题是扫描版财报没有文本层,解析出来是乱码或空白。如果遇到这种情况,说明PDF本身是扫描图片,需要先做OCR,这不在本文最小示例范围内。
4.2 第二步:清理与切片
财报全文通常几万字,直接塞给大模型会超出上下文窗口。需要按章节或按固定长度切成多个片段。为了控制成本,也可以先截取包含关键指标的部分。
最简单的方式是保留前12000字符或前几个章节。真正生产环境中,建议结合检索方式只提取相关片段。
4.3 第三步:定义四个Agent
每个Agent由系统提示词和用户指令组成。系统提示词承担角色设定,用户指令说明具体任务。为了让输出稳定,要求每个Agent返回固定结构的JSON。
4.4 第四步:并行调用
四个Agent相互独立,理论上可以并行请求。但如果使用免费或低配额API,建议串行调用,避免触发限流。每得到一个Agent回复,先把原始文本存下来,方便后续排查。
4.5 第五步:汇总评分
拿到四份JSON后,分别提取score、risks、highlights和missing_data,生成一张汇总表。这张表的作用不是直接给“买入”或“卖出”结论,而是告诉人应该进一步研究哪些风险点。
4.6 第六步:人工复核
最后一步永远是人。每个Agent都可能产生幻觉,哪怕提示词写了“不要编造”,模型仍有概率输出模型未提供的数据。复核时重点看两个地方:一是Agent提取的财务数字是否与原文一致,二是对风险的描述有没有超出财报片段本身。
这个流程的核心价值是:它把投研中“最花时间的阅读和整理”交给工具,同时把“最重要的判断”留给人。
5. 完整示例:搭建一个可运行的四Agent投研工作流
下面是一个最小可运行示例。它不追求完整工程化,而是让读者先跑通全流程。
5.1 用MarkItDown解析年报PDF
在项目目录下创建research_agent.py,先实现PDF解析部分。
# research_agent.py import os from dotenv import load_dotenv load_dotenv() from markitdown import MarkItDown def pdf_to_markdown(pdf_path: str) -> str: md = MarkItDown() result = md.convert(pdf_path) return result.text_content需要说明的是,不同版本的MarkItDown在API细节上可能有差异。如果导入失败,建议查看当前安装包的官方文档,或者退回到命令行方式:
markitdown annual_report.pdf -o annual_report.md5.2 定义四个大师Agent的提示词
接着在同一个文件中定义四个Agent。为了避免模型编造数字,每个Agent的指令里都加上“只基于给定财报片段,找不到的数据标记为未披露”。
AGENTS = { "graham": { "name": "格雷厄姆-安全边际Agent", "system": "你是一位极度保守的价值投资分析师,擅长检查资产质量、负债结构和现金流风险。你只基于用户提供的财报片段做分析,绝不推测未提供的数据。", "instruction": ( "请从财报片段中提取以下指标:总资产、总负债、有息负债、流动资产、摊销费用、净利润、经营现金流。" "然后按格雷厄姆的安全边际框架评估财务稳健性。" "输出JSON格式,不要输出其他文字。\n" '{"score": 0到100的整数, "highlights": ["亮点1", "亮点2"], ' '"risks": ["风险1", "风险2"], "missing_data": ["未披露项1"]}' ), }, "buffet": { "name": "巴菲特-护城河Agent", "system": "你是一位长期价值投资者,擅长从商业模式和现金流质量判断企业是否具备可持续竞争优势。你只基于给定财报片段下结论。", "instruction": ( "请从财报片段中提取:毛利率、ROE、经营现金流、资本开支、自由现金流。" "重点回答:这家公司靠什么赚钱?这个模式是否容易被复制?现金流质量如何?" "输出JSON格式,不要输出其他文字。\n" '{"score": 0到100的整数, "highlights": [], "risks": [], "missing_data": []}' ), }, "munger": { "name": "芒格-逆向思维Agent", "system": "你是一位善于逆向思考的投资哲学家。你的任务不是找买入理由,而是找出可能让公司变差的因素。你只基于给定财报片段分析。", "instruction": ( "请从财报片段中寻找负面信号,包括但不限于:应收账款异常增长、存货积压、经营现金流与利润背离、" "管理层薪酬异常、关联交易、商誉减值风险。列出最可能导致公司价值变差的三个因素。" "输出JSON格式,不要输出其他文字。\n" '{"score": 0到100的整数, "highlights": [], "risks": ["因素1", "因素2", "因素3"], "missing_data": []}' ), }, "fisher": { "name": "费雪-成长质量Agent", "system": "你是一位成长股分析师,注重营收增长质量、研发投入和管理层执行力。你只基于给定财报片段做判断。", "instruction": ( "请从财报片段中提取:营收增速、研发费用、毛利率变化、管理层讨论中的战略方向。" "判断这家公司的成长是来自行业东风、产品竞争力,还是会计调整。" "输出JSON格式,不要输出其他文字。\n" '{"score": 0到100的整数, "highlights": [], "risks": [], "missing_data": []}' ), }, }这段代码的核心不是提示词本身,而是“角色隔离”。每个Agent只回答自己职责范围内的问题,避免一个模型同时扮演四个角色时出现“既要又不要”的矛盾。
5.3 调用OpenAI兼容接口
下一步实现通用调用函数。使用OpenAI兼容的chat completions接口,这样可以对接不同模型服务商。
import requests import json import re MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4o-mini") API_KEY = os.getenv("OPENAI_API_KEY") BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1") def call_llm(system_prompt: str, user_content: str) -> str: if not API_KEY: raise RuntimeError("未找到 OPENAI_API_KEY,请检查 .env 文件") url = f"{BASE_URL.rstrip('/')}/chat/completions" headers = { "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json", } payload = { "model": MODEL_NAME, "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_content}, ], "temperature": 0.2, } resp = requests.post(url, headers=headers, json=payload, timeout=120) resp.raise_for_status() data = resp.json() return data["choices"][0]["message"]["content"]这里的temperature设置为0.2,目的是让输出更稳定,尽可能贴近同一套分析逻辑。
5.4 解析Agent返回的JSON
大模型经常在JSON外面套一层Markdown代码块,或者夹带解释文字,所以需要一段清洗函数。
def parse_agent_json(raw_text: str) -> dict: text = raw_text.strip() text = re.sub(r"^```json\s*", "", text) text = re.sub(r"```$", "", text).strip() try: return json.loads(text) except json.JSONDecodeError: return { "score": None, "highlights": [], "risks": [f"JSON解析失败,原始内容:{raw_text[:200]}"], "missing_data": [], }如果JSON解析失败,不要直接丢弃原始内容,要把它写入结果,方便后续人工检查。
5.5 主流程:四个Agent依次分析同一份财报
最后写主流程,把前面所有部分串起来。
def run_research(pdf_path: str, max_chars: int = 12000): print(f"开始解析PDF:{pdf_path}") markdown_text = pdf_to_markdown(pdf_path) print("对财报内容进行切片...") context = markdown_text[:max_chars] results = {} for agent_key, agent_cfg in AGENTS.items(): print(f"正在调用:{agent_cfg['name']}") user_content = ( f"{agent_cfg['instruction']}\n\n" f"===财报片段 begin===\n{context}\n===财报片段 end===" ) raw = call_llm(agent_cfg["system"], user_content) results[agent_key] = { "raw": raw, "parsed": parse_agent_json(raw), } return results if __name__ == "__main__": res = run_research("annual_report.pdf") with open("agent_results.json", "w", encoding="utf-8") as f: json.dump(res, f, ensure_ascii=False, indent=2) print("分析完成,结果已保存到 agent_results.json")运行方式:
python research_agent.py这个最小示例没有做并发调用,也没有做完整的章节检索,但已经足够让读者理解整个多智能体协作的数据流向。
5.6 生成一张可视化的汇总表
为了让结果更易读,下面用pandas生成汇总表。注意这里的数字只是结构示意,不代表任何真实公司结论。
import pandas as pd def build_summary_table(results: dict) -> pd.DataFrame: rows = [] for agent_key, result in results.items(): parsed = result["parsed"] rows.append({ "agent": agent_key, "score": parsed.get("score"), "highlights": ";".join(parsed.get("highlights", [])), "risks": ";".join(parsed.get("risks", [])), "missing_data": ";".join(parsed.get("missing_data", [])), }) return pd.DataFrame(rows)在__main__里调用这个函数并输出CSV:
table = build_summary_table(res) table.to_csv("agent_summary.csv", index=False, encoding="utf-8-sig")生成CSV后,可以用Excel打开,按风险列排序,优先研究所有Agent共同提到的风险点。
6. 运行结果与效果验证
6.1 预期输出
正常运行时,控制台会依次输出四个Agent的调用日志,最后在项目目录下生成两个文件:agent_results.json和agent_summary.csv。
agent_results.json的结构大致如下:
{ "graham": { "raw": "### 财务报表分析\n\n{...}", "parsed": { "score": 71, "highlights": ["现金储备充足", "有息负债率较低"], "risks": ["经营性现金流连续两年下滑"], "missing_data": ["分业务毛利率未披露"] } } }注意:上面只是格式示例,真实结果取决于你输入的财报片段和大模型回复。不要把这个样例当成任何公司的真实分析结论。
6.2 如何判断流程是否成功
判断标准不是“四个Agent都说好”,而是下面几点:
- 四个Agent都返回了可解析的JSON,没有大量解析失败项。
- 每个Agent提取的财务指标与财报片段中的数字一致。
- 四个Agent的风险描述出现交集,或者互补,而不是互相矛盾。
- 缺少数据时,Agent明确写出
missing_data,而不是凭空补一个数字。
如果出现“成功解析但完全没法用”的情况,优先怀疑财报切片是否选错了章节。比如只截取了公司简介,却没有截取财务数据,结果自然没有参考价值。
6.3 运行失败时先看哪里
运行失败时,按下面顺序排查:
- 是否完成
.env配置,API_KEY是否存在; - 网络到模型服务端是否通畅;
- PDF是否有文本层;
- 是否超出模型上下文窗口;
- 是否触发了API限流。
不要把大模型当成PDF解析器,也不要把财报PDF直接作为用户消息发给模型。先转成Markdown,再切片,是最稳妥的路径。
7. 常见问题与排查思路
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| PDF解析后乱码 | 扫描版PDF没有文本层 | 打开PDF复制文字测试 | 先做OCR,或换用可复制的PDF版本 |
| 请求返回401 | API密钥错误或未设置 | 检查.env和当前会话环境变量 | 重新配置密钥,不要硬编码在代码里 |
| 请求返回429 | 触发了限流 | 查看响应头中的限流信息 | 降低并发,增加重试和休眠 |
| 上下文超限 | 财报切片过大 | 统计切片字符数和token数 | 缩小max_chars或按章节检索 |
| 返回内容无法解析JSON | 模型输出了额外文字或代码块 | 打印原始返回内容 | 增加清洗函数,或使用function calling |
| Agent编造财务数字 | 提示词约束不够强 | 检查引用是否能在原文找到 | 增加“只基于给定内容”并让输出引用段落编号 |
| 四个Agent结论冲突 | 切片内容侧重不同 | 检查每个Agent拿到的上下文是否一致 | 固定同一份切片,记录上下文版本 |
还有一个很容易被忽略的问题:模型输出中的财务数字可能来自训练语料记忆,而不是给定财报。这很危险。解决方式是在提示词里要求Agent必须从输入片段中引用原文,并说明“如果片段中没有该指标,请写入missing_data”。即使这样,仍然需要在人工复核阶段抽查几个关键数字。
8. 多Agent投研工作流的最佳实践与工程建议
8.1 让输出可追溯
不要只让Agent输出结论,要让它们输出依据。可以在提示词中增加“引用原文片段”或“标注可能位于财报哪个章节”。这样人工复核时能快速定位。
如果调用的模型支持function calling,优先使用结构化输出功能,这会显著减少JSON解析问题。
8.2 缓存与成本控制
同一份财报不需要反复分析。可以把每个Agent的输入提示词哈希后作为缓存键,把输出存入本地JSON或数据库。这样第二次运行时直接读缓存,大幅节省API费用。
另外,可以先让便宜的模型做初筛,用更强大的模型只复检“风险交集”部分。多智能体的成本不是单次调用成本乘以四那么简单,上下文越长,成本增长越快,所以切片策略要精心设计。
8.3 数据合规与安全边界
财报数据只从官方公开渠道获取,不要使用任何破解或绕过访问限制的方法。API密钥要使用最小权限,只配置必要权限,并在泄露风险出现时及时轮换。
这个工具的定位是辅助研究,不应直接连接交易系统。不要写“Agent输出分数大于80就自动买入”这种代码,这不是技术问题,而是风险控制问题。研究输出到交易执行之间,一定要留出人工判断的缓冲带。
8.4 提示词也要做版本管理
很多人改提示词只靠复制文件,很快就分不清哪个版本效果更好。建议把提示词单独放到prompts/目录下,用Git管理版本。每次调整都记录改动原因,并把不同版本在同一个测试财报上的输出结果留档。
8.5 不要迷信评分
四个Agent的score是模型基于片段给出的相对估计,不是金融模型的定量评级。不同厂商、不同模型、不同温度参数下,分数都会变化。建议把score当作研究排序参考,而不是绝对结论。真正值得关注的是risks列表里的交叉重复项。
8.6 人工复核清单
人工复核不是重新读一遍财报,而是对照下面几条检查:
- 关键财务指标是否有明显偏差;
- 风险描述是否与片段一致;
- Agent是否漏掉了明显问题;
- 四个Agent的结论是否有相互矛盾但未被识别的地方。
9. 总结与后续学习方向
这套四Agent投研工作流的核心,是把四位投资大师的分析框架转成可执行、可审计的提示词和代码。你不需要真的掌握格雷厄姆每一句话,但你会被迫思考“格雷厄姆到底关心什么”“巴菲特为什么强调自由现金流”,这个过程本身就有价值。
如果你只想拿来跑通,建议今天就用一份公开年报PDF测试;如果你想继续深入,可以做三件事:第一,用开源模型替代商业API,在本地跑通全流程,让数据不出内网;第二,把固定截断改成按章节检索,结合向量数据库做RAG;第三,把多Agent结果接入回测系统,但要警惕历史数据过拟合。
还有一点必须提醒:AI工具能帮你更快地读完一份财报,但它读不到财报之外的管理层诚信、行业格局变化和黑天鹅事件。任何把Agent评分当作唯一决策依据的做法,都是在放大风险而不是控制风险。
如果你期望的是“自动抓一个买入信号然后躺赚”,这套方案会让你失望;如果你想要的是让研究过程更快、更透明、更可追溯,那现在就可以从一份年报PDF开始试跑。