今年我干了一件很多人觉得“不务正业”的事:给A股装了个会辩论的AI。不是那种每秒预测涨停的“黑科技”,而是让几个AI角色围着一只股票开“辩论会”,分成多头、空头和中立裁判,最后给出一份带完整逻辑链的投研思路。技术栈也没多玄乎——大模型API、LangGraph、AKShare,加上一大半提示词工程。这个项目的初衷很简单:我自己看研报、刷股评、盯资金流向,越来越觉得“信息太多、观点太少”,所以想用AI把多空双方先吵明白,再由我自己做决策。顺便说一句,这套东西不是自动交易信号源,它只是一个把数据整理成可辩论观点的工具。如果你也对“多智能体协作”“金融大模型应用”感兴趣,这篇文章值得看完。
1. 整体设计与思路拆解:为什么AI要“先吵一架”
1.1 直接问大模型“能不能买”,基本是废话
大家应该都试过把一只股票代码丢给ChatGPT或者国产大模型,问它“未来走势怎么样”。多数时候你得到的不是一段免责声明,就是一本正经的幻觉。原因其实不难理解:大模型本质是“文本续写机器”,它会对你的问题做“最可能的顺延”,而不是做真正的逻辑推理。尤其是涉及股票这种多空矛盾密集的场景,单一模型很难同时兼顾“基本面利好”和“技术面回调风险”。
还有一个更隐蔽的问题——解释性太弱。它说“建议关注”,但背后的理由往往是从语料里统计出来的相关性,不是针对这只股票当下的数据推导。你去追问,它就开始“以上内容仅供参考”。这种体验多了,你就会发现:与其让一个大模型“全知全能”,不如让多个Agent分饰不同角色,把对立观点都摆上桌面。
1.2 辩论框架的本质:红队蓝队对抗加第三方仲裁
我采用的设计思路是“对抗式生成(Adversarial Generation)”。简单说,就是模仿人类社会里最古老的思辨方式——辩论。系统里有一个“多方Agent”(Bull),专门负责找看涨理由;一个“空方Agent”(Bear),专门负责找看跌理由;还有一个“裁判Agent”(Judge),负责综合双方论据、指出漏洞、给出最终倾向。
这三者不是简单的一问一答,而是形成一个公开可追溯的辩论闭环。多方和空方都要从真实数据里取证据,不能凭空捏造;裁判不能骑墙,必须对每个陈述做强度打分。这样做最大的好处是,即使最终结论错了,你也可以回头翻看是哪个环节漏了什么数据,而不是面对一个“黑箱”输出发呆。
1.3 架构选型:从数据到结论的五层结构
整个系统我拆成了五层:
| 层级 | 职责 | 具体组件 |
|---|---|---|
| 数据层 | 采集行情、财务、新闻、资金流 | AKShare、Tushare、新闻API |
| 状态层 | 维护辩论过程中的上下文与消息记录 | LangGraph StateGraph |
| Agent层 | 多空双方与裁判的角色提示、工具调用 | OpenAI兼容SDK + Function Calling |
| 工作流层 | 控制辩论轮次、判定结束条件 | LangGraph edges与循环逻辑 |
| 输出层 | 结构化结论(评级、置信度、风险点) | JSON输出 |
这层架构不是一开始就设计好的,是我踩了不少坑后总结出来的。最开始的版本是“一条龙”式的:先让一个Agent读全部数据,然后生成多空观点。结果发现信息过载严重,一个Agent根本记不住几十个指标。后来才改成现在这种“按需取数、分角色辩论”的模式。
2. 核心技术点:多Agent协作、模型配置与成本控制
2.1 为什么选LangGraph而不是AutoGen或CrewAI
做多Agent编排,市面上已经有不少现成框架。我对比过AutoGen、CrewAI和LangGraph,最后选了LangGraph,核心原因有三个:
- AutoGen偏重自由会话式的多Agent聊天,适合开放式讨论,但辩论赛需要严格的回合顺序和裁决节点,LangGraph的图结构更贴合这种流程。
- CrewAI偏重“分配任务给角色”,但它的角色协作方式更像团队流水线,缺少自然的对抗轮次和条件跳转。
- LangGraph有清晰的State管理,每个Agent的输出会累积到状态的history字段里,后面节点能看到前面所有发言记录,这正好满足交叉质询的需求。
当然,如果你不想引入额外框架,直接用LangChain写顺序调用也行。但一旦你需要在“裁判判定此刻论点已足,提前终止辩论”这类逻辑上花时间,LangGraph的条件边(conditional edge)会省很多事。
2.2 模型选型与API配置
模型方面,我没有一开始就锁死某个厂商。现在大模型的API接口基本都兼容OpenAI格式,所以我在代码里只配置一个“model_name”变量,想换模型只改这一处。我自己主力使用DeepSeek-V3和Qwen-Max,偶尔切GLM-Zero做交叉验证。
这里有必要说下为什么不用本地部署的模型。财经分析需要处理的数据量大,本地小模型在逻辑推演上还是偏弱。用API虽然每轮有一点成本,但换来的是稳定输出和可用性。而且辩论场景天然需要多次调用,API的成本其实可控,单只股票完整跑一轮辩论大约调用8到10次接口,花费不到1元人民币。
需要注意的是,API Key不要硬编码到代码里。我踩过一次坑,把Key写进测试脚本里被同步到Git仓库,虽然项目是私库,但习惯不好。正确做法是放到环境变量或影响目录下的.env文件里,再用python-dotenv加载。
2.3 提示词设计:让角色“有话可说”且“有据可查”
提示词是这个项目最重要的工程。我总结了一套角色提示词的“三段式”结构:
- 设定身份与目标:比如“你是拥有10年经验的A股价值投资者,你负责从基本面角度寻找看涨理由”。
- 限定信息范围:告诉模型“你只能调用查询工具获取实时数据,你的每个论据都必须来自查询结果或历史行情,不能凭空猜测”。
- 约束输出格式:要求模型用Markdown输出“论点 + 对应指标 + 风险提示”,方便裁判理解。
一个典型的Bull Agent提示词片段:
你是一名偏向成长股的价值投资者。 你的目标是从基本面、资金面、消息面三个维度,找出该股票值得看涨的理由。 你必须调用 get_financial_indicators 获取财务指标,调用 get_realtime_quote 获取实时行情。 每提出一个观点,必须引用你获得的数据,不得猜测。 输出格式为: - 观点:一句话结论 - 依据:数据指标名称与数值 - 置信度:高/中/低你就会发现,模型的输出质量立刻改善,至少它不会说“某股票前景广阔”这种正确的废话,而是会告诉你“营收同比增长23.5%,连续三个季度上升”这种可验证的内容。
2.4 成本优化和延迟控制
多Agent辩论最怕的就是调用次数失控。我最初的版本设了5轮质询,单次辩论调用次数飙到30次以上,等到跑完一轮,盘中都快收盘了。后来做了三个优化:
- 固定最大辩论轮次:多空各陈述一轮后,只允许两轮交叉反驳,然后直接进入裁判阶段。
- 条件提前终止:裁判Agent在每轮结束后判断“双方是否还有新证据”,如果没有,就立刻终止并输出结论。
- 缓存当日行情数据:同一只股票当天重复分析时,数据层直接返回缓存结果,避免重复请求数据源。
实际跑下来,单轮辩论时间控制在20秒以内,成本也能打下来。对于个人分析和复盘已经足够。
3. 实操过程与核心环节实现:从零搭建辩论工作流
3.1 项目结构与依赖
我建议用Python 3.11以上版本,目录结构敲成下面这样:
market_debate/ ├── main.py # 入口 ├── agents/ │ ├── bull.py # 多方 │ ├── bear.py # 空方 │ └── judge.py # 裁判 ├── tools/ │ ├── data_fetcher.py # AKShare数据采集 │ └── search_handler.py # 新闻搜索 ├── graph/ │ └── workflow.py # LangGraph工作流 └── config/ └── settings.py # API配置依赖就四个:langgraph,openai,akshare,pandas。其中akshare是开源免费的数据接口库,能拿到A股交易数据、财务数据、龙虎榜、资金流向等基础数据。需要注意,AKShare的接口偶尔会更新失效,最好给取数函数包一层异常捕获。
3.2 核心代码实现:辩论状态与工作流
先定义辩论状态(State)。在LangGraph里,State是各节点之间传递消息的容器。我的状态里至少有四个字段:股票代码、数据集(DataFrame序列化后的字典)、历史消息列表、最终结论。
from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, END class DebateState(TypedDict): stock_code: str company_name: str data: dict # 存储实时行情与财务数据 history: List[dict] # 存储所有Agent发言 conclusion: str # 裁判最终输出然后定义两个关键节点:数据准备节点和辩论节点。数据准备节点先调用AKShare把股票的基本面快照抓下来,放在State的data字段里。辩论节点是多空双方轮流发言,每个节点都从data中检索与自己论点相关的数据,并把发言追加到history。
def prepare_data_node(state: DebateState): code = state["stock_code"] quote = fetch_realtime_quote(code) fin = fetch_financial_indicators(code) news = fetch_recent_news(code) state["data"] = { "quote": quote, "financial": fin, "news": news } return state def bull_node(state: DebateState): history = state["history"] + [{ "role": "bull", "content": bull_agent.invoke(...) }] return {"history": history}工作流构建部分,我用LangGraph将节点串起来:
graph = StateGraph(DebateState) graph.add_node("prepare_data", prepare_data_node) graph.add_node("bull", bull_node) graph.add_node("bear", bear_node) graph.add_node("judge", judge_node) graph.set_entry_point("prepare_data") graph.add_edge("prepare_data", "bull") graph.add_edge("bull", "bear") graph.add_edge("bear", "judge") graph.add_edge("judge", END)如果你想要两轮交锋,可以在bear_node后面加一个条件边,判断是否进入第二轮双方互驳,然后再进入裁判。
3.3 Agent工具函数设计:让AI按需取数
这里有个关键点:不要把所有数据一次性塞给模型。大模型的上下文窗口是有限的,你把AKShare返回的几百条数据全部丢进去,模型反而抓不住重点。正确的做法是给Agent提供“工具函数”,让它自己在需要的时候调用。
我写了一个DataToolkit类,暴露给模型的是三个函数:
get_realtime_quote(stock_code):返回当前价格、涨跌幅、成交量、换手率。get_financial_indicators(stock_code):返回PE、PB、ROE、营收增长率、净利润增长率。get_news_sentiment(stock_code):返回最近一周相关新闻标题和情感倾向。
每个Agent在生成回复前,可以先对工具函数的返回值做分析,再写观点。这样即使我不显式告诉模型某个指标,模型也知道去查。实际测试下来,这种“按需取数”比“数据灌入”的准确率高不少,因为模型不会落入“数据噪声陷阱”。
3.4 裁判Agent的“终审”逻辑
裁判是这个系统里最难设计的角色。如果裁判太宽松,它就只会“综合考虑多方面因素”然后和稀泥;如果裁判太激进,又容易在缺乏证据时做出极端判断。我在裁判的提示词里加了三个强制要求:
- 逐条评价多空双方的论点,指出哪条论据最扎实,哪条是过度延伸。
- 找出双方共同忽略的风险点,比如“政策变化”“大股东减持”等。
- 输出结构化的JSON结果,字段包括
rating(看多/中性/看空)、confidence(0-1)、arguments(双方主要论据)、risk_points(风险点)。
裁判返回示例:
{ "rating": "中性偏多", "confidence": 0.62, "arguments": { "bull": "营收连续三季度超预期,毛利率提升明显", "bear": "当前PE处于近三年高位,短期估值收敛风险大" }, "risk_points": [ "行业政策变动可能影响下游需求", "限售股解禁对资金面的扰动" ] }这个输出格式我最满意,因为它独立于上面的辩论过程,可以单独保存成历史记录,方便日后复盘对照。
4. 常见问题与排查技巧实录
4.1 模型输出太泛泛而谈怎么办
问题表现:生成了“公司基本面良好,未来有望持续增长”这样的废话。原因就是你没有约束它必须引用数据。我处理办法是在提示词里加“证据链检测”:每个观点必须带上具体指标名,如果指标没有出现在工具返回值里,就标记为“未验证”。这一步可以大幅提升输出质量。
注意:就算加了强制引用,模型也可能“二次创造”一个看起来很像指标的数据。所以裁判阶段我还会做一个“数据核对”,拿输出中的数字与原始数据源比对,不一致就打回重写。
4.2 辩论轮次太多导致超时
如果辩论总在“我就说一句……”“我再补充一点……”中陷入循环,靠增加轮次解决不了问题。我在LangGraph的条件边里加了一个判断:如果本轮双方的输出长度小于某个阈值,就视为“没有再提出新证据”,自动进入裁判。另一个办法是设定硬性最大轮次,比如3轮,到点强制截止。
4.3 多空双方“一致看多”或“一致看空”
有时候因为模型训练语料里某只股票的热度太高,多头和空头会被“带节奏”,导致辩论变成二人转。我的对策是在空方提示词里特意强调“你是一次模型辩论中的空方,你的策略是寻找任何可能导致股价下跌的逻辑,包括估值过高、盈利下滑、资金流出等,不能因为市场情绪乐观就放弃立场”。换句话说,空方Agent的角色不是预测,而是“唱反调”。辩论的意义正在于提供反直觉角度,而不是迎合大众情绪。
4.4 API接口不稳定、限流怎么办
尤其是免费额度期,调用频率过高会触发限流。我做了三个层面的容错:第一,使用tenacity库加指数退避重试,重试3次还失败就切换备用模型;第二,在本地缓存当日数据,同一只股票不重复请求;第三,将辩论任务做成异步队列,避免盘中高峰一次性打满配额。
4.5 合规与风险提示
我必须提醒每一位想复刻这个项目的朋友:这套系统只适合做信息整合和研究思路拓展,不能作为实盘交易的自动信号源。项目里我内置了一个“风险哨兵”模块,如果检测到某只股票近期涨幅过大或换手率异常,就会在结论里单独弹出一条“注意追高风险”。这不只是为了合规,更是为了让系统输出的内容更冷静。
5. 应用场景扩展与实际体验
5.1 不只是股票,还可以辩论行业轮动
这套“辩论+裁判”的框架完全可以迁移到其他金融场景。比如把辩题换成“下个季度应该超配白酒还是新能源”,输入数据换成行业指数估值和资金流向,就让多个Agent分别站在不同行业立场上辩论。我实测下来,行业辩论比个股辩论更有价值,因为它更偏向配置逻辑而不是短期预测。
5.2 结合知识库和研报
如果你能拿到合规的研报文本(例如公司公告、业绩说明会纪要),可以把这些文档存入向量数据库,让Agent在辩论时“检索并引用原文”。这种RAG模式能显著提高论据的可信度。我试过在辩方Agent的工具箱里增加一个search_report(company_name, keyword)函数,让它从本地研报库中检索相关段落,再结合实时数据生成论点。效果比只靠API模型要扎实得多。
5.3 我的复盘实践与三条铁律
项目跑了两三个月后,我开始每周做一次“辩论复盘”:把系统输出的历史结论与后续一周的实际走势做对比,统计正确率和偏差原因。目前我对结果的预期已经放得很平——AI能帮我在10分钟之内整理好一份涉及财务、资金、消息面的多空资料,但决策权始终在我自己这里。关于AI工具的使用,我给自己定了三条铁律:
- 不拿大模型结论作为直接交易信号。
- 每一条AI观点必须能追溯到具体数据来源。
- 定期复盘辩论记录与行情走势,持续修正系统和提示词。
最后再告诉你一个小心得:我把辩论结论摘要成“一句话版”,每天盘前推送到自己手机上。真正动手操作时,我还是会自己打开行情软件,看K线、看量能。AI把资料备齐了,拍板的人还得是自己。