1. 项目整体设计一支 AI 投研团队的架构拆分1.1 项目目标与范围界定这个项目的起点很朴素我想验证一下当多个 AI Agent 协作起来能不能真正完成搜集信息 → 产出研报 → 给出交易建议 → 执行模拟下单这一整套投资研究流程。不是那种只生成一段文字报告的 Demo而是要让多个 Agent 各司其职像一家小型投研团队那样运作起来。项目范围划得很清楚只做模拟交易不碰真实资金。所有交易撮合都基于历史行情数据回放或者实时行情的模拟撮合。这一点很重要因为自主交易系统在真实环境下的风险是未知的必须先在一个可控环境里观察它的行为模式。整个系统由四个核心 Agent 组成数据采集 Agent、研报分析 Agent、交易决策 Agent、风控 Agent。另外还有一个调度中枢负责编排它们之间的协作流程。注意市面上很多开源项目把AI 炒股做成了对话机器人——你问它XX 股票怎么样它给你一篇华丽的回答。但这不是我想要的东西。我希望系统能主动工作而不是被动回答问题。所以这里的 Agent 全部是自主运行的而不是聊天式的。1.2 Agent 职责矩阵在设计 Agent 分工时核心原则是职责单一化。每个 Agent 只做一件事并且把这件事做好而不是让一个大模型同时承担所有任务。原因下面会展开讲。Agent核心职责输入输出数据采集 Agent拉取行情、财务、新闻数据并清洗目标标的列表结构化数据集研报分析 Agent基于数据集输出投资研究报告结构化数据集研报 置信度评分交易决策 Agent根据研报和当前持仓生成交易计划研报 持仓快照交易指令集风控 Agent校验交易指令是否合法交易指令 账户信息放行/拦截 理由调度中枢编排 Agent 协作、维护状态全局配置事件序列这个分工映射的是一个真实投研团队的结构数据工程师 → 研究员 → 交易员 → 风控员。把人和人之间数十年的行业分工模式搬到了多 Agent 系统里目的是利用彼此的边界来校准输出质量。1.3 通信与协作机制为什么不用单个大模型硬扛你可能会问直接把所有信息和任务塞给一个大模型不行吗我最初也觉得可以直到跑了几轮实验才发现单个大模型在长链路任务中有三个致命问题第一个问题是上下文污染。当模型需要同时处理新闻、财务指标、行情走势、持仓信息再让它输出交易指令时它会把这些信息搅在一起经常出现因为今天的新闻情绪好所以突破止损位也要持有这种逻辑断裂的结论。第二个问题是目标漂移。让同一个模型既做分析又做决策它的性格会在分析师和交易员之间反复横跳。上一轮它还是个保守的研究员下一轮它就可能变成一个追涨的投机客。缺乏一致性。第三个问题是上下文超限。当你同时喂给它几十只标的的数据时token 很容易超窗。强制截断又会丢失关键信息。所以最终采用了编排者-工作者架构调度中枢负责任务分发和状态管理每个 Agent 通过结构化的 JSON 消息通信各用各的上下文窗口互不干扰。架构上是一条流水线调度中枢 → 数据采集 Agent → 研报分析 Agent → 交易决策 Agent → 风控 Agent → 模拟交易网关每个环节的输出都是下一个环节的输入下一个环节不会看到上一个环节的原始数据源只消费它的产出物。这种设计带来的一个额外好处是可以单独替换任何一个环节而不影响其他环节——比如我把研报 Agent 的模型从 A 换成 B交易决策 Agent 不受任何影响。2. 多 Agent 研报流水线的核心细节2.1 数据预处理层Agent 之前的隐形工程很多人低估数据预处理在这个项目里的分量。研报分析 Agent 的输入质量直接决定输出质量。市面上有大量免费开源的行情和财报接口但原始数据没法直接用因为不同数据源之间时间戳不统一、字段名不一致、复权因子缺失。我在数据采集 Agent 里做的第一件事是字段标准化。不管来自哪个数据源统一映射成以下结构{ symbol: ASSET_A, datetime: 2025-01-15 14:30:00, open: 12.35, high: 12.80, low: 12.20, close: 12.65, volume: 1523400, source: mock_market_data }注意上面是模拟标的不是真实股票代码。跑模拟交易我只用模拟数据源这样才能控制变量、复现问题。第二件事是缺失值处理。比如某只标的有几天停牌成交量是 0如果不处理研报 Agent 可能把这解读成无人交易甚至流动性危机产生误导。我采用的方法是停牌日的 OHLC 都沿用前收盘价成交量填 0并在数据字典里额外加一个trading_status字段标记。第三件事是新闻情绪预判。原始新闻内容太长直接全部塞给研报分析 Agent 成本太高。我在数据采集阶段先用一个小模型做一次粗粒度的情绪打分输出 sentiment_score-1.0 到 1.0和实体标识涉及哪只标的研报分析 Agent 只吃这个打分结果而不是原文。这一层预处理的工作量大约占整个项目的四成。但它带来的收益是巨大的——研报 Agent 的无效输出率从初版的 35% 降到了 10% 以内。2.2 研报 Agent 的 Prompt 设计把量化思维写进指令里研报分析 Agent 是整个系统的大脑它的 Prompt 设计我迭代了七八个版本。这里直接放一版经过验证的设置你是一名资深证券分析师专门负责对给定的标的进行多维度研究分析。 你的任务是 1. 从趋势、估值、情绪、财务四个维度分别进行评判 2. 每个维度必须给出具体依据引用输入数据中的数字 3. 综合得出一个总分0-100和投资评级强烈看好/中性/回避 4. 输出严格遵循给定 JSON Schema 硬性要求 - 禁止编造数据所有结论必须来自输入数据集 - 如果你认为数据不足以支撑结论置信度字段必须降低 - 输出必须是合法 JSON不要包含任何 Markdown 代码块标记关键技巧在于**禁止编造数据和降低置信度这两条约束**。大模型天生有讨好用户的倾向你让它分析它总会给出一个看似完整的结论。但如果数据不够它就会开始编。所以必须给它一条合法的退路——数据不足就降低置信度而不是强行下结论。研报的输出格式定义为{ symbol: ASSET_A, date: 2025-01-15, analysis_dimensions: { trend_score: 72, valuation_score: 58, sentiment_score: 65, financial_score: 80 }, composite_score: 68, rating: 中性偏乐观, confidence: 0.73, reasoning: 趋势指标显示该标的价格站上中期均线财务维度现金流改善明显但估值已接近历史区间上沿且新闻情绪热度偏高需警惕回调。, risk_notes: [估值偏高, 短期情绪过热] }这个 JSON 结构经历了很多次设计调整。最初 rating 用买入/卖出后来发现模拟交易场景中Agent 需要的是程度判断而不是二值判断所以改成了中性偏乐观这类趋势性描述配合 0-100 分的 composite_score。重要经验给 Agent 设计输出格式时一定要给每个字段设计好取值边界。比如评分必须是 0-100 的数字枚举类字段必须限定在给定的值域里。如果没有边界模型会自由发挥下游的交易决策 Agent 拿到各种奇形怪状的输入处理逻辑就要写一堆条件分支。2.3 Agent 间消息传递机制JSON 协议是核心Agent 之间直接用自然语言对话听起来很酷但工程上非常脆弱。我尝试过让研报 Agent 直接输出自然语言描述给交易决策 Agent后果是后者的解析逻辑要处理价格大幅上升涨了不少走势强劲这类模糊表达根本没法稳定解析。最终方案是统一消息协议。调度中枢定义了一套 JSON 消息格式所有 Agent 的输入输出都遵循同一套 schema{ message_id: msg_20250115_001, source_agent: research_agent, target_agent: decision_agent, message_type: research_report, content: { ... }, timestamp: 2025-01-15T09:00:00Z }这套机制的好处有三个第一可追溯。每个消息都有唯一 ID后续如果要回放某一天的系统决策过程只需把当天的消息全部捞出来按时间线排好。第二可校验。在消息传递层做一层 schema 校验不符合格式的消息直接丢弃并触发告警不会把脏数据传给下游。第三可审计。风控 Agent 拦截交易指令后可以往来源回溯看是哪一步分析导致了这条危险指令。2.4 批次调度逻辑一次跑完一个投资组合研报 Agent 不是对着单只标的做分析而是要对整个观察列表做批量分析。调度中枢的批次逻辑是这样的从模拟账户读取当前持仓列表从自选池读取候选标的列表合并去重后形成本次研究批次按批次逐标的调用研报 Agent全部完成后汇总研报集合推送给交易决策 Agent这里有一个性能考虑市面上主流大模型 API 都有并发限制和速率限制。全批次串行跑会很慢所以我加了一个简单的并发控制层——同时最多跑 3 个标的的研报任务每个任务之间间隔 200ms。实测单批次 20 个标的的完整研报生成时间控制在 10 分钟以内算是可以接受的节奏。对于模拟交易来说这个速度够了。3. 自主交易模块与风控系统的实现3.1 交易决策链路从研报到交易指令交易决策 Agent 拿到研报集合后要做的事情是决定买什么、卖什么、用什么价格、用什么仓位。它的输入包括三部分研报集合每个标的的综合评分、评级、置信度当前持仓快照手中的标的、成本价、数量账户约束模拟总资产、现金余额输出的交易指令集长这样{ orders: [ { symbol: ASSET_A, action: buy, order_type: limit, limit_price: 12.50, quantity: 800, reason: 综合评分68分趋势面与财务面共振预期中期向上, confidence: 0.73 } ], portfolio_comment: 维持现有持仓结构增配标A至目标仓位10% }决策逻辑的核心规则并不复杂。我给它定义了三条优先级置信度低于 0.4 的研报不产生交易信号。宁可错过也不要在模型都没把握的时候动手。评分在 60 分以上的标的才允许买入但买入仓位与评分挂钩。60-69 分最多占目标仓位的 5%70-79 分最多 8%80 分以上最多 12%。持仓中评分低于 45 分的标的触发卖出信号不留情面。这里有个容易被忽视的细节agent 的交易理由必须写入订单结构。风控 Agent 拦截时可以根据理由来判断是理性调仓还是模型幻觉导致的随机交易。这个字段在数据追溯时非常关键。3.2 风控规则和参数设计给 Agent 套上缰绳风控 Agent 是整个系统里最重要的环节。没有风控的自主交易就是一辆没有刹车的车跑得快但随时可能翻。我把风控规则分成了四层第一层单笔订单合法性订单数量必须是正数且不超过单票持仓上限买入价格不能超过当前市场参考价的 3% 以上防止恶意高买卖出价格不能低于参考价的 5% 以下防止恐慌砸盘第二层组合层面单标的市值占比不超过总资产的 20%单行业暴露不超过总资产的 40%行业信息在数据采集 Agent 阶段注入总现金比例低于 5% 时强制禁止新的买入订单第三层行为频率同一标的在 24 小时内最多允许 4 次下单全天最多 20 笔交易超过则触发熔断连续 3 次订单被风控拦截后交易决策 Agent 进入冷却状态冷却时间 30 分钟第四层账户保护回撤超过 10% 时全市场停止交易进入保护模式单日亏损超过 5% 时卖出所有亏损标的风控 Agent 本身就是一个大模型驱动 规则引擎混合体。纯规则就能判断的条件比如仓位上限走规则引擎需要用语义理解才能判断的比如这个订单理由是理性调仓还是恐慌交易走大模型判断。两者结合的方式是规则引擎先跑一遍硬性违规直接拦截没被拦截的再送大模型做语义复核。注意风控 Agent 的执行逻辑不依赖于大模型的自觉。规则引擎的代码是传统的 if-else 逻辑不经过模型推理。它相当于是第一道闸门永远在线、永远可控。这是我在项目中坚持的一个底线。3.3 回测与模拟撮合用历史行情验证逻辑模拟交易网关是系统的落点。它负责把交易指令在模拟环境中执行。核心逻辑是一个低延迟撮合器import time from dataclasses import dataclass from typing import List dataclass class SimulatedOrder: symbol: str action: str # buy / sell limit_price: float quantity: int submitted_at: float class SimpleMatcher: 简化版模拟撮合器基于最近成交价判断限价单是否可成交 def __init__(self, tick_data): self.tick_data tick_data # {symbol: [(timestamp, price), ...]} self.current_prices {} self.fills [] def set_market_price(self, symbol: str, timestamp: float) - float: 找到该时间点最近的成交价作为参考价 prices self.tick_data[symbol] latest prices[0] for ts, price in prices: if ts timestamp: break latest (ts, price) return latest[1] def match(self, order: SimulatedOrder, timestamp: float) - bool: ref_price self.set_market_price(order.symbol, timestamp) if order.action buy and order.limit_price ref_price: # 买单价高于或等于市价成交 fill_price min(order.limit_price, ref_price * 1.01) self.fills.append((order, fill_price, timestamp)) return True elif order.action sell and order.limit_price ref_price: # 卖单价低于或等于市价成交 fill_price max(order.limit_price, ref_price * 0.99) self.fills.append((order, fill_price, timestamp)) return True return False这个撮合器做了大幅简化真实交易要考虑滑点、订单簿深度、手续费模型。但模拟交易阶段核心目的是验证 Agent 的决策链路是否合理没必要把微观结构做得过于复杂。不过两个细节我坚持加了手续费模拟券商佣金和滑点模型按 5 个基点模拟大单冲击。如果不加这两样系统会高估策略收益回测结果失真严重。回测框架的用法很直接。把历史行情数据切成一段一段逐日回放让系统在每一天结束时先跑一遍研报流水线再决定第二天的操作循环往复。我用过去两年的日线数据跑了两周回测结果如下模拟数据策略版本累计收益率最大回撤年化波动率交易次数无风控裸跑18.2%-23.5%28.4%847带基础风控9.6%-8.2%19.6%213带完整风控 置信度过滤12.1%-5.4%16.3%186无风控的版本虽然收益数字更好看但回撤高达 23.5%对于模拟交易来说是一个不可接受的风险水平。带上完整风控后回撤大幅压缩到 5.4%交易次数也明显收敛——说明风控系统不只是拦截订单它还在改变决策 Agent 的行为习惯知道乱下单会被拦截之后交易决策 Agent 的输出明显更理性了。需要特别说明以上回测结果是基于模拟数据和简化假设得出的仅用于验证系统逻辑是否合理。任何真实市场中的策略都需要更严谨的回测框架和更长的时间周期本文内容不构成投资建议请勿直接用于实盘交易。4. 实操中的常见问题与排查技巧4.1 Agent 上下文污染研报内容串台第一个让我头疼的问题是上下文污染。第一版系统中所有 Agent 共享同一个上下文池结果多次出现研报 Agent 把标的 A 的财务数据写给标的 B 的情况。追查后发现是调度中枢在拼接上下文时把前一个标的的研报片段留在了缓存里新的请求继承了残留数据。解决方案很彻底每个 Agent 的上下文改成完全独立。调度中枢为每个 Agent 维护自己的上下文窗口每个任务开始前强制清空上一轮遗留内容任务之间的消息通过结构化消息传递而不是靠共享上下文。排查这个问题的技巧是在消息队列里加一个全链路 trace ID。每个任务从采集到研报到交易指令都携带同一个 trace ID。一旦发现输出异常按 trace ID 捞全部消息马上就能定位是哪一代环节的数据串了。4.2 风控误杀如何区分拦截和误杀跑了一阵子之后我发现风控 Agent 过于激进。有几次交易决策 Agent 发出了完全合理的调仓指令结果因为单票仓位超过 20%被拦截——虽然实际情况是标的 A 的市值因为行情上涨自然涨到了 21%并不是新的买入行为导致的超配。这里暴露的问题风控规则是静态的但仓位是动态变化的。价格上涨会导致自然超配此时正确的做法不是拦截而是提醒决策 Agent是否需要再平衡。我给风控 Agent 加了一个状态字段over_limit_reason区分新增交易导致的超限和市场波动导致的被动超限只有前者触发拦截。另一个常见的误杀场景是同一标的一天内多次下单被频率限制拦截。这其实是我故意设置的熔断机制但决策 Agent 第一次被拦截后如果立即重试一模一样的订单会被再次拦截形成无意义循环。解决方案是给拦截响应增加一个cooldown_until字段调度中枢在这个时间窗内直接跳过同类订单而不是反复尝试。4.3 Token 成本控制研报流水线如何跑得更经济多 Agent 系统的 token 消耗速度远超直觉。我第一版每天跑 20 个标的的完整研报流水线每次要调用多次大模型接口日消耗 token 数达到了数十万级别。虽然模拟交易不涉及真实资金但 token 账单是真实的必须优化。三个优化手段效果显著第一缓存重复分析结果。如果某只标的的研报在 4 小时内已经生成过且期间没有新的行情数据或新闻事件到达直接复用旧研报不再重新调用模型。这一步大约节省了 35% 的 token。第二分级抽样分析。不是每天全量分析所有自选标的。优先分析当前持仓 近期触发过交易信号 新加入自选的标的其他标的每三天做一次摘要级分析即可。全量分析改为每周一次。第三用小模型做粗筛。数据采集 Agent 生成的情绪打分和趋势预判用参数量小得多的模型完成。只有当预判结果显示标的有潜在信号时才触发完整研报分析否则研报 Agent 根本不启动。这一步把研报流水线的日均调用量压缩了将近一半。成本控制的核心思想是让贵的大模型做高价值判断让便宜的小模型做低价值筛选。这个分层思路适用于所有多 Agent 项目不只是金融场景。4.4 常见问题速查表症状可能原因解决方案研报内容张冠李戴Agent 上下文残留隔离上下文窗口加入全链路 trace ID指令被拦截但原因不明风控规则缺乏状态字段区分主动超限与被动超限同一订单反复重试决策 Agent 未读取拦截止损期响应增加 cooldown_until调度层跳过Token 消耗过大重复分析 全量分析结果缓存 分级抽样 小模型粗筛模型输出非法 JSONPrompt 约束不足硬性规定纯 JSON 输出增加 schema 校验回测结果过于完美忽略手续费和滑点在撮合器中加入成本模型5. 工具选型与个人实操体会5.1 Agent 框架选型为什么选它当前市面上的 Agent 编排框架不少常见有 LangGraph、AutoGen、CrewAI。我最终选择了自己写调度中枢而不是直接用现成框架原因是我需要精确控制消息流。金融场景对消息的可审计性要求很高每个决定都要能回溯到具体的数据和推理过程。通用框架虽然上手快但它们在底层会自己维护很多隐式上下文出问题时回溯链路很痛苦。我不需要太多复杂交互模式。现成框架比项目里的流水线能承载更丰富的多 Agent 对话、群聊、分层嵌套等模式而我的场景就是一条直线——数据进来指令出去。为此付出的学习成本和框架抽象反而可能是负担。如果你不想自己造轮子可以参考以下对比框架优势劣势适合场景LangGraph状态图定义清晰支持循环和条件分支概念门槛较高需要复杂状态流转的项目AutoGen多智能体对话原生支持调试方便在长链路流水线中消息冗余较多研究原型、对话式多智能体场景CrewAI上手快角色定义直观自定义消息格式时灵活度受限中小型角色扮演式任务自研调度完全可控消息协议自定义需要自己处理并发、重试、状态持久化对链路质量要求高的生产级项目这个项目的规模处于拟合选择范围内我认为这套自研方案最合理。而且调度中枢的代码量其实并不大——核心部分只有几百行 Python。如果你只是做原型验证完全可以用 CrewAI 或者 LangGraph 快速搭起来。5.2 模型选型与链路延迟实测研报分析 Agent 我用的还是云端的大模型因为研报生成需要较强的推理能力。交易决策 Agent 其实也可以用小一号的模型因为它的输入已经是高度结构化的研报 JSON决策逻辑大部分由规则模板完成大模型只负责做最后的情境判断。风控 Agent 的语义复核部分也用同级别的模型。链路延迟是我在整个工程里花了最多功夫优化的维度。实测第一版完整链路采集 → 研报 → 决策 → 风控 → 撮合单标的平均耗时 4.2 分钟瓶颈在研报 Agent 的输出阶段——模型生成一篇多维度研报需要较长推理时间。优化手段是流式输出 并行切片。把研报拆成趋势、估值、情绪、财务四个独立任务并发执行然后再合并。延迟从 4.2 分钟降到 1.8 分钟。代价是每次研报需要四次模型调用、token 消耗略有增加但考虑到模拟交易时效性用 token 换延迟是划算的。5.3 最后分享两个实操心得整个项目做下来最有价值的不是那套能跑通的多 Agent 系统而是过程中的两个教训。第一个教训不要迷信 Agent 的能力边界。大模型在单点任务上表现可能惊艳但一旦进入长链路、多步骤、强约束的执行环境它需要的外部校验和规则约束远超想象。这个项目里真正让系统稳定运行的是那些看起来不太 AI的规则引擎和 schema 校验而不是模型本身有多聪明。风控 Agent 最可靠的防线是那段完全由 if-else 写成的风控规则代码而不是大模型的推理能力。第二个教训自主交易系统先风控后策略。我最早把大部分精力花在研报 Agent 上觉得只要分析足够准确交易自然就能盈利。后来回测才意识到真正吃掉收益的往往不是分析不准而是风控缺位导致的尾部风险。把风控做好了系统即使不是最赚钱的至少是最难亏大钱的。对任何自主交易系统来说安全下线永远比冲锋陷阵重要。模拟交易阶段多踩几个坑总好过实盘之后被市场教育。