1. 为什么要把 A 股分析拆成多角色协同
很多人第一次用大模型做 A 股分析,都是同一个套路:打开对话框,输入「帮我看看今天大盘怎么样」,然后等一段洋洋洒洒的回复。用几天就会发现三个问题——信息是碎的,今天问的和昨天问的接不上;视角是单一的,一个模型既当宏观分析师又当交易员,结论经常自相矛盾;节奏是被动的,你不问它就不动,盘中该提醒的时候它沉默。
OpenClaw 这类 Agent 框架的价值,恰好在于把「一问一答」升级成「有分工的流水线」。所谓多角色协同,说白了就是给同一个模型戴上不同的帽子:一顶帽子专门盯宏观和政策,一顶帽子专门做个股基本面和技术面,还有一顶帽子只干一件事——挑毛病、算风险。三个角色各写各的分析,最后由调度层汇总成一份可执行的报告。
这套流程适合谁?适合有一定 Python 基础、想把自己的分析框架沉淀成代码的散户和量化爱好者;也适合做投研工具的产品同学,想验证多 Agent 协作在金融场景里到底能跑出什么效果。它不适合指望「输入代码就躺赚」的人,因为最终决策仍然要你自己拍板。
我这次要复现的,是一条完整链路:行情抓取角色负责拉数据,策略生成角色负责出观点,风险复核角色负责唱反调,三者通过统一的模型 API 通道调用大模型,最后跑一次历史回测验证流程是否自洽。关键点在于,所有角色共用一套 Key 和 endpoint,这样切换模型、统计成本、排查报错都只在一个地方改。下面从环境准备开始,一步步把这条链路搭起来。
2. TaoToken 统一 Key 接入的前置准备
多角色协同最怕的就是配置散落各处:抓取角色用一个 Key,策略角色用另一个,风控角色又换一个 base_url,结果某个角色报 401 你都不知道是哪份配置的问题。所以第一步不是写 Agent,而是先把模型接入层统一掉。
我采用的是 TaoToken 的 API 通道,官网入口是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API 基址固定为 https://taotoken.net/api 。它的作用是把不同厂商的模型收敛到一套 OpenAI 兼容协议下,你只需要维护一个 Base URL、一个 Key、一组 Model ID,三个角色都从这里取。
先拿 Key。登录后进入控制台,地址是 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,在 API Keys 页面创建一个新 Key,复制出来存到环境变量里,别硬编码进脚本。创建 Key 的具体页面在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。
拿到 Key 之后,建议先别急着写 Agent,用最朴素的方式验证通道是否通。我习惯先跑一个最小请求,确认 Base URL、Key、Model ID 三件套没问题,再去搭复杂的多角色逻辑。这一步能省掉后面大量「到底是网络问题还是代码问题」的扯皮。
环境变量这样设置,Linux/macOS 用 export,Windows 用 set 或写进 .env:
export TAOTOKEN_API_KEY="sk-你的Key" export TAOTOKEN_BASE_URL="https://taotoken.net/api"模型 ID 建议先选一个通用对话模型做联调,等流程跑通再按角色换更合适的模型。比如宏观角色可以用长上下文模型读政策文件,策略角色用推理强的模型,风控角色用响应快的模型。这些都在同一套 Key 下切换,不用改接入代码。
有一点要提醒:Key 属于敏感信息,别写进 Git 仓库,也别贴到公开的 issue 里。我一般用 .env 加 python-dotenv 加载,.env 写进 .gitignore。多角色系统里每个角色都读同一个环境变量,这样轮换 Key 时只改一处。
3. 可复制的多角色配置与调度脚本
这一节是核心,给出可以直接抄的配置和代码。整体结构分三层:配置层(settings)、角色层(三个 Agent 的 prompt 和工具)、调度层(按时间或事件触发)。
先看配置。我用一个 settings.json 管理模型接入和角色参数,路径放在项目根目录的 config/settings.json:
{ "llm": { "base_url": "https://taotoken.net/api", "api_key_env": "TAOTOKEN_API_KEY", "default_model": "gpt-4o-mini", "timeout": 60 }, "roles": { "macro": { "model": "gpt-4o-mini", "temperature": 0.3, "system_prompt": "你是宏观政策分析师,只负责解读国际形势、国内政策、宏观数据对A股的影响,输出要点不超过5条,每条必须给出逻辑链。" }, "strategy": { "model": "gpt-4o-mini", "temperature": 0.5, "system_prompt": "你是选股与策略专家,基于给定的行情和基本面数据,输出2-3个候选方向,每个方向说明基本面依据、技术面位置、预期差来源。" }, "risk": { "model": "gpt-4o-mini", "temperature": 0.2, "system_prompt": "你是风控官,只做一件事:对策略角色给出的每个方向提出反驳,指出最可能失效的场景,并给出仓位上限建议。" } }, "schedule": { "morning": "09:00", "afternoon": "14:00" } }注意 base_url 写的是 https://taotoken.net/api ,不带任何多余路径。api_key_env 指向环境变量名,而不是 Key 本身,这样配置可以进版本库。
接下来是调度脚本。我用 Python 写一个最小可跑的版本,依赖 openai 和 apscheduler:
import os import json from openai import OpenAI from apscheduler.schedulers.blocking import BlockingScheduler with open("config/settings.json", "r", encoding="utf-8") as f: cfg = json.load(f) client = OpenAI( base_url=cfg["llm"]["base_url"], api_key=os.environ[cfg["llm"]["api_key_env"]], ) def call_role(role_name, user_content): role = cfg["roles"][role_name] resp = client.chat.completions.create( model=role["model"], temperature=role["temperature"], messages=[ {"role": "system", "content": role["system_prompt"]}, {"role": "user", "content": user_content}, ], ) return resp.choices[0].message.content def run_pipeline(market_snapshot): macro = call_role("macro", market_snapshot) strategy = call_role("strategy", f"宏观结论:{macro}\n行情快照:{market_snapshot}") risk = call_role("risk", f"策略方向:{strategy}") report = f"【宏观】\n{macro}\n\n【策略】\n{strategy}\n\n【风控】\n{risk}" with open("reports/latest.md", "w", encoding="utf-8") as f: f.write(report) return report if __name__ == "__main__": scheduler = BlockingScheduler() scheduler.add_job(lambda: run_pipeline("隔夜外盘与今日政策摘要"), "cron", hour=9, minute=0) scheduler.add_job(lambda: run_pipeline("盘中资金流向与技术结构"), "cron", hour=14, minute=0) scheduler.start()这段代码里,三个角色共用同一个 client,只是 system_prompt 和 temperature 不同。行情抓取角色我暂时用字符串代替,实际项目里可以接 Tushare 或行情接口,把返回的 JSON 塞进 market_snapshot。
如果你用的是 Claude Code 或 Cline 这类工具做开发,配置方式略有不同。以 Claude Code 为例,需要在 settings 里指定 Base URL 和 Key,模型 ID 填 TaoToken 支持的名称。三件套缺一不可:Base URL 是 https://taotoken.net/api ,Key 从控制台拿,Model ID 按角色选。Cline 的 MCP 配置同理,把 endpoint 指向 TaoToken,Key 填环境变量引用。
调度层还有一个细节:角色之间要有依赖顺序,宏观先跑,策略基于宏观结论,风控基于策略结论。不要三个角色并行,否则风控拿不到策略输出,反驳就无从谈起。串行虽然慢一点,但逻辑链完整。
4. 验证请求与一次完整回测
配置写完,先别急着上定时任务,手动跑一次验证请求,确认每个角色都能正常返回。最直接的方式是写一个测试脚本,逐个角色调用:
if __name__ == "__main__": snapshot = "隔夜美股下跌1.2%,油价上涨3%,国内发布新能源补贴细则" print(call_role("macro", snapshot))跑通后你会看到宏观角色输出几条带逻辑链的要点。如果这里报错,先看错误类型,下一节专门讲排查。
三个角色都通了之后,做一次完整回测。回测的目的不是验证策略赚钱,而是验证流程自洽:给定一段历史行情,三个角色能否依次产出结论,风控能否真的提出有效反驳。
我选一段有明确政策事件的历史区间,把当时的行情快照和政策文本喂进去,记录三个角色的输出。重点看两件事:策略角色的方向是否有基本面和技术面依据,风控角色的反驳是否具体到场景而不是空话。如果风控只会说「注意风险」,说明 prompt 需要加约束,要求它必须指出失效场景和仓位上限。
回测脚本可以这样组织:
def backtest(events): results = [] for ev in events: macro = call_role("macro", ev["snapshot"]) strategy = call_role("strategy", f"{macro}\n{ev['snapshot']}") risk = call_role("risk", strategy) results.append({"date": ev["date"], "macro": macro, "strategy": strategy, "risk": risk}) return results跑完把结果存成 JSON,人工过一遍。我实测下来,第一版 prompt 里风控角色经常输出「建议谨慎」,加了「必须给出具体失效场景和仓位百分比」之后,输出质量明显提升。这就是多角色协同的调优方式:不是改模型,而是改角色约束。
验证成功的标志是:三个角色的输出能串成一条完整逻辑链,从宏观判断到策略方向再到风险边界,中间没有断裂。到这一步,endpoint 改到 TaoToken 的联调就算完成了。
5. 本篇常见报错排查
多角色系统跑起来,报错基本集中在接入层。下面按真实遇到的错误对照排查。
401 Unauthorized。最常见的原因是 Key 没读到或读错。检查环境变量名是否和 settings.json 里的 api_key_env 一致,检查 Key 有没有多余空格。还有一种情况是 Key 被禁用或额度耗尽,去控制台确认状态。注意 base_url 必须是 https://taotoken.net/api ,多写或少写路径都可能触发 401。
local proxy failed 或连接超时。这类错误通常是本地网络环境或代理配置导致的。检查你的运行环境是否能正常访问外网,如果公司网络有出口限制,换一个网络环境再试。不要在任何脚本里硬编码代理地址,保持接入层干净。
reading choices 相关报错,比如 KeyError: 'choices'。这说明返回体结构和你预期的不一样,通常是请求根本没成功,返回的是错误 JSON。打印完整 response 看内容,多半是模型 ID 写错或参数不合法。确认 Model ID 是 TaoToken 支持的名称,temperature 在合理范围内。
OAuth 或鉴权类报错。如果你用的是 Claude Code 这类工具,报 OAuth 错误通常是工具自身的登录态和 API Key 模式冲突。切到 API Key 模式,把 Base URL、Key、Model ID 三件套填全。CC Switch 切换配置时也要确认这三项都指向 TaoToken,别只改了 Base URL 忘了 Model ID。
角色输出为空或截断。检查 max_tokens 设置,多角色串行时如果上下文太长,后面的角色可能拿不到完整输入。适当精简上游输出,或者给每个角色单独设 token 上限。
排查顺序建议:先单独测接入层(一个最小请求),再测单个角色,最后测整条流水线。这样能快速定位是接入问题还是逻辑问题。
6. 把这条链路用起来
跑通之后,你可以按自己的节奏扩展。比如给行情抓取角色接真实数据源,把 market_snapshot 换成接口返回的 JSON;给策略角色加技术指标计算工具,让它基于真实数据出方向;给风控角色加持仓数据,让它算具体仓位。
模型对话入口在 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,可以先用它手动验证 prompt 效果,再写进配置。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,参数细节都在里面。如果你打算长期跑这套多角色系统,Coding Plan 在 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,适合需要稳定调用和成本可控的场景。
最后说一个我踩过的坑:别一上来就追求五个角色、十个工具。先把三个角色跑顺,确认逻辑链完整,再逐个加。角色越多,调度和排查的复杂度是指数上升的。三个角色能跑出稳定输出,比五个角色互相打架有价值得多。