AI Agent驱动投研流水线:架构拆解、实现与踩坑实录
2026/9/9 4:36:34 网站建设 项目流程

过去半年我一直在做一件事:把一套原本靠研究员手动操作的工作流,改造成由AI自主驱动的投研流水线。这套体系的输入是财报、公告、券商研报、舆情数据,输出是一份结构化的公司分析纪要、异动归因说明和风险提示清单。整个过程涉及AI大模型、AI Agent、知识库、自动化调度,也踩了无数个数据坑,今天这篇文章不聊空概念,直接拆架构、讲实现、放代码,把我在落地过程中怎么设计Agent、怎么控制幻觉、怎么处理数据污染的经验全写出来。

这套东西适合谁看?如果你正在做投研相关的工作,无论是买方研究员、卖方分析师,还是做量化策略的人,并且已经受够了每天打开几十份PDF、几百页财报手动摘数,那么这篇内容会对你有直接帮助。如果你本身就是工程师,想用AI Agent重构一条业务流程,我的踩坑记录和代码框架同样可以参考。

1. AI自主驱动投研,本质上是把“脏活累活”自动化

先说清楚一个前提:AI自主驱动不是要替代研究员的判断,而是把“找信息、读信息、整理信息、校验信息”这类高耗时、低创造性的事情全部自动化。研究员真正值钱的地方是建立分析框架、做关键假设、下投资结论,这些短时间很难被替代,但信息处理环节确实可以让大模型加Agent架构全权接管。

1.1 投研的脏活累活到底有多少

我在上面提到的这套体系之前,团队里的研究员每天的工作节奏是这样的:早上先打开行情软件看持仓和自选股的隔夜新闻,然后开始读券商研报,重点看行业景气度、目标价调整、盈利预测变化;下午要参读上市公司公告,特别是定期报告、业绩预告、重大合同;晚上还要整理自己的调研纪要,把沟通内容结构化归档。

这个流程最大的问题不是某个环节难,而是信息量太大了。一家中型券商每个交易日发布的个股研报加行业报告动辄几百份,一个研究员覆盖三五十只股票,一天下来光读报告就要四五个小时。更麻烦的是,数据口径不统一,不同券商对同一家公司的营收预测可能差好几个版本,财务数据在公告里是A口径,在研报里又变成B口径,人工核对一遍极其费劲。

我最早尝试用传统方案去优化,比如写爬虫自动抓公告、写正则从PDF里抽财务数字,结果投入产出比很差。公告版式和研报模板五花八门,正则规则写了几百条还是经常漏数据。后来换成了大模型加Agent的方式,把“抽取规则”改成“语义理解”,从根上绕开了规则维护的泥潭。

1.2 AI自主驱动与传统量化投研的本质差别

很多人一听到AI投研,第一反应是量化交易。这里必须做个区分:传统量化投研的核心是因子模型和统计套利,数据必须是干净的数值型字段,然后跑回归、算相关性,整个过程是全自动的,但它不处理语义信息,读不懂“公司加大海外市场拓展力度”这句话对基本面意味着什么。

AI自主驱动投研的核心在自然语言理解和任务拆解。它不是让模型去预测股价,而是让模型去扮演一个“研究助理”:你说“帮我看看某消费电子公司最近三个季度的经营质量是否改善”,它会自动拆解成几个子任务——读取历史财报、提取核心财务指标、搜索最近研报观点、对比行业趋势、输出结构化结论。

打个比方,传统量化是在处理已经变成“Excel表格”的世界,AI Agent要处理的是“一堆文档”的世界,它自己把文档变成表格,还能顺带给出解读。这正是两个技术路线最大的分水岭。

1.3 这套思路适合谁,不适合谁

先说适合谁。第一类是券商、基金、资管机构里做投研支持的人,每天和研报、财报打交道,最需要自动化工具;第二类是负责信评风控的人员,需要持续跟踪发债主体的经营变化,AI Agent非常适合做持续监控;第三类是对上市公司做深度研究的个人投资者,虽然没有机构那么多数据源,但同样可以利用公开财报和公告搭建一个轻量级的分析系统。

不适合谁?如果你以为搭建完AI投研流程就可以自动选股、自动稳赚,那我劝你趁早放弃。大模型对未来的预测能力并没有好到可以替代投资决策,它擅长的是处理“已有信息”,而不是创造“新的有效信息”。把它定位成一个超级研究助理,而不是财富密码,才是正确的打开方式。

2. 系统架构与核心Agent拆解

整体架构我按四层来设计:数据层、模型层、Agent层、应用层。这个分层不是拍脑袋定的,是在实际运行中反复调整后才定下来的。早期我试图把所有逻辑塞到一个Agent里,结果任务一长就乱,后来才拆成多Agent协作的模式,稳定性明显提升。

2.1 整体分层:从数据源到用户界面的完整链条

数据层负责统一接入各类数据源。我这里主要接了三类源:一是上市公司公告和定期报告,来源是公开的交易所信息披露网站;二是券商研报,通过合规渠道购买或授权获取;三是行情和舆情数据,包括每日股价、成交量、新闻标题和热度指数。

数据层做四件事:采集、解析、清洗、入库。采集用调度框架定时拉取,解析现在全部交给大模型做文档结构化,清洗则是把明显错误的数据剔除,最后统一落到数仓里,公司维度、财务周期维度、事件类型维度都提前建好索引。

模型层相对简单,主要管理大模型API和私有化部署的小模型。大模型负责理解和生成,小模型负责做分类打标、敏感信息识别、格式校验这些轻量任务。把任务拆到不同规模的模型上,是为了成本可控,如果所有请求都走大模型,一个月下来API账单会非常惊人。

Agent层是整个系统的核心,我拆出了五个角色:信息采集Agent、研报解析Agent、逻辑推理Agent、风险校验Agent、报告生成Agent。每个Agent只做一件事,通过一个协调器串联整个流程。下面的表格整理了各Agent的职责、输入输出和关键模型。

Agent名称核心职责主要输入主要输出推荐模型档位
信息采集Agent抓取公告、研报、舆情,做去重与更新股票池清单、调度时间表结构化原始事实表小模型+规则
研报解析Agent从PDF和网页中抽取观点、预测、评级原始研报文件研报观点三元组大模型
逻辑推理Agent基于事实表做指标计算和趋势判断事实表、历史数据分析结论与依据链大模型
风险校验Agent检查前后数据矛盾、异常波动、模型幻觉分析结论草稿、原始事实风险提示清单大模型+小模型
报告生成Agent将结论组装成结构化投研纪要结论图、风险清单Markdown投研纪要大模型

2.2 多Agent协作是如何避免“一个Agent干到底”的混乱

最开始我也用过单Agent模式:一个Prompt里面包含所有工具和上下文,让它自己决定先查财报、再读研报、最后写总结。问题非常明显,任务一旦超过三步,模型就会“迷路”,要么漏掉关键步骤,要么把工具调用顺序搞乱,而且排查问题时完全不知道卡在哪一步。

拆成多Agent之后,每个Agent负责一段独立操作,边界清晰,可以单独测试和替换。举个例子,研报解析Agent只做“从研报中抽取观点”,它对输入输出有严格规范:输入是PDF解析后的纯文本,输出是标准JSON,格式为观点ID、来源研报、目标公司、观点类别、原文摘要、发布时间。

这样的设计带来三个直接好处。第一,任何一个模块出问题都可以单独重跑,不需要把整个流水线全部重来;第二,可以针对每个Agent单独调优提示词,效果迭代很快;第三,关键路径上可以插入人工审批节点,比如在风险校验Agent发现问题时强制转人工,系统是可解释的。

2.3 大模型和小模型的选型逻辑

选型的时候,我的原则是:能用小模型解决的绝不上大模型。大模型目前仍然是按Token计费,一篇研报解析几千到上万Token,全流程跑下来成本不低。而有些任务是典型的分类问题,比如判断一篇公告属于“重大合同”还是“关联交易”,用微调过的小模型效果又好又便宜。

大模型我主要用在三个场景:研报观点的语义压缩、跨文档信息聚合、生成分析结论。这三个场景对语言理解和逻辑推理要求高,小模型确实顶不上。比如跨文档信息聚合,需要把三份不同券商对同一公司的业绩预测整合到一个表格里,并标注预测分位数,没有强大的语义对齐能力是做不好的。

小模型我用在了公告分类、敏感信息识别、格式校验、相似文章去重。这些任务数据量大、逻辑固定,用小模型跑批量非常划算。实测下来,公告四分类的F1分数超过0.97,和之前用大模型跑的效果相差不大,但成本只有大模型方案的五分之一左右。

2.4 知识库和向量检索怎么用起来

投研系统里有一个很头疼的问题:同行业公司、历史案例、产业链关系这类知识是高度结构化的,直接塞进Prompt既浪费Token又干扰模型注意力。我采用“知识库+向量检索”的方式,把这类沉淀知识独立放出去,需要时再检索回来。

具体做法是把近三年的行业深度报告、公司深度报告、产业链图谱等文档切块后做向量化,存到向量数据库里。每当逻辑推理Agent要分析一家公司时,会先检索这家公司所属行业近三个月的深度报告片段,作为分析背景注入Prompt。

这一步非常关键。它让模型在回答“这家公司毛利率下降的主要原因”时,不只是盯着财务数据本身,还能参考行业报告中的竞争格局、原材料价格走势等上下文信息。实际验证下来,加了RAG检索之后,分析结论的完整性和深度都有明显提升,引用依据也从“感觉”变成了可追溯的源文档。

3. 从0到1实现一个投研分析智能体

这一部分直接上实操。我以一个真实场景为例:分析某消费电子公司的季度经营质量。整条流水线跑一遍,你会看到数据怎么抽、Agent怎么编排、Prompt怎么设计、结果怎么校验。

3.1 第一步:把非结构化PDF变成结构化JSON

投研数据90%以上都是半结构化或纯文本,第一步永远是把PDF和网页变成结构化字段。我最早用规则加正则,后来换成大模型抽取,准确率和通用性都上来了。

这里给出一个财务指标抽取的Prompt模板,简单但非常实用:

SYSTEM_PROMPT_FIN_EXTRACT = """ 你是一个金融文档结构化助手。用户会给你一段上市公司财务报告的正文内容, 请从这段内容中抽取以下指标,并以JSON格式返回字段: - company_name: 公司全称 - period: 报告期,格式YYYY-MM-DD - revenue: 营业总收入(元) - revenue_yoy: 营业总收入同比增速(百分比) - net_profit: 归母净利润(元) - net_profit_yoy: 归母净利润同比增速(百分比) - gross_margin: 毛利率(百分比) - operating_cashflow: 经营活动现金流净额(元) 抽取原则: 1. 只保留报告原文出现过的数字,缺失字段填空字符串。 2. 单位必须换算成“元”,如果是“万元”要乘以10000。 3. 同比增速保留百分号前的数字,比如5.3%返回5.3。 4. 如果有调整后数据,优先采用调整后口径。 5. 不要填写任何推断值,无法确定就返回空。 """

这个Prompt设计的关键在于“边界约束”。如果不限定单位、不加口径说明,模型会随机生成“营收大约50亿”这种模糊值。加了“只保留原文出现过的数字”之后,抽取结果基本可以做到零幻觉。实测对三种PDF版式不同的财报进行抽取,关键财务字段的准确率在95%以上。

抽取完成之后再写一段简单的数据校验代码,因为大模型偶尔还是会把“元”和“万元”搞混。校验逻辑就是用上期数据加同比增速反推本期营收,计算误差,如果超过1%就标记为疑似异常。

def validate_fin_extract(extract, prev_fin): revenue_implied = prev_fin["revenue"] * (1 + extract["revenue_yoy"] / 100) diff = abs(revenue_implied - extract["revenue"]) / extract["revenue"] return diff < 0.01

我把这个函数放在数据入库之前,一旦校验失败,数据不会直接写进事实表,而是进入待人工复核队列。这个复核机制帮我挡住了无数次数据污染事故。

3.2 第二步:Agent编排,让协调器调度多个模型协作

有了结构化数据,接下来就是Agent编排。这里我推荐用类似于LangGraph的方式,把Agent和工具节点定义成有向图,而不是让模型自由调用。图结构的好处是执行路径固定、方便追踪、也能控制成本。

下面是一个精简的协调器代码示例,展示多Agent怎么协作:

from langgraph.graph import StateGraph, END def collect_node(state): # 拉取指定公司的公告、研报、行情数据 facts = collection_agent.run(state["company_code"], state["period"]) return {"facts": facts} def analyze_node(state): # 逻辑推理Agent,基于事实表计算指标趋势 conclusion = reasoning_agent.run( facts=state["facts"], industry_context=rag_retrieve(state["company_code"]) ) return {"draft": conclusion} def risk_node(state): # 风险校验Agent,检查矛盾数据与模型幻觉 risk_report = risk_checker.run(draft=state["draft"], facts=state["facts"]) if risk_report.has_error: state["need_manual_review"] = True return {"risk": risk_report} def generate_node(state): # 报告生成Agent,组装最终输出 report = report_agent.run( draft=state["draft"], risk=state["risk"] ) return {"report": report} graph = StateGraph() graph.add_node("collect", collect_node) graph.add_node("analyze", analyze_node) graph.add_node("risk", risk_node) graph.add_node("generate", generate_node) graph.set_entry_point("collect") graph.add_edge("collect", "analyze") graph.add_edge("analyze", "risk") graph.add_conditional_edges("risk", lambda s: "manual" if s.get("need_manual_review") else "generate") graph.add_edge("generate", END)

这套编排写完之后,整条流程变得非常直观:采集 -> 分析 -> 校验 -> 生成,出了状态错误只需要看图上卡在哪个节点。这里的“图”是执行流程,不是用来展示的示意图,工程上调试极其方便。

3.3 第三步:提示词如何设计才能让分析结论“有依据”

Agent跑起来的核心在于提示词。我在设计分析提示词时特别强调“依据链”,要求模型每个结论都要标注数据来源和推算逻辑,不能凭空下结论。以下是我在逻辑推理Agent里用的一套核心约束:

你是一位研究员助理,请根据提供的facts数据,对公司的经营质量进行分析。要求:1)每个判断必须引用facts的字段或行业上下文,说明判断依据;2)对于趋势变化,要用至少两个连续期的数据进行对比;3)不要给出投资建议,只做事实描述和逻辑归因;4)如果数据不足,明确说“暂无法判断”,不要推测。

这套约束的核心是“不让它做预测”。模型一旦被要求闭嘴承认不知道,输出质量反而会更好。实测在没有约束前,模型经常编造“未来有望保持增长”这类空话,加了约束之后,它学会了回答“当前数据无法判断未来趋势”。

报告生成Agent的提示词则更偏格式控制,我会直接给出输出模板,要求模型把结论和依据填到指定位置。这样做的好处是最终生成的纪要风格统一,后续人工复核也更轻松。

3.4 第四步:人工复核节点设计,AI负责跑量,人负责把关

再好的Agent也会犯错,人工复核节点绝对不能省。我在流程里设计了两处强制人工复核:第一处是数据抽取异常时,也就是上文的校验函数报错;第二处是风险校验Agent检测到“高置信度矛盾”时。

高置信度矛盾是指两个信息源在同一指标上差异超过阈值,比如两份研报对同一家公司下一财年的营收预测,一个说增长20%,一个说下降5%,两者相差过大。这时候模型会输出一条风险提示,并建议人工复核。系统不会自动把这份矛盾信息写进报告,而是先把任务挂起,等人工介入。

这样做的好处是我心里一直有底——AI只是帮我初步处理了95%的常规工作,剩余5%最需要专业判断的部分仍然握在研究员手里。从效率和风险两个维度看,这是目前最合理的平衡点。

4. 踩坑记录与问题排查,这些是普通文档不会写的内容

这套系统上线到现在,踩过的坑比写过的代码还多。下面整理几个最典型的疑难问题,以及我的排查思路和解决方案,希望你能绕开这些点。

4.1 数据污染:比模型幻觉更致命的坑

所有用大模型做数据抽取的人都会担心幻觉,但实际跑下来我发现,数据污染比幻觉更常见也更危险。什么情况算数据污染?同一指标在不同文档里口径不同,比如A公告的“营业收入”是含税口径,B研报用的是不含税口径,抽出来之后直接对比,得出来的趋势判断就是错的。

排查思路说起来简单:入库之前必须做口径一致性校验,至少要在同一份分析报告里保证所有数据点来自同一口径。我的做法是为数据字典增加“口径标签”字段,并在抽取阶段让模型输出口径说明,入库时按口径分组存储。计算趋势时,只用同口径的数据点。

还有一个容易忽略的污染来源是文档版本。同一份季报可能有修订版,如果采集Agent把旧版本和新版本都抽了一遍,就会形成两条看似矛盾的数据。解决办法是为每个公告建立唯一编码,按修订时间倒序保留最新版本。

4.2 模型幻觉怎么控制,我的三层防线

应对模型幻觉我搭了三层防线。第一层是在提示词里强制“找不到就写找不到”,禁止模型自己补全。第二层是通过事实校验Agent做二次检查,让它把输出里的每个关键数字与原始文本比对,不一致就标记。第三层是人工复核节点,专门兜底前两层没拦住的问题。

三层防线建立之后,严重幻觉的发生率从早期的百分之几降到了千分之一左右。这个比率我认为是可以接受的,因为投资分析本身就有不确定性,只要每次错误都有标记、能追溯,风险就可控。

4.3 跑批性能与API成本怎么平衡

初期我所有任务都走大模型API,一个月跑下来成本非常惊人。后来做了三件事:一是把公告分类、格式校验等高频简单任务全部切到小模型,二是把研报解析做成了“只做增量”,不是每天全量重新解析所有研报,只有新入库的才处理;三是合理设置批量大小和并发数,避免频繁触发API限流。

这里给一个参考配置:并发数设置为10,每批请求间隔0.5秒,大模型窗口设为32K上下文,单次解析控制在6000 Token以内。这样既能保证处理速度,又能让成本按月稳定在一个可接受的水平。

从效果上看,调整后每天的解析量翻了接近一倍,成本反而下降了三成。如果你在建类似系统,我建议你每周看一次各Agent的Token消耗分布,主动优化那些占比高但价值低的调用。

4.4 Agent任务跑偏的典型症状与对策

Agent跑偏最常见的症状是“该做的没做,不该做的全做了”。比如你让它分析经营质量,它却把重点放在股价涨跌上;或者你让它只用一个行业的数据,它却把宏观经济数据也拉了进来。

排查方法非常粗暴但有效:打开执行日志,逐节点检查输入输出。我会在每次任务结束后打印一个JSON摘要,包含每个节点花了多少时间、调用了哪些工具、模型返回的原始输出是什么。只要看一眼JSON摘要,就能快速定位跑偏发生在哪一步。

对策有两种。第一种是给Agent加上“行动边界”:在系统提示词里明确哪些操作允许、哪些操作禁止,甚至允许回复“这个任务超出我的能力范围”。第二种是把任务拆得更细,让每个Agent只处理单一目标,减少自由发挥的空间。这两个对策配合使用,跑偏概率会大幅下降。

4.5 常见问题速查表

现象可能原因排查方法解决方案
财务数据前后对不上口径不一致或版本混用检查口径标签和公告编码按口径分组,保留最新版本
分析结论出现明显错误数据污染或模型幻觉用校验函数反推误差触发人工复核,加强事实校验
Agent跑偏主题任务边界定义不清查看执行日志JSON摘要明确行动边界,细化任务拆解
月度API成本过高简单任务也调用大模型按Agent统计Token消耗简单任务切换到小模型
大量请求限流或超时并发设置过大查看日志中的错误码调整并发数,增加间隔时间
研报解析结果混乱PDF原文排版复杂检查PDF转文本质量预处理分栏、表格转换为Markdown

5. 后续还能怎么扩展

这套体系目前已经稳定运行了挺长时间,我还在继续加新东西。一个是把事件驱动机制做进去,比如上市公司发布业绩预告或重大合同公告时,自动触发对应股票的深度分析任务,而不是每天固定时间批量跑。

另一个设想是把分析结论以图结构呈现,公司、行业、上下游、可比公司之间自动建立关系图谱。AI Agent在分析新公司时,可以顺着图谱找到相关的行业背景和同业数据,分析深度会比现在提高一大截。

最后想说的是,AI自主驱动投研这套东西,最大的价值不是帮你“一秒生成研报”,而是把研究员从重复劳动中解放出来,让他们把时间花在真正需要判断力的事情上。从我自己的实践体感来看,如果你也想做类似的系统,一开始的定位一定要准确:让AI做执行层,让人做决策层,这个边界划得越清晰,系统的实用价值就越大。我强烈建议你先拿3到5家公司的公开数据搭一个最小原型,跑通之后再逐步扩容,千万不要一上来就追求大而全。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询