用LangGraph构建多空AI辩论系统,赋能A股投研分析
今年我干了一件很多人觉得“不务正业”的事给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 StateGraphAgent层多空双方与裁判的角色提示、工具调用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看多/中性/看空、confidence0-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把资料备齐了拍板的人还得是自己。
上一篇/下一篇内容由系统自动关联
返回资讯列表 →