多Agent LLM金融交易框架解析：角色分工、记忆管理与编排实战

简介：TradingAgents是一套基于多Agent大语言模型的金融交易框架，面向量化研究者、算法交易开发者及对大模型智能体应用感兴趣的进阶学习者。该框架模拟真实交易团队中不同角色的分工协作，通过多个LLM Agent共同完成市场分析、观点研判与交易决策，适合用于研究智能体协作机制、回测交易策略及探索LLM在金融场景的工程化落地。压缩包共918个文件，大小约11.45MB，以Python源码（532个py）、编译生成的pyc文件及Markdown文档为主，并包含项目依赖配置、环境脚本、CSV历史行情数据（如600519.SH 2010—2025年数据）等，py文件承载核心逻辑，CSV提供回测样本，目录结构完整，便于本地复现与二次开发。截至目前已有621人学习下载，内含可运行的工程代码与配套说明，能帮助读者快速部署环境、理解多Agent交易决策流程，并基于自带数据开展策略验证。

1. TradingAgents：多Agent LLM金融交易框架到底解决了什么问题

你让一个大模型帮忙看一只股票，给它最近六十天K线和几条新闻，让它给个加仓建议，大多数时候你会得到一段正确的废话：注意风险、分批建仓、不要追高。问题是它不会告诉你，为什么是三成仓而不是五成仓，为什么今天是卖出而不是等待。TradingAgents 这类多Agent LLM金融交易框架想解决的就是这个：把一次交易决策拆成多个角色，让分析师出报告、交易员出信号、风控官出否决意见，每个环节都有可审计的中间产物。它适合两类人，一类在研究交易决策自动化的从业者，另一类是想找一个多Agent编排示例来做工程验证的AI开发者。这篇笔记把框架拆开讲，从角色设计、记忆管理讲到编排循环，最后落在最容易翻车的几个点上。

2. 为什么多Agent比单Agent更适合交易决策：角色拆分的三个理由

2.1 单一LLM的黑匣子问题：交易决策需要“可追问的中间产物”

单Agent直问行情，本质上是一次性完成信息检索、观点形成、风险权衡三个步骤，而这三步各自的权重不可见。模型说“建议观望”，你不知道它是被哪一条新闻影响的，也不知道它有没有认真读过财报里的现金流。交易决策和写文案不一样，错了要能回放：为什么买、为什么没买、为什么止损，每一步都要能拿出来复盘。否则模型连续给出十个正确建议，你也不敢信，因为第十一个错误你会完全找不到原因。

多Agent方案把长链路拆成短链路，每个角色只做一件事，输出一份独立报告。分析师说基本面边际变化是什么，交易员说基于这份报告信号是什么，风控官说这个信号能不能过。每一份输出都可以单独拿去解释、校验、甚至否定。

这个“可追问”的特性，恰恰是金融场景的硬需求。机构里的复盘、留痕、归因离不开中间产物，个人独立研究也需要把“我以为”固定成“它输出过”。

2.2 分析师、交易员、风控官：多Agent交易的典型角色划分与心智模型

这类框架在角色划分上基本有一个共识，就是把一家小公司的前中后台搬进LLM。前台是研究员，负责加工新闻和数据；中台是分析师和交易员，一个负责形成观点，一个负责把观点变成动作；后台是风控官，专职否决。

我一般会先固定四个角色的职责边界，再细化每个角色的输入输出：

每个角色的System Prompt把职责边界写死，同时要求结构化输出。这是多Agent编排示例里最容易抄到的一层，先把角色定义好，再谈协作。

实际复现时角色不用一次拉满。最小可行配置通常是研究员加交易员加风控官三件套，跑通了再拆分析师，再拆执行角色。角色越多，编排成本越高，幻觉来源也越多，后面避坑章节会展开讲。

2.3 框架选型：LangChain、Dify、CrewAI能直接套，但有两个绕不开的取舍

市面上成熟的Agent框架不少，LangChain、Dify、CrewAI都能做多Agent编排，也有现成的工具调用和记忆模块。如果说目标是快速验证“多Agent能不能干这个活”，CrewAI按角色定义团队的方式最接近TradingAgents的原始形态，LangChain更适合你准备深度定制调度逻辑的场景，Dify则适合偏产品化、可视化拖拽的方案。

但直接用通用框架有两个绕不开的取舍。第一个记忆策略是黑盒的，通用框架的记忆窗口按对话轮次管理，而交易场景要求按时间窗口和事件类型管理，该记住三个月的宏观判断，就不能被五分钟前的一条异动消息刷掉。第二个重试语义是按聊天设计的，模型输出格式不对就再问一次，但交易信号重复生成会直接放大模拟成交次数，回测数据立刻失真。

所以落地时宁可先写一层薄薄的编排壳。下面是我常用的一个最小角色注册骨架，不依赖具体框架：

from dataclasses import dataclass, field @dataclass class MarketContext: symbol: str ohlcv: list news: list @dataclass class AgentSpec: role: str # 角色名，如 trader system_prompt: str # 角色系统提示词 input_keys: list[str] # 允许读哪些上下文字段 output_schema: dict # 输出必须满足的 json schema max_retries: int = 2 # 单次输出校验失败的重试上限 class TradingAgent: def __init__(self, spec: AgentSpec, llm): self.spec = spec self.llm = llm def run(self, ctx: MarketContext) -> dict: # 只给角色喂它被允许读的字段，避免上下文污染 filtered = {k: v for k, v in ctx.__dict__.items() if k in self.spec.input_keys} reply = self.llm.chat( system=self.spec.system_prompt, user=json.dumps(filtered, ensure_ascii=False) ) return self._validate(reply) def _validate(self, reply: str) -> dict: # 按 output_schema 校验，不合格就抛给上层编排器重试 return self.spec.output_schema.validate(reply)

这个骨架的关键点是input_keys，它直接限制了每个角色能看到的上下文范围。研究员不能直接读原始行情，交易员不能读未经过滤的新闻列表，这是避免角色越权的最简单手段。output_schema则把每个角色的输出锁成结构化数据，让下一环能稳定消费。max_retries设个2就够，超过这个次数意味着当前数据或提示词有问题，重试再多也是浪费token。

通用框架当然可以用，定位是当传输层，而不是当决策层。你真正要写的核心逻辑是角色边界、消息契约和记忆策略，这三件事放在你自己的代码里，比埋进框架配置里好维护得多。

3. 交易记忆与状态管理：上下文窗口、记忆队列与收益缓存

3.1 行情和新闻混进同一个上下文，token爆炸是必然的

金融场景的输入是典型的时序流：每分钟的tick、日线K线、财报文本、舆情新闻，全量塞进同一个上下文窗口，第一个炸的就是token长度。哪怕窗口够大，模型也会被大量低信息量tick噪声干扰，把短线波动误读成趋势反转。

解决思路是分层压缩，而不是全量记忆。K线这类数值型数据，保留最近N根的OHLCV摘要，更老的部分用统计特征替代，比如区间最高最低、波动率、均线斜率。新闻这类文本数据，先做相关性打分，只保留和当前持仓标的直接相关的高分条目，其余只留标题级摘要。

记忆队列的本质就是一个带淘汰策略的优先级队列，不是简单的列表缓存。写入时带时间戳和源类型，读取时按角色过滤并按优先级排序。淘汰策略通常看两个维度：时间衰减，越早的消息权重越低；事件重要性，风控否决记录比一条普通新闻重要得多。

3.2 用Python实现一个最小记忆队列与收益缓存骨架

我习惯把记忆管理和收益缓存分开写。记忆管上下文，收益管结果，两者混在一起会让回测数据无法追溯。下面是一个可以直接跑的最小骨架：

import time import json from collections import deque class MemoryQueue: def __init__(self, max_len: int = 100, ttl_seconds: float = 86400 * 7): self.max_len = max_len self.ttl = ttl_seconds self.items = deque() def push(self, source: str, role: str, content: dict): # 每条记忆带时间戳，role 用于下游按角色过滤 self.items.append({ "ts": time.time(), "source": source, "role": role, "content": content, }) self._prune() def _prune(self): now = time.time() # 先按时间淘汰过期消息，再按长度淘汰最老的 self.items = deque( [it for it in self.items if now - it["ts"] < self.ttl] ) while len(self.items) > self.max_len: self.items.popleft() def read(self, role: str = None, limit: int = 20) -> list: # 按角色过滤后取最近 limit 条 filtered = [it for it in self.items if role is None or it["role"] == role] return list(filtered)[-limit:] class ReturnCache: def __init__(self): self.trades = [] def record_fill(self, symbol: str, side: str, price: float, qty: float): # 只有当订单真正成交时才写缓存，信号阶段不写 self.trades.append({ "ts": time.time(), "symbol": symbol, "side": side, "price": price, "qty": qty, }) def realized_pnl(self, symbol: str) -> float: # 简单算已实现盈亏，真实场景还要算手续费和滑点 total = 0.0 for t in self.trades: if t["symbol"] != symbol: continue if t["side"] == "buy": total -= t["price"] * t["qty"] else: total += t["price"] * t["qty"] return total

MemoryQueue里max_len设多少合理，取决于上下文预算。我的经验值是：一个角色一轮决策能读的资料不超过2000 token，按每条消息平均100 token算，max_len设20左右就够。设太大模型照样读不完，设太小又会丢掉长周期判断。ttl则对应策略的持仓周期，短线策略设三天，中线策略设两周。ReturnCache的record_fill是关键动作，它必须在确认成交后调用，如果把它挂在信号生成处，回测收益会虚高得离谱，这是后面避坑章要专门讲的。

如果底层用PyTorch基础框架做特征计算，记忆队列里不要存Tensor，只存特征索引或文件路径，下次需要时重新计算或加载。原因是队列对象会跟着进程常驻内存，Tensor一多直接吃掉几GB显存，纯属给自己挖坑。

3.3 记忆写入与读取的时序：谁决定哪条消息留下

很多复现方案在记忆这块翻车的点，不是队列实现，而是写入时机和读取时机搞错。写入时机应该是事件驱动的，不是轮询驱动的。研究员发完摘要写一条，交易员生成信号写一条，风控官否决写一条，每个角色只写自己的产出，不写原始数据。读取时机则是下一轮决策开始时，按当前角色从MemoryQueue里拉取它被允许看到的那部分。

过滤规则我一般分三层：

这三层过滤要在read阶段完成，不要在push阶段做。原因是同一份记忆对分析师和风控官的可用性不同，push阶段一刀切会直接损失信息。把过滤逻辑集中在read方法里，后面想调权重只需要改一个地方。

4. 交易编排不是聊天流：从分析到下单的最小完整循环

4.1 回合状态机：研究→分析→决策→风控的顺序不能乱

多Agent交易的编排和普通聊天最大的区别，是需要确定性的回合顺序。聊天可以随时插话，交易不行，研究结论没出，分析师就没有输入；分析师报告没成，交易员就不该生成信号；信号没过风控，就不能进入成交记录。

我实现这一步用的是最朴素的状态机，不用任何流程引擎，一个字典加一个当前状态指针就够了：

class TradingLoop: def __init__(self): self.transitions = { "RESEARCH": "ANALYST", "ANALYST": "TRADER", "TRADER": "RISK", "RISK": "DONE", } self.current = "RESEARCH" def step(self) -> str: # 返回当前轮到哪个角色执行 role = self.current self.current = self.transitions[role] return role def run(self, agents: dict, ctx: MarketContext): # 按状态机顺序依次执行，任一步失败则整轮终止 while self.current != "DONE": role = self.step() result = agents[role].run(ctx) ctx.record(role, result)

这个循环简单，但有两个必须守住的约定。第一，一轮周期内每个角色只执行一次，禁止交易员在分析报告产出前抢跑。第二，任意一个角色输出校验失败，整轮终止并标记为异常，而不是跳过它继续往下走。跳过的代价是链条里缺了一环，下一轮上下文里全是残缺信息，污染比失败更难处理。

实际扩展时，RISK状态的输出会有三种：通过、拒绝、降仓。通过直接进成交模拟，拒绝整轮结束，降仓则回传TRADER调整仓位参数。这个回传动作相当于状态机里的条件分支，放在上层循环判断就行，不需要额外装一整套编排框架。

4.2 结构化消息契约：为什么必须让LLM输出JSON而不是自然语言

多Agent之间传自然语言，第一轮可能还正常，第二轮就开始失控。常见事故包括仓位写成了“半仓”而不是0.5，方向写成了“看着买”而不是buy，置信度写成了高而不是0.87。自然语言没有类型约束，下游角色解析字段时只能用正则猜，猜错就静默吃一个错误数据。

所以每个角色的输出必须锁成JSON，并配一个JSON Schema做运行时校验。下面是交易员输出信号的标准结构：

{ "signal": { "symbol": "600519", "side": "buy", "position_pct": 0.3, "price_limit": 1500.0, "confidence": 0.72 }, "rationale": "分析师报告确认基本面边际改善，价格站稳20日均线", "risk_flags": ["短期波动率偏高"] }

side字段只允许buy、sell、hold三个值，position_pct必须是一个0到1之间的浮点数，confidence低于0.6时风控官会直接弹回。rationale必须是自然语言，但它是给人类看的，不给下游解析。risk_flags是交易员自己预判的风险点，给风控官做参考。

校验失败时的处理要明确，我的惯例是：第一次失败返回具体错误信息重新生成，第二次失败降级成hold并标记异常，而不是让模型无限重试。原因是同一个输入重复生成两次还不过schema，说明提示词与输出要求不匹配，再试只会烧token。

4.3 重试与降级：LLM返回不了JSON，或者拒绝交易，怎么办

自建多Agent框架时，最容易被低估的是异常处理。LLM不是数据库，它不会稳定返回合法JSON，也不会每次都给出可用决策。复现这类框架时，我建议把重试矩阵提前写在编排壳里，而不是等出了事故再补。

这套重试矩阵就是直接可用的多Agent编排示例。核心原则是：跳过比乱下好，降级比挂起好。整轮卡死是金融系统最不能接受的故障，宁可一个交易动作都不做，也不能让流程停在半路。

超时控制也必须在编排层做。每个角色调用LLM设一个最长等待时间，超时就按表格里的降级动作走。真实交易场景里行情不等人，但模拟环境里更重要的是让每一轮都有确定结果，这样回测才不会因为偶发超时而产生空洞。

5. 多Agent协同的5个翻车现场：死锁、幻觉与上下文污染排查

复现这类框架时，我踩过和见过的高频故障基本集中在五类，每一条都对应一个具体的架构决策错误。下面按现象、原因、解决的顺序写，方便对照排查。

5.1 死锁：分析师等交易员，交易员等分析师

现象：日志显示两个角色都在等待对方输出，循环挂起不动，直到超时中断整轮任务。

原因：角色提示词里写了“等待上一位分析师的报告后再输出”，但编排器的状态机并没有把依赖关系表达清楚。LLM并不知道自己的回合位置，它只是在响应输入的文本，一旦提示词里出现“等”字，它就会真的停在那里。

解决：状态机的step方法必须显式告诉每个角色当前轮次和经过校验的上游输出。提示词里不要出现“等待”，而是直接在上游结论输出后拼接这一条消息。“本轮输入已包含分析师报告，请直接基于报告生成信号”，比写“等待报告”可靠得多。

5.2 幻觉式复盘：LLM把没发生的交易写成盈利

现象：回合结束后，某个角色在报告里写“本策略今日盈利1.2%，表现优于基准”，但日志里根本没有发生过任何成交。

原因：复盘提示词没有绑定真实的成交记录，模型根据行情信息脑补了结果。金融LLM对盈亏数字特别容易产生幻觉，因为它训练数据里到处是这类表述，模型天然倾向于编一个好看的盈利数字。

解决：复盘环节的输出必须严格依赖ReturnCache里的成交记录，提示词明确写“以下数据是系统中的全部成交记录，禁止推断或补充不存在的交易”。如果ReturnCache为空，复盘角色的唯一输出是“当日无成交”，不允许生成任何盈亏数字。

5.3 上下文污染：短线噪声压过长期判断

现象：某条突发波动消息一进来，交易员立刻否定了分析师三天前给出的长期看多判断，频繁转向。

原因：交易员被允许读取了未过滤的原始新闻和tick数据，短线噪声和长线逻辑混在一起，模型天然被最近的消息带节奏。

解决：把context过滤逻辑严格化。交易员的input_keys只包含分析师的长期报告和风控的最近否决记录，不包含原始行情和新闻。研究员负责过滤原始信息，分析师负责提炼长期判断，交易员只消费这两层的产出，这是三层结构存在的意义。

5.4 自说自话：下游Agent无视System Prompt

现象：风控官基于规则拒绝了交易员的高杠杆信号，但交易员在下一轮输出里仍然重复同样的信号，像没看见拒绝记录一样。

原因：对话型LLM在持续多轮交互中，System Prompt的约束力会被新的对话内容稀释。交易员的提示词里写了“必须遵守风控结论”，但当风控结论出现在多轮之前的消息里，模型就“忘了”。

解决：每一轮角色执行前，都要用当前系统提示词加上最近相关的风控结论重新构造Prompt，而不是叠加在一个已经跑了很多轮的长上下文上。同时要求消息契约里带一个risk_status字段，风控状态每次都显式传给交易员。

5.5 收益记录失真：时序写错，回测虚高

现象：回测结果显示年化收益高得离谱，资金曲线比真实策略好一大截，甚至每笔都是最高点卖出最低点买入。

原因：收益缓存写早了，把信号生成阶段当作成交阶段，实际回测撮合逻辑并没有执行。很多复现方案图省事，交易员一输出buy信号就写一笔买入记录，风控后置的降仓逻辑全部落空。

解决：记录成交的唯一入口是ReturnCache.record_fill方法，且它只能在风控通过、模拟撮合并确认成交后调用。信号生成、风控复核、订单撮合三个环节各自留下日志，回测结束后比对三方记录，只要成交数量和撮合日志对不上，这轮回测结果就不作数。

这五个翻车现场有一个共同根源，设计阶段只考虑了角色怎么说话，没考虑角色之间的状态怎么同步。死锁是同步缺失，幻觉是同步缺失，上下文污染也是同步缺失。把记忆队列、成交缓存、轮次状态机做扎实，这五类问题能消掉80%。

6. 验证一套多Agent交易框架：从回测到纸上交易的三个门槛

6.1 事件驱动回放，别一次性灌入K线

回测时最容易犯的错，是把整段历史K线一次性丢给上下文，让模型“看着未来做决策”。正确做法是事件驱动回放，按时间逐根K线推进，每推进一根就调用一轮完整的多Agent循环，模型永远只看得见当期及以前的数据。

我会在回测脚本里放一个时间戳断言：每个角色输出里记录的ts，必须不大于当前回放的K线时间。出现ts超限直接算这轮回测失效。

6.2 用LLM-as-Judge校验每条信号的可解释性

回测收益曲线好看还不够，还要看每条信号能不能自圆其说。我引入第二个LLM当裁判，给定分析师报告和交易员信号，让它判断两者逻辑是否一致。不一致率超过一定比例，说明角色之间实际在打分离策略，决策链条是断的。

这一步不要求裁判LLM懂交易，只要求它能识别逻辑矛盾。它判断的是“依据和结论是否匹配”，而不是预测是否正确，所以噪声很低。

6.3 纸上交易灰度期：固定仓位，只跑一周

收益指标过了阈值，也不要立刻上真金白银，先在模拟盘跑一周。我对灰度期的要求有两个：一是固定仓位不许变，排除人为干预的变量；二是每天手动抽查两个角色的原始输出一次，确认不是只有最终收益好看，而是过程本身正常。

早期有一版框架回测收益翻倍，复盘后才发现收益缓存挂在信号层，所有成交都没过风控。那次之后我养了个习惯：任何一次回测改完代码，先跑一条已知案列检查成交笔数是不是预期值，再跑全量数据。这个习惯帮我拦下了至少三轮因时序写错导致的虚高收益。希望帮到你。

本文还有配套的精品资源，点击获取