用 Python 搭建多智能体投资系统:从研究到交易执行的完整实战
- 2026-03-15 22:56:51
2026年重磅升级已全面落地！欢迎加入专注财经数据与量化投研的【数据科学实战】知识星球！您将获取持续更新的《财经数据宝典》与《量化投研宝典》，双典协同提供系统化指引；星球内含300篇以上独有高质量文章，深度覆盖策略开发、因子分析、风险管理等核心领域，内容基本每日更新；同步推出的「量化因子专题教程」系列（含完整可运行代码与实战案例），系统详解因子构建、回测与优化全流程，并实现日更迭代。我们持续扩充独家内容资源，全方位赋能您的投研效率与专业成长。无论您是量化新手还是资深研究者，这里都是助您少走弯路、事半功倍的理想伙伴，携手共探数据驱动的投资未来！
引言
你是否想过，AI 不仅能聊天，还能像一个投资团队一样协作完成从市场调研到下单交易的全流程？
最近，一篇技术文章详细介绍了如何利用 Google ADK（Agent Development Kit）、A2A（Agent-to-Agent）协议和 MCP（Model Context Protocol）构建一个多智能体投资平台。该系统由 5 个专业化 AI 智能体组成，各司其职、协同工作，覆盖了投资管理的完整生命周期。
本文将为你拆解这个系统的核心设计思想、关键代码实现和架构亮点，帮助你理解多智能体（Multi-Agent）系统的实际落地方式。
一、系统总览：5 个智能体的投资流水线
整个系统模拟了一家投资公司的工作流程，将投资过程拆分为 5 个阶段，每个阶段由一个专门的智能体负责：
序号 智能体名称 职责 输出 1 Research Agent（研究智能体） 收集市场数据、SEC 文件、新闻情绪，形成投资论点 候选股票、投资论点、确信度评分 2 Security Selection Agent（证券筛选智能体） 基于价值、动量、质量因子进行评分筛选 通过/拒绝的股票列表及理由 3 Portfolio Strategy Agent（投资组合策略智能体） 均值-方差优化，确定资金分配权重 目标权重、预期收益、波动率、夏普比率 4 Risk Monitoring Agent（风险监控智能体） 验证 VaR、集中度、合规限制 APPROVED 或 VETOED 决策 5 Trade Execution Agent（交易执行智能体） 将分配方案转化为实际市场订单 订单 ID、成交价、佣金、滑点
为什么要这样拆分？ 每个智能体只关注自己的职责，研究智能体不需要知道怎么计算 VaR，风险监控智能体也不需要解析 SEC 文件。这种窄化的上下文能产出更可靠、更高质量的结果。
二、三大核心概念：Tools、Prompts、Memory
每个智能体由三个要素定义：
1. Tools（工具）：智能体与外部世界交互的接口，例如获取市场数据、执行交易等结构化函数。
2. Prompts（提示词）：系统指令，定义了智能体的角色、可用工具、边界和输出格式。
3. Memory（记忆）：存储智能体的输入/输出。在一次工作流运行中，所有先前智能体的输出会自动对后续智能体可用。
三、MCP 与 A2A：智能体如何通信和使用工具
3.1 MCP（Model Context Protocol，模型上下文协议）
MCP 是一个开放标准，定义了 AI 智能体如何发现、描述和调用外部工具。核心思想是最小权限原则——每个智能体只能访问注册给它的工具集。
例如，Research Agent 的工具列表为：
%%%代码块%%%python
Research Agent 可以调用的工具
tools = [market_data, sec_filings, news_sentiment, yfinance_docs]
注意：它无法调用 execute_trade，因为该工具不在其工具集中
%%%代码块%%%
即使提示词试图触发交易操作，智能体也无法调用 execute_trade 函数，因为它根本不存在于可调用上下文中。
3.2 A2A（Agent-to-Agent，智能体间通信）
本系统通过 Google ADK 的 SequentialAgent 实现智能体间的顺序通信。每个智能体的输出会被追加到会话状态（Session State）中，下一个智能体的上下文窗口自动包含所有先前智能体的输出。
四、关键代码解析
4.1 智能体规格定义
使用 Python dataclass 声明每个智能体的配置：
from dataclasses import dataclass@dataclassclass AgentSpec: """智能体规格定义，包含名称、提示词文件和模型""" name: str # 智能体名称 prompt_file: str # 对应的提示词 YAML 文件 model: str = "openai/gpt-4o-mini" # 默认使用的 LLM 模型# 定义所有智能体的规格AGENT_SPECS = { "research": AgentSpec( name="research_agent", prompt_file="research_agent.yaml", ), "security_selection": AgentSpec( name="security_selection_agent", prompt_file="security_selection_agent.yaml", ), # ... 其余智能体同理}
4.2 智能体工厂函数
工厂函数是创建智能体的唯一入口，职责单一：
from google.adk.agents import LlmAgentfrom google.adk.tools import BaseTooldef create_investment_agent( spec: AgentSpec, tools: list[BaseTool] = None,) -> LlmAgent: """根据规格和工具列表创建一个投资智能体""" # 加载系统提示词 instruction = None if spec.prompt_file: prompts = load_prompt(spec.prompt_file) instruction = prompts.get("system") agent_tools = tools or [] # 创建 Google ADK 的 LlmAgent 实例 agent = LlmAgent( name=spec.name, # 智能体名称 model=spec.model, # 使用的 LLM 模型 instruction=instruction, # 系统提示词 tools=agent_tools, # 可用工具列表 ) return agent
4.3 工具实现示例（交易执行桩）
所有工具都继承 Google ADK 的 BaseTool 类。以下是交易执行工具的示例：
class TradeExecutionStub(BaseTool): """交易执行工具的桩实现，模拟真实的券商 API""" def __init__(self): super().__init__( name="execute_trade", description="通过券商 API 执行买卖订单", ) async def run_async(self, *, args: dict, tool_context) -> dict: """异步执行交易逻辑""" symbol = args.get("symbol", "UNKNOWN") # 股票代码 quantity = args.get("quantity", 0) # 交易数量 # 模拟成交价格 fill_prices = { "NVDA": 875.20, "MSFT": 415.60, "GOOGL": 175.40, "AAPL": 225.80, } filled_price = fill_prices.get(symbol, 150.0) # 返回交易执行结果 return { "order_id": f"ORD-{symbol}-{quantity}", # 订单 ID "status": "filled", # 订单状态 "filled_price": filled_price, # 成交价格 "commission": round(quantity * filled_price * 0.0001, 2), # 佣金 "slippage_bps": 2.1, # 滑点（基点） } @property def input_schema(self) -> dict: """定义工具的输入参数 JSON Schema""" return { "type": "object", "properties": { "symbol": {"type": "string"}, # 股票代码 "quantity": {"type": "integer"}, # 数量 "action": {"type": "string", "enum": ["BUY", "SELL"]}, # 买/卖 "type": {"type": "string", "enum": ["MARKET", "LIMIT", "VWAP", "TWAP"]}, # 订单类型 }, "required": ["symbol", "quantity", "action", "type"], }
4.4 注册中心与工具绑定
注册中心负责将工具分配给对应的智能体，实现最小权限原则：
def register_all(registry: ComponentRegistry) -> None: """创建所有工具实例并绑定到对应的智能体""" # 实例化各个工具 market_data = MarketDataStub() # 市场数据工具 sec_filings = SecFilingsStub() # SEC 文件工具 news_sentiment = NewsSentimentStub() # 新闻情绪工具 # 远程 MCP 工具：通过 SSE 连接 yfinance 文档 yfinance_docs = McpToolset( connection_params=SseConnectionParams( url="https://gitmcp.io/ranaroussi/yfinance" ) ) # 研究智能体：绑定市场数据、SEC 文件、新闻情绪、yfinance 文档 registry.register_agent( "research", create_investment_agent( spec=AGENT_SPECS["research"], tools=[market_data, sec_filings, news_sentiment, yfinance_docs], ), ) # 其余智能体同理，各自绑定所需的工具子集
4.5 编排器：串联整个工作流
编排器使用 SequentialAgent 将 5 个智能体按顺序串联：
from google.adk.agents import SequentialAgentfrom google.adk.runners import InMemoryRunnerfrom google.genai import typesasync def run_workflow(self, mandate: dict) -> dict: """执行完整的投资工作流""" # 构建用户消息（投资指令） user_message = ( f"Investment mandate: {mandate.get('mandate')}\n" f"Portfolio size: ${mandate.get('portfolio_size', 1_000_000):,.0f}\n" f"Risk tolerance: {mandate.get('risk_tolerance', 'moderate')}\n" ) # 构建顺序执行的工作流 workflow = SequentialAgent( name="investment_workflow", description="完整的投资决策工作流", sub_agents=[ registry.get_agent("research"), # 第 1 步：研究 registry.get_agent("security_selection"), # 第 2 步：证券筛选 registry.get_agent("portfolio_strategy"), # 第 3 步：投资组合策略 registry.get_agent("risk_monitoring"), # 第 4 步：风险监控 registry.get_agent("trade_execution"), # 第 5 步：交易执行 ], ) # 创建运行器并执行 runner = InMemoryRunner(agent=workflow) await runner.session_service.create_session( app_name=runner.app_name, user_id="investment_system", session_id="session_1", ) # 收集各智能体的输出 results = [] async for event in runner.run_async( user_id="investment_system", session_id="session_1", new_message=types.Content( role="user", parts=[types.Part(text=user_message)], ), ): if event.content and event.content.parts: text = "".join( part.text for part in event.content.parts if hasattr(part, "text") and part.text ) if text: # 记录每个智能体的名称和输出 results.append({"agent": event.author, "output": text}) return {"results": results, "status": "success", "mandate": mandate}
五、智能体间的信息传递机制
信息在智能体之间的传递通过 Google ADK 的会话状态（Session State）实现。以下是一个完整的传递示例：
第 1 步：Research Agent 输出研究结果，自动追加到会话中。
第 2 步：会话状态变为：
session = [ {"role": "user", "content": "Investment mandate: 寻找 3-5 只美国大盘科技股..."}, {"role": "research_agent", "content": '{"tickers": ["NVDA", "MSFT", "GOOGL"], ...}'},]
第 3 步：Security Selection Agent 启动时，它的上下文窗口包含以上所有内容。LLM 从上下文中提取 tickers 列表，然后调用 security_screener 工具进行评分。
第 4 步：筛选完成后，结果继续追加到会话中，供下一个智能体读取。
重要提示：JSON 输出合约（Output Contract）实际上就是智能体之间的接口定义。如果 Research Agent 遗漏了 tickers 字段，下游的 Security Selection Agent 可能无法正常工作。因此，在生产环境中建议引入 Pydantic 等工具进行输出验证。
六、架构亮点：SOLID 设计原则
该系统严格遵循 SOLID 设计原则：
• 单一职责：每个智能体只负责一个阶段，工厂函数只负责创建智能体，注册中心只负责组件查找。 • 开放封闭：新增工具只需继承 BaseTool，新增智能体只需添加 AgentSpec 和 YAML 文件，无需修改现有代码。 • 里氏替换：桩工具和生产工具可通过 BaseTool 接口互换，切换到真实 API 无需改动智能体代码。 • 接口隔离：AgentSpec 只包含构建智能体所需的字段，不存在臃肿的配置对象。 • 依赖倒置：注册中心和工厂依赖 BaseTool 和 LlmAgent 抽象，不依赖具体的桩类。
七、多智能体编排模式对比
除了当前使用的顺序执行模式，Google ADK 还支持其他编排模式：
模式 特点 适用场景 SequentialAgent（顺序执行） 严格按顺序执行，可预测、易调试 线性投资流水线 ParallelAgent（并行执行） 子智能体并发执行 同时进行基本面、技术面、情绪分析 LlmAgent 监督模式 外层 LLM 动态决定调用哪些子智能体 需要灵活跳过或重试步骤的场景 迭代循环模式 反复调用某个智能体直到输出满足质量阈值 候选集过大需要逐步缩小范围
总结
这篇文章展示了一个用 Python 和 Google ADK 构建的 5 智能体投资系统。核心要点包括：
1. 专业化分工：将投资流程拆分为研究、筛选、策略、风控、执行 5 个阶段，每个智能体聚焦单一职责。 2. 最小权限原则：通过 MCP 协议，每个智能体只能访问被明确授权的工具，从结构上约束能力边界。 3. 会话状态传递：智能体之间通过 Google ADK 的 Session State 隐式传递 JSON 格式的中间结果。 4. SOLID 架构：整个系统高度模块化，新增工具或智能体无需修改现有代码。 5. 审计可追溯：每个智能体的输出独立记录，形成完整的决策审计链。
对于 Python 学习者来说，这个项目是学习异步编程（async/await）、设计模式（工厂模式、注册中心模式）、抽象基类和多智能体系统架构的绝佳实践案例。
免责声明：本文仅供学习参考，不构成任何投资建议。在做出投资决策前，请咨询专业投资顾问。
参考文章
加入专注于财经数据与量化投研的知识星球【数据科学实战】，获取本文完整研究解析、代码实现细节。 财经数据与量化投研知识社区
2026年全面升级已落地！【数据科学实战】知识星球核心权益如下：
1. 双典系统赋能：获赠《财经数据宝典》与《量化投研宝典》完整文档，凝练多年实战经验，构建系统化知识框架； 2. 量化因子日更教程（2026重磅新增）：每日更新「量化因子专题教程」，配套完整可运行代码与实战案例，深度拆解因子构建、回测与优化全流程； 3. 量化文章专题教程库：300+篇星球独有高质量教程式文章，系统覆盖策略开发、因子研究、风险管理等核心领域，内容基本每日更新，并配套精选学习资料与实战参考； 4. PyBroker实战课程：赠送《PyBroker-入门及实战》视频课程，手把手教学，快速掌握量化策略开发技能； 5. 财经数据支持：定期更新国内外财经数据，为策略研发提供精准、可靠的数据基础； 6. 顶尖学者与行业专家分享：年度邀请学术界博士与业界资深专家开展前沿论文精讲与实战案例分享，不少于4场，直击研究前沿与产业实践；专家直连答疑：与核心开发者及领域专家实时互动，高效解决投研实战难题； 7. 专业社群与专属福利：加入高质量交流社群，获取课程折扣及更多独家资源。
星球已沉淀丰富内容生态——涵盖量化文章专题教程库、因子日更系列、高频数据集、PyBroker实战课程、专家深度分享与实时答疑服务。无论您是初探量化的学习者，还是深耕领域的从业者，这里都是助您少走弯路、高效成长的理想平台。诚邀加入，共探数据驱动的投资未来！
2026年重磅升级已全面落地！欢迎加入专注财经数据与量化投研的【数据科学实战】知识星球！您将获取持续更新的《财经数据宝典》与《量化投研宝典》，双典协同提供系统化指引；星球内含300篇以上独有高质量文章，深度覆盖策略开发、因子分析、风险管理等核心领域，内容基本每日更新；同步推出的「量化因子专题教程」系列（含完整可运行代码与实战案例），系统详解因子构建、回测与优化全流程，并实现日更迭代。我们持续扩充独家内容资源，全方位赋能您的投研效率与专业成长。无论您是量化新手还是资深研究者，这里都是助您少走弯路、事半功倍的理想伙伴，携手共探数据驱动的投资未来！
引言
你是否想过，AI 不仅能聊天，还能像一个投资团队一样协作完成从市场调研到下单交易的全流程？
最近，一篇技术文章详细介绍了如何利用 Google ADK（Agent Development Kit）、A2A（Agent-to-Agent）协议和 MCP（Model Context Protocol）构建一个多智能体投资平台。该系统由 5 个专业化 AI 智能体组成，各司其职、协同工作，覆盖了投资管理的完整生命周期。
本文将为你拆解这个系统的核心设计思想、关键代码实现和架构亮点，帮助你理解多智能体（Multi-Agent）系统的实际落地方式。
一、系统总览：5 个智能体的投资流水线
整个系统模拟了一家投资公司的工作流程，将投资过程拆分为 5 个阶段，每个阶段由一个专门的智能体负责：
为什么要这样拆分？ 每个智能体只关注自己的职责，研究智能体不需要知道怎么计算 VaR，风险监控智能体也不需要解析 SEC 文件。这种窄化的上下文能产出更可靠、更高质量的结果。
二、三大核心概念：Tools、Prompts、Memory
每个智能体由三个要素定义：
1. Tools（工具）：智能体与外部世界交互的接口，例如获取市场数据、执行交易等结构化函数。
2. Prompts（提示词）：系统指令，定义了智能体的角色、可用工具、边界和输出格式。
3. Memory（记忆）：存储智能体的输入/输出。在一次工作流运行中，所有先前智能体的输出会自动对后续智能体可用。
三、MCP 与 A2A：智能体如何通信和使用工具
3.1 MCP（Model Context Protocol，模型上下文协议）
MCP 是一个开放标准，定义了 AI 智能体如何发现、描述和调用外部工具。核心思想是最小权限原则——每个智能体只能访问注册给它的工具集。
例如，Research Agent 的工具列表为：
%%%代码块%%%python
Research Agent 可以调用的工具
tools = [market_data, sec_filings, news_sentiment, yfinance_docs]
注意：它无法调用 execute_trade，因为该工具不在其工具集中
%%%代码块%%%
即使提示词试图触发交易操作，智能体也无法调用 execute_trade 函数，因为它根本不存在于可调用上下文中。
3.2 A2A（Agent-to-Agent，智能体间通信）
本系统通过 Google ADK 的 SequentialAgent 实现智能体间的顺序通信。每个智能体的输出会被追加到会话状态（Session State）中，下一个智能体的上下文窗口自动包含所有先前智能体的输出。
四、关键代码解析
4.1 智能体规格定义
使用 Python dataclass 声明每个智能体的配置：
from dataclasses import dataclass@dataclassclass AgentSpec: """智能体规格定义，包含名称、提示词文件和模型""" name: str # 智能体名称 prompt_file: str # 对应的提示词 YAML 文件 model: str = "openai/gpt-4o-mini" # 默认使用的 LLM 模型# 定义所有智能体的规格AGENT_SPECS = { "research": AgentSpec( name="research_agent", prompt_file="research_agent.yaml", ), "security_selection": AgentSpec( name="security_selection_agent", prompt_file="security_selection_agent.yaml", ), # ... 其余智能体同理}4.2 智能体工厂函数
工厂函数是创建智能体的唯一入口，职责单一：
from google.adk.agents import LlmAgentfrom google.adk.tools import BaseTooldef create_investment_agent( spec: AgentSpec, tools: list[BaseTool] = None,) -> LlmAgent: """根据规格和工具列表创建一个投资智能体""" # 加载系统提示词 instruction = None if spec.prompt_file: prompts = load_prompt(spec.prompt_file) instruction = prompts.get("system") agent_tools = tools or [] # 创建 Google ADK 的 LlmAgent 实例 agent = LlmAgent( name=spec.name, # 智能体名称 model=spec.model, # 使用的 LLM 模型 instruction=instruction, # 系统提示词 tools=agent_tools, # 可用工具列表 ) return agent4.3 工具实现示例（交易执行桩）
所有工具都继承 Google ADK 的 BaseTool 类。以下是交易执行工具的示例：
class TradeExecutionStub(BaseTool): """交易执行工具的桩实现，模拟真实的券商 API""" def __init__(self): super().__init__( name="execute_trade", description="通过券商 API 执行买卖订单", ) async def run_async(self, *, args: dict, tool_context) -> dict: """异步执行交易逻辑""" symbol = args.get("symbol", "UNKNOWN") # 股票代码 quantity = args.get("quantity", 0) # 交易数量 # 模拟成交价格 fill_prices = { "NVDA": 875.20, "MSFT": 415.60, "GOOGL": 175.40, "AAPL": 225.80, } filled_price = fill_prices.get(symbol, 150.0) # 返回交易执行结果 return { "order_id": f"ORD-{symbol}-{quantity}", # 订单 ID "status": "filled", # 订单状态 "filled_price": filled_price, # 成交价格 "commission": round(quantity * filled_price * 0.0001, 2), # 佣金 "slippage_bps": 2.1, # 滑点（基点） } @property def input_schema(self) -> dict: """定义工具的输入参数 JSON Schema""" return { "type": "object", "properties": { "symbol": {"type": "string"}, # 股票代码 "quantity": {"type": "integer"}, # 数量 "action": {"type": "string", "enum": ["BUY", "SELL"]}, # 买/卖 "type": {"type": "string", "enum": ["MARKET", "LIMIT", "VWAP", "TWAP"]}, # 订单类型 }, "required": ["symbol", "quantity", "action", "type"], }4.4 注册中心与工具绑定
注册中心负责将工具分配给对应的智能体，实现最小权限原则：
def register_all(registry: ComponentRegistry) -> None: """创建所有工具实例并绑定到对应的智能体""" # 实例化各个工具 market_data = MarketDataStub() # 市场数据工具 sec_filings = SecFilingsStub() # SEC 文件工具 news_sentiment = NewsSentimentStub() # 新闻情绪工具 # 远程 MCP 工具：通过 SSE 连接 yfinance 文档 yfinance_docs = McpToolset( connection_params=SseConnectionParams( url="https://gitmcp.io/ranaroussi/yfinance" ) ) # 研究智能体：绑定市场数据、SEC 文件、新闻情绪、yfinance 文档 registry.register_agent( "research", create_investment_agent( spec=AGENT_SPECS["research"], tools=[market_data, sec_filings, news_sentiment, yfinance_docs], ), ) # 其余智能体同理，各自绑定所需的工具子集4.5 编排器：串联整个工作流
编排器使用 SequentialAgent 将 5 个智能体按顺序串联：
from google.adk.agents import SequentialAgentfrom google.adk.runners import InMemoryRunnerfrom google.genai import typesasync def run_workflow(self, mandate: dict) -> dict: """执行完整的投资工作流""" # 构建用户消息（投资指令） user_message = ( f"Investment mandate: {mandate.get('mandate')}\n" f"Portfolio size: ${mandate.get('portfolio_size', 1_000_000):,.0f}\n" f"Risk tolerance: {mandate.get('risk_tolerance', 'moderate')}\n" ) # 构建顺序执行的工作流 workflow = SequentialAgent( name="investment_workflow", description="完整的投资决策工作流", sub_agents=[ registry.get_agent("research"), # 第 1 步：研究 registry.get_agent("security_selection"), # 第 2 步：证券筛选 registry.get_agent("portfolio_strategy"), # 第 3 步：投资组合策略 registry.get_agent("risk_monitoring"), # 第 4 步：风险监控 registry.get_agent("trade_execution"), # 第 5 步：交易执行 ], ) # 创建运行器并执行 runner = InMemoryRunner(agent=workflow) await runner.session_service.create_session( app_name=runner.app_name, user_id="investment_system", session_id="session_1", ) # 收集各智能体的输出 results = [] async for event in runner.run_async( user_id="investment_system", session_id="session_1", new_message=types.Content( role="user", parts=[types.Part(text=user_message)], ), ): if event.content and event.content.parts: text = "".join( part.text for part in event.content.parts if hasattr(part, "text") and part.text ) if text: # 记录每个智能体的名称和输出 results.append({"agent": event.author, "output": text}) return {"results": results, "status": "success", "mandate": mandate}五、智能体间的信息传递机制
信息在智能体之间的传递通过 Google ADK 的会话状态（Session State）实现。以下是一个完整的传递示例：
第 1 步：Research Agent 输出研究结果，自动追加到会话中。
第 2 步：会话状态变为：
session = [ {"role": "user", "content": "Investment mandate: 寻找 3-5 只美国大盘科技股..."}, {"role": "research_agent", "content": '{"tickers": ["NVDA", "MSFT", "GOOGL"], ...}'},]第 3 步：Security Selection Agent 启动时，它的上下文窗口包含以上所有内容。LLM 从上下文中提取 tickers 列表，然后调用 security_screener 工具进行评分。
第 4 步：筛选完成后，结果继续追加到会话中，供下一个智能体读取。
重要提示：JSON 输出合约（Output Contract）实际上就是智能体之间的接口定义。如果 Research Agent 遗漏了
tickers字段，下游的 Security Selection Agent 可能无法正常工作。因此，在生产环境中建议引入 Pydantic 等工具进行输出验证。
六、架构亮点：SOLID 设计原则
该系统严格遵循 SOLID 设计原则：
• 单一职责：每个智能体只负责一个阶段，工厂函数只负责创建智能体，注册中心只负责组件查找。 • 开放封闭：新增工具只需继承 BaseTool，新增智能体只需添加AgentSpec和 YAML 文件，无需修改现有代码。• 里氏替换：桩工具和生产工具可通过 BaseTool接口互换，切换到真实 API 无需改动智能体代码。• 接口隔离： AgentSpec只包含构建智能体所需的字段，不存在臃肿的配置对象。• 依赖倒置：注册中心和工厂依赖 BaseTool和LlmAgent抽象，不依赖具体的桩类。
七、多智能体编排模式对比
除了当前使用的顺序执行模式，Google ADK 还支持其他编排模式：
总结
这篇文章展示了一个用 Python 和 Google ADK 构建的 5 智能体投资系统。核心要点包括：
1. 专业化分工：将投资流程拆分为研究、筛选、策略、风控、执行 5 个阶段，每个智能体聚焦单一职责。 2. 最小权限原则：通过 MCP 协议，每个智能体只能访问被明确授权的工具，从结构上约束能力边界。 3. 会话状态传递：智能体之间通过 Google ADK 的 Session State 隐式传递 JSON 格式的中间结果。 4. SOLID 架构：整个系统高度模块化，新增工具或智能体无需修改现有代码。 5. 审计可追溯：每个智能体的输出独立记录，形成完整的决策审计链。
对于 Python 学习者来说，这个项目是学习异步编程（async/await）、设计模式（工厂模式、注册中心模式）、抽象基类和多智能体系统架构的绝佳实践案例。
免责声明：本文仅供学习参考，不构成任何投资建议。在做出投资决策前，请咨询专业投资顾问。
参考文章
财经数据与量化投研知识社区
2026年全面升级已落地！【数据科学实战】知识星球核心权益如下：
1. 双典系统赋能：获赠《财经数据宝典》与《量化投研宝典》完整文档，凝练多年实战经验，构建系统化知识框架； 2. 量化因子日更教程（2026重磅新增）：每日更新「量化因子专题教程」，配套完整可运行代码与实战案例，深度拆解因子构建、回测与优化全流程； 3. 量化文章专题教程库：300+篇星球独有高质量教程式文章，系统覆盖策略开发、因子研究、风险管理等核心领域，内容基本每日更新，并配套精选学习资料与实战参考； 4. PyBroker实战课程：赠送《PyBroker-入门及实战》视频课程，手把手教学，快速掌握量化策略开发技能； 5. 财经数据支持：定期更新国内外财经数据，为策略研发提供精准、可靠的数据基础； 6. 顶尖学者与行业专家分享：年度邀请学术界博士与业界资深专家开展前沿论文精讲与实战案例分享，不少于4场，直击研究前沿与产业实践；专家直连答疑：与核心开发者及领域专家实时互动，高效解决投研实战难题； 7. 专业社群与专属福利：加入高质量交流社群，获取课程折扣及更多独家资源。
星球已沉淀丰富内容生态——涵盖量化文章专题教程库、因子日更系列、高频数据集、PyBroker实战课程、专家深度分享与实时答疑服务。无论您是初探量化的学习者，还是深耕领域的从业者，这里都是助您少走弯路、高效成长的理想平台。诚邀加入，共探数据驱动的投资未来！
好文推荐
1. 用 Python 打造股票预测系统：Transformer 模型教程（一）
2. 用 Python 打造股票预测系统：Transformer 模型教程（二）
3. 用 Python 打造股票预测系统：Transformer 模型教程（三）
4. 用 Python 打造股票预测系统：Transformer 模型教程（完结）
6. YOLO 也能预测股市涨跌？计算机视觉在股票市场预测中的应用
9. Python 量化投资利器：Ridge、Lasso 和 Elastic Net 回归详解
好书推荐
