做投资研究这几年我最大的感受就是信息量太大一个人根本看不过来。所以我从去年开始尝试把AI Agent引入投资研究工作流用多智能体分工处理数据采集、新闻舆情、财报分析和风险总结把原来每天要花两三个小时的盯盘和翻资料压缩成十分钟就能拿到一份带完整逻辑链和来源引用的结构化报告。这篇文章就把我的经验整体整理一遍从架构设计、技术选型到关键代码、踩坑记录都会讲到最后还有一套给新手的从0到1练手路径。适合正在做量化投研、准备用Agent替代重复劳动的朋友参考。1. 为什么选择用AI Agent重构投资研究流程1.1 传统投资研究方式的真实痛点先说清楚我要解决的痛点。做股票、期货这类二级市场的投研本质上是在跟信息赛跑而信息是无限膨胀的。以前我的流程大致是早晨起来先看隔夜外盘再刷一遍财经新闻然后翻公告、研报最后自己手工汇总成当天的观察清单。这套流程有两个致命的毛病第一是覆盖不全。一个人精力的上限摆在那里盯了三五个行业其他行业冒出的重要变化根本没精力处理。我试过同时跟踪二十只股票每只股票每天光公告、新闻、大宗交易记录就能产生几十条信息这还没算技术面的价格形态。坚持了两周就发现所谓跟踪基本变成了“看标题式关心”很多关键信息其实都漏过去了。第二是情绪干扰。人天生容易受账户浮盈浮亏影响涨了容易乐观跌了容易悲观。这导致同一个消息在不同情绪状态下你解读出来的“风险等级”完全不一样。更麻烦的是这种偏差你自己很难察觉。后来我意识到如果能把信息收集、初步分析、风险提示这些重复劳动交给程序自己只做决策层的工作就能大幅减少这种情绪污染。1.2 我期待Agent方案解决什么问题开始动手前我给自己列了几个明确的目标信息采集要全分析过程要快结论要能回溯。所谓“回溯”就是说任何一个结论比如“这只股票最近负面舆情增多”你点开之后要能看到它依据了哪几条新闻、哪几份公告而不是黑盒模型告诉你一个结论就没有然后了。这也是我最终放弃“直接调用大模型API问几个问题”这种简单方案的原因。单次问答模式看起来能用但做不到持续跟踪和多人协作更做不到按标准流程批量处理几十只股票。AI Agent恰恰适合这种场景——它可以把“搜集数据—分析基本面—评估新闻情绪—生成报告”拆成多个步骤每一步都有明确输入输出还能动态调整。相当于我请了四个研究员各自守着一块任务最后把分析结果汇总到我这里。1.3 Agent方案和RPA、普通脚本的本质区别有人可能会说这不就是爬虫加脚本吗我之前也做过纯爬虫方案写死了采集频率和规则结果每次业务逻辑一调整就要改一堆代码。Agent方案最大的不同在于它有推理和规划能力。还是用新闻分析举例子。RPA脚本能做的极限是抓取新闻列表按关键词过滤然后输出一个“提到XX公司多少条”的统计。但AI Agent能做的是先识别出新闻里涉及的公司、事件类型、影响方向再根据信源权威性加权最后结合当前股价位置给出“这个消息已经被price in多少”的推测。这些步骤RPA没法用固定规则写死因为每一条新闻的表达方式都不同只有具备语义理解能力的模型才能在一个动态流程里完成。2. 整体架构设计与核心流程2.1 系统整体架构分层整个项目我分成了四层数据接入层、分析Agent层、流程编排层、服务输出层。每一层的职责都很明确没有把逻辑混在一起。层级职责核心组件数据接入层行情、新闻、公告、财务数据的采集与清洗AkShare、Tushare、自建爬虫、httpx异步客户端分析Agent层分工完成基本面、技术面、舆情、风险分析LangChain工具调用、各Agent提示词模板流程编排层管理多Agent执行顺序、条件分支和状态传递LangGraph StateGraph服务输出层对外提供HTTP接口、任务调度与结果缓存FastAPI、Redis、Celery这套分层的出发点是可替换性。比如今天AkShare接口改版我只需要改数据接入层里的一个小模块分析Agent和编排层完全不用动。同样的道理今天想从GPT换到DeepSeek或Qwen只需要在模型工厂里改一个配置项所有Agent的代码框架保持不变。2.2 数据源选型与接入策略数据源是整个项目的地基这里踩的坑最多。我的选型逻辑很简单能用免费稳定接口解决的不用爬虫能一次拿全量数据的不要逐条请求。行情数据我主要用AkShare和Tushare两个库。AkShare胜在接口全、更新快从A股实时行情到期货夜盘数据都有覆盖Tushare胜在数据结构规范、权限体系完整做历史回测时更可靠。我的实测经验是日常实时行情用AkShare就够但如果要拉五年以上的日线数据做回测Tushare的积分接口更稳。新闻类数据是最麻烦的。财经网站有反爬机制而且不同网站的栏目结构经常变。我的做法是自建一个小型采集服务定时抓取几个可信度比较高的财经门户的财经头条和个股新闻板块存到数据库后做去重和清洗。这里特别提醒一句抓取公开网页要注意目标网站的robots协议和访问频率别把别人的站点打挂了。我这边控制单IP请求频率在每秒两次以内同时设置请求失败自动退避。公告数据我用巨潮资讯网的公开接口这个源的好处是权威、格式统一。财务数据则在财报季从AkShare对准报告期批量拉取。宏观数据如果不涉及敏感细分项也可以用公开的统计接口但我不做预测只做参考这点后面会细说。2.3 分析Agent如何分工我设计了五个Agent每只只负责一件事这样提示词可以写得非常聚焦不会出现一个Agent什么都会但什么都做不精的情况。新闻舆情Agent负责抓取目标标的相关新闻做情感极性判断和热度趋势分析。基本面Agent负责读财报指标比如营收增速、净利率、资产负债率判断与上一报告期的变化情况。技术面Agent负责计算均线、MACD、量价配合度生成短期趋势描述。估值Agent负责把当前市盈率、市净率放到历史区间里输出“处于历史什么水位”的判断。综合风控Agent负责汇总前面四个Agent的输出检查是否有矛盾点生成最终的风险提示和总结报告。这种分工还有一个好处就是单Agent的上下文窗口占用非常低不至于把新闻全文一股脑塞给一个模型。金融大模型本来就容易在长上下文里丢失关键信息切短输入以后准确率提升很明显。3. 关键实现与实操记录3.1 环境准备与项目结构整个项目基于FastAPI和LangGraph构建模型层用LangChain的OpenAI兼容接口方便随时切换不同厂商的模型。如果你要复现环境依赖大概是这样pip install fastapi uvicorn langchain langchain-openai langgraph pip install akshare tushare pandas redis asyncpg这些依赖不代表你全部都要用AkShare的数据如果只走同步接口不装asyncpg也没关系。我的原则是先让流程跑通再考虑性能和并发不要在第一步就引入过多依赖。项目目录我按模块组织入口很清晰investment_research/ ├── main.py # FastAPI入口 ├── agents/ # 各Agent的提示词和调用逻辑 │ ├── news_agent.py │ ├── fundamental_agent.py │ ├── technical_agent.py │ └── risk_agent.py ├── datasources/ # 数据接入层 │ ├── market.py │ └── news.py ├── workflows/ # LangGraph编排 │ └── research_graph.py └── services/ # 业务服务层 └── research_service.py别看目录小我后来加了定时任务模块、缓存模块、回调告警模块目录结构基本还能维持住靠的就是一开始把“数据”和“分析逻辑”彻底分离。3.2 数据接入模块实现先拿行情数据举例。AkShare的实时行情接口返回的是全市场快照数据量很大我每次只提取目标代码那一行import akshare as ak import pandas as pd def fetch_realtime_quote(symbol: str) - dict: df ak.stock_zh_a_spot_em() row df[df[代码] symbol] if row.empty: return {error: fsymbol {symbol} not found} row row.iloc[0] return { symbol: symbol, name: row[名称], price: float(row[最新价]), change_pct: float(row[涨跌幅]), volume: float(row[成交量]), amount: float(row[成交额]), }这个接口实测第一次调用会比较慢因为要拉全市场数据所以我加了缓存——同一分钟内请求同一只股票直接返回上一次结果避免反复全表扫描。新闻采集也类似抓下来的标题和正文会存入本地SQLite再做一次简单的去重按股票代码建立索引。3.3 新闻舆情Agent的构建舆情Agent是整个系统里最有用的一个也是提示词写得最细致的。我给它设计了三个输出维度情感得分、影响领域、热度趋势。from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate model ChatOpenAI( modelqwen-plus, temperature0.2, max_tokens512, ) prompt ChatPromptTemplate.from_messages([ (system, 你是专业的金融舆情分析师。请根据下面的新闻标题和摘要分析这条新闻对指定上市公司的影响。 要求 1. 先判断情感倾向输出positive/neutral/negative之一。 2. 给出一个-1到1之间的情感分数正数代表利好。 3. 判断影响领域只能从【业绩】【管理层】【行业政策】【市场情绪】【供应链】中选一个最匹配的。 4. 用不超过50字说明判断理由必须引用新闻原文中的关键句。 ), (human, 公司{company}\n新闻时间{time}\n新闻标题{title}\n新闻摘要{summary}), ]) def analyze_one_news(news_item: dict) - dict: result model.invoke({ company: news_item[company], time: news_item[time], title: news_item[title], summary: news_item[summary], }) # 这里会接一个JSON解析器把结果转成结构化字段 return parse_llm_output(result.content)核心点在于“必须引用新闻原文中的关键句”。加了这个约束之后模型编造理由的情况少了很多就算它判断错了我也能顺着它引用的句子去复核不至于出现完全找不到依据的结论。3.4 用LangGraph编排多Agent流程LangGraph对我来说最大的价值是状态传递和条件分支。它不像LangChain原来的Chain那样固定顺序而是能根据前的节点输出决定下一步走向这个特性在投研场景里太实用了。举个实际例子基本面Agent如果发现最新财报季数据还没披露就跳到“等待季报模板”分支不再去做无意义的盈利预测如果新闻舆情Agent发现负面新闻数量超过阈值风控Agent就会被提前触发在最终报告里加大风险警告权重。from langgraph.graph import StateGraph, END from typing import TypedDict, List class ResearchState(TypedDict): symbol: str news: List[dict] fundamental: dict technical: dict risk_score: float report: str def collect_data(state: ResearchState) - ResearchState: # 拉取行情和新闻这里做缓存处理 state[news] news_service.get_related(state[symbol]) return state def analyze_fundamental(state: ResearchState) - ResearchState: state[fundamental] fundamental_agent.run(state[symbol]) return state def analyze_technical(state: ResearchState) - ResearchState: state[technical] technical_agent.run(state[symbol]) return state def analyze_news_sentiment(state: ResearchState) - ResearchState: neg_count sum(1 for n in state[news] if n[sentiment] negative) state[risk_score] state.get(risk_score, 0) neg_count * 0.1 return state def generate_report(state: ResearchState) - ResearchState: state[report] risk_agent.summarize(state) return state graph StateGraph(ResearchState) graph.add_node(collect, collect_data) graph.add_node(fundamental, analyze_fundamental) graph.add_node(technical, analyze_technical) graph.add_node(news, analyze_news_sentiment) graph.add_node(report, generate_report) graph.set_entry_point(collect) graph.add_edge(collect, fundamental) graph.add_edge(collect, technical) graph.add_edge(collect, news) graph.add_edge(fundamental, report) graph.add_edge(technical, report) graph.add_edge(news, report) graph.add_edge(report, END) app graph.compile()compile之后就可以直接传初始状态运行了。LangGraph还会自动把每一步的状态变化记录下来这对事后复盘特别重要——我可以清楚看到一条结论是在哪一步、基于什么数据产生的。3.5 FastAPI接口与并发扛量实践热搜里有个词很准“AI Agent怎么扛并发”。我实际开发中确实被并发问题教育过。一开始我只是简单把流程串起来单线程跑然后发现两个问题一是同步请求大模型太慢一个标的完整流程要一二十秒二是多个请求同时进来程序直接卡死。我的解决方案分几层。首先是FastAPI接口把同步函数改成异步并且用后台任务模式处理“接收请求—立即返回任务ID—后台跑研究流程—完成后主动推送给用户”。这样用户体验好很多不用干等。import asyncio from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel app FastAPI() sem asyncio.Semaphore(10) # 控制同时执行的Agent任务数 class ResearchRequest(BaseModel): symbol: str app.post(/research) async def create_research(req: ResearchRequest, background_tasks: BackgroundTasks): task_id generate_task_id(req.symbol) background_tasks.add_task(run_research_with_semaphore, task_id, req.symbol) return {task_id: task_id, status: queued} async def run_research_with_semaphore(task_id: str, symbol: str): async with sem: await run_research_pipeline(task_id, symbol)这里最难控制的是大模型的QPS配额。不同厂商的模型都有速率限制一旦超限就会被限流甚至封禁。我的做法是维护一个信号量限制同时进行的大模型调用数另外加了一层带指数退避的重试机制。如果有连续失败就把任务丢进延迟队列避免雪崩。除了接口层Redis缓存也很关键。我做了两级缓存第一级是对数据源的缓存AkShare全市场行情两分钟内只拉一次第二级是对分析结果的缓存同一只股票在股票池里短时间重复请求直接用上次生成的报告不再重复调用大模型。实测下来并发从5路撑到30路没有太大问题成本也省了一半多。4. 踩坑与排查实录4.1 大模型幻觉金融场景尤其严重不管用什么模型投资研究里最大的风险就是幻觉。金融数据讲究精确到小数点模型如果编一个假的毛利率出来那整个分析就废了。我踩过最离谱的一次是让基本面Agent总结某公司的盈利情况模型信誓旦旦地写“净利润同比大幅增长35%”我拿原始财报一核对实际数据是下降12%。后来我加了两个防线一是所有数字必须带出处模型必须给出它引用的公告编号或字段名称二是增加一个独立的校验Agent专门去对比模型输出和结构化财务数据不一致就标记为“数据冲突”并重新生成。加了这两道防线之后基本再没出现过明显的数字幻觉。4.2 上下文塞爆与Token超限新闻数量一多把所有正文塞进Prompt肯定不行。有一次我跟踪一只热门股票当天相关新闻六十多条全塞进去直接超过了模型的上下文窗口跑出一个报错。这个问题我用MapReduce的方式解决。先让模型对每条新闻只抽“情感标签影响领域关键句”不写长分析然后汇总所有标签再做整体情绪判断。这样每条新闻最多消耗两三百个Token六十条也才一万多Token完全在可控范围。再往后新闻量更大时我用向量数据库做过一版先把新闻嵌入存储分析时只取跟当前标的相关度最高的二十条效果更稳。4.3 Agent跑飞、死循环和超时多Agent编排最担心的就是流程跑飞。LangGraph虽然提供了状态机能力但节点之间如果某个工具返回了意外格式节点函数没有做好容错整个流程就会卡在中间节点不回传。我遇到过一个典型问题新闻采集服务被目标网站反爬拦截返回了空列表。新闻Agent拿到空列表后继续往下一个节点传风控Agent发现“没有新闻”竟然给出“舆情平稳”的结论——这等于把数据缺失误判成利好很危险。修复方案是每个Agent节点加校验输入数据不满足最小数量时必须在结论里标注“数据缺失可信度降级”并且不允许直接把缺失当正常。另外一定要设置LangGraph的recursion_limit和每个节点的执行超时。我最初没设超时有一次模型服务超限导致节点重试了二十多次白白浪费了很长时间。后来统一设定为单节点最多重试3次、单次等待不超过30秒跑飞的情况基本绝迹。4.4 成本控制单次分析烧掉的Token比想象中多投资研究的Agent流程会反复调用大模型如果设计不当成本会高得吓人。我第一版跑完整流程时单只股票的分析居然要消耗五万Token一个月跟踪二十只股票费用直接超预算。优化思路主要有三条。第一条能用小模型解决的绝不用大模型新闻情感分类这类简单任务用7B量级的模型就很好综合报告这种需要深度推理的才上最强的模型。第二条多Agent之间传递中间结果时只传结构化摘要不传原始文本减少Token量。第三条相似的查询请求走缓存同一个标的在短时间内不做重复分析。这套组合下来单标的成本从五万Token降到了一万以内而输出质量几乎没有变化。5. 从0到1的实践路径与个人建议5.1 新手保持节奏的几个练手项目如果你刚接触AI Agent我建议不要一上来就照着淘宝数据搭建多Agent分析系统先做三个小项目练手。第一个单Agent新闻问答。只做一个新闻采集模块加一个Prompt让它回答“最近三天有哪些关于某公司的重要新闻分别是什么方向”。这个项目能帮助你熟悉大模型调用、JSON解析和基本的缓存逻辑。第二个RAG检索问答。把公司公告文本切片存入向量库让Agent根据用户问题检索相关段落再回答。这个项目能帮你建立起对上下文窗口和检索质量的直觉。第三个双Agent协作。一个Agent负责找数据一个Agent负责审核数据再加一个简单的LangGraph流程把它们连起来。到了这一步你基本就掌握了多Agent协作的核心。做完这三个项目再动工做完整的投研系统你的难度感受会完全不一样。5.2 低代码平台与自建方案怎么选现在市面上也有不少Agent搭建平台比如扣子、Dify这类可视化编排工具适合做原型验证和中小规模场景。我用过一段时间最大的感受是上手快、内置了常用的模型网关和工具节点非常适合验证你的提示词和业务流程设计是否合理。但如果你的需求像投资研究这样涉及大量自定义数据源、自建爬虫、高频定时任务和复杂的并发控制我还是推荐自建。平台通常会限制你的数据接入方式和执行环境而投资研究恰恰在数据源和合规控制上要求很高。我的建议是先用平台把流程跑通确认业务逻辑后再切到自建代码方案。5.3 关于期货交易方向的一些提醒有几个朋友问过我既然能做股票研究是不是也能做个AI Agent直接做期货交易技术上确实有空间期货的行情数据结构更规范很多接口都是现成的。但我个人的真实建议是交易执行和研究辅助完全是两码事不要混在一起。研究辅助的目标是输出“可能性”和“风险点”错了顶多浪费一点精力但自动交易执行出错了是真金白银的亏损。期货还有保证金、强平、滑点、手续费这些问题任何一个没处理好趋势判断再准也可能亏在交易环节。我自己目前把AI Agent定位在“研究副驾”而不是“自动司机”——所有交易决策仍然需要人确认。另外个人搭建全自动程序化交易系统还涉及合规报备和账户门槛的问题各个平台的要求不一样。建议想往这个方向走的朋友先把行情接入、信号生成、研究报告输出这套研究链路做扎实再考虑要不要碰执行端。我个人做这套系统的最大收获并不是省了多少小时盯盘而是把过去依赖经验和情绪的判断变成了一套可以审计、可以回放、可以不断改进的流程。每一个结论都有出处每一次分析都留痕这种“可回溯性”本身就非常有价值。后面我还在计划把历史报告归档、把分析结果跟实际行情走势做批量对照慢慢建立起自己的策略复盘数据库。如果你也在做类似的项目很欢迎一起交流。
