基于CrewAI框架构建多智能体量化投资分析系统实战指南

1. 项目概述：当AI智能体遇上量化投资

最近在GitHub上看到一个挺有意思的项目，叫liangdabiao/easy_investment_Agent_crewai。光看名字，就能嗅到一股“让投资变简单”的味道，而且把“智能体”（Agent）和“CrewAI”这两个当下AI应用开发领域的热门词结合在了一起。作为一个在金融科技和自动化工具开发领域摸爬滚打多年的从业者，我立刻就被吸引了。这本质上是一个尝试用多智能体协作框架（CrewAI）来构建一个简易量化投资分析系统的开源项目。

简单来说，它想解决一个很实际的问题：对于大多数个人投资者，甚至是一些小型团队，搭建一套专业的量化分析流程门槛太高了。你需要懂数据获取、懂因子计算、懂策略回测、懂风险控制，还得会写代码把它们串起来。而这个项目的思路，是把这些复杂的任务“分包”给不同的AI智能体去完成。比如，让一个智能体专门负责从网络或数据库里抓取最新的市场数据，另一个智能体负责根据预设的规则计算技术指标，再有一个智能体扮演“分析师”的角色，综合所有信息给出买卖建议。最后，可能还有一个“经理”智能体来协调整个流程，并做出最终决策。CrewAI框架正好为这种分工协作提供了天然的舞台。

这个想法非常契合当前AI Agent的发展趋势——从单点工具走向协同工作流。对于想入门量化、又不想从头造轮子的朋友，或者对于想探索多智能体系统在金融领域具体落地的开发者来说，这个项目提供了一个绝佳的、可实操的起点。它剥离了复杂系统的外壳，直指核心：如何定义角色、如何设计任务、如何让智能体们有效沟通并产出有价值的结论。接下来，我就结合自己搭建类似系统的经验，对这个项目进行深度拆解，并补充大量实战中才会遇到的细节和思考。

2. 核心架构与智能体角色设计解析

2.1 为何选择CrewAI框架？

在深入代码之前，我们先聊聊选型。为什么是CrewAI，而不是LangChain或其他Agent框架？这背后有几个关键考量。

首先，“Crew”（团队）的概念是核心。LangChain更侧重于构建单个、功能强大的链（Chain）或智能体，虽然也能组装成工作流，但“团队协作”并非其第一设计理念。而CrewAI从诞生之初就将多智能体协作作为原子特性。它内置了Agent、Task、Crew、Process这几个核心抽象，几乎是为我们设想中的“投资分析团队”量身定做。你可以非常直观地定义：谁是研究员（Agent），他的目标是什么（Task），他需要向谁汇报工作（通过Task的output指定为下一个Task的input），以及整个团队按照什么流程运转（Process，如顺序执行或分层协同）。

其次，上下文管理与角色隔离。在投资分析中，不同角色的知识背景和任务焦点截然不同。数据收集者不需要理解复杂的夏普比率计算公式，策略分析师也不必关心API调用的重试机制。CrewAI允许为每个Agent独立配置llm（大语言模型）、role（角色）、goal（目标）和backstory（背景故事）。这个backstory特别有用，它能有效地将角色专业知识“注入”到智能体的提示词（Prompt）中，让它在执行任务时更像一个真正的领域专家，而不是一个通用的聊天机器人。这大大降低了提示工程（Prompt Engineering）的复杂度。

再者，任务驱动的明确性。每个Task都有清晰的description（描述）、expected_output（期望输出）和可选的agent（执行者）。这种结构强迫开发者必须想清楚：这个任务到底要产出什么？是一段格式化文本、一个JSON对象，还是一个决策代码？这种明确性对于构建稳定、可预期的自动化系统至关重要。在量化投资中，模糊的指令会导致灾难性的结果。

最后，生态与易用性的平衡。CrewAI建立在LangChain等成熟库之上，可以方便地集成各种工具（Tools），比如搜索引擎、数据库连接器、代码执行环境等。同时，它又通过更高层次的抽象，隐藏了许多底层复杂性，让开发者能快速搭建出原型。对于easy_investment_Agent_crewai这样的项目目标——“简易投资”，这个选择非常明智。

2.2 投资分析团队的角色蓝图

参考项目思路并结合实战，一个最小可行（MVP）的量化投资智能体团队通常需要以下四个核心角色：

1. 数据工程师智能体

• 角色（Role）：Market Data Fetcher & Cleaner
• 目标（Goal）：准确、及时地获取指定标的（如股票代码、指数代码）的原始市场数据（行情、基本面、舆情等），并进行初步清洗和格式化。
• 背景故事（Backstory）：你是一个严谨的金融数据工程师，对数据源的稳定性、数据的准确性和时效性有极致追求。你熟知各类金融API（如雅虎财经、聚宽、AKShare等）的调用方式和限制，擅长处理缺失值、异常值和数据格式转换。
• 关键工具：网络请求工具（如requests库封装）、数据解析工具、本地/远程数据库连接工具。
• 输出：一个结构化的数据集（如Pandas DataFrame的JSON表示或保存为CSV文件路径），包含时间、开盘价、最高价、最低价、收盘价、成交量等字段。

2. 量化分析师智能体

• 角色（Role）：Quantitative Indicator Analyst
• 目标（Goal）：基于数据工程师提供的数据，计算一系列预先定义的技术指标（如移动平均线MA、相对强弱指数RSI、布林带Bollinger Bands）和基本面指标（如市盈率PE、市净率PB），并生成初步的信号。
• 背景故事（Backstory）：你是一名经验丰富的量化分析师，精通技术分析和多因子模型。你能快速实现各种指标公式，并理解每个指标在市场不同阶段的应用场景和局限性。你对数据噪声敏感，会主动进行平滑处理。
• 关键工具：数值计算库（如pandas,numpy）、技术指标计算库（如TA-Lib的封装）。
• 输出：一个增强的数据集，在原始数据基础上新增了指标列，以及一个基于简单规则（如“金叉”、“死叉”、“超买”、“超卖”）的初级信号列表。

3. 策略研究员智能体

• 角色（Role）：Investment Strategy Researcher
• 目标（Goal）：综合量化分析师的指标信号，结合当前的市场环境描述（可由另一个智能体或固定输入提供），进行更深层次的逻辑推理，生成具体的投资建议报告。
• 背景故事（Backstory）：你是一名冷静、全面的策略研究员。你不盲目相信单一指标，而是善于综合多种信息，权衡风险与收益。你的报告逻辑清晰，包含看多/看空/中性的观点、核心依据、关键风险点以及建议的仓位管理思路。
• 关键工具：主要依赖LLM强大的分析和文本生成能力。可以集成新闻摘要工具作为市场环境输入的来源。
• 输出：一份结构化的投资建议报告，通常包括：标的名称、分析时间、综合观点、主要论据（数据支撑）、风险提示、操作建议（如“买入”、“持有”、“卖出”或更细化的仓位百分比）。

4. 风控与执行经理智能体

• 角色（Role）：Risk Manager & Executive
• 目标（Goal）：审核策略研究员的报告，根据预设的风控规则（如最大回撤限制、单标的仓位上限、黑名单过滤）进行最终决策，并模拟或执行交易指令。
• 背景故事（Backstory）：你是这个投资团队的最终负责人，性格保守，纪律严明。你的首要任务是保护资本安全，其次才是追求收益。你有一本明确的风控手册，任何违反手册的建议都会被否决或修改。
• 关键工具：风控规则引擎（可以是一组if-else逻辑或一个配置文件），模拟交易账簿。
• 输出：最终决策指令（“执行交易”或“放弃”），以及详细的决策日志。在模拟环境中，会更新模拟持仓和资金曲线。

注意：角色设计并非一成不变。在项目初期，为了简化，可以将“量化分析师”和“策略研究员”合并。但随着系统复杂度的提升，职责分离是必然趋势，这符合单一职责原则，也让每个智能体的提示词设计更专注、效果更好。

3. 从零搭建：环境准备与核心代码实现

3.1 基础环境搭建与依赖安装

假设我们使用Python作为开发语言。首先需要一个干净的虚拟环境。我强烈推荐使用conda或venv进行隔离。

# 创建并激活虚拟环境 (以conda为例) conda create -n investment_agent python=3.10 conda activate investment_agent # 安装核心框架 pip install crewai pip install 'crewai[tools]' # 安装一些可选工具依赖 # 安装数据分析必备库 pip install pandas numpy yfinance akshare # yfinance和akshare是获取金融数据的常用库 pip install ta # 一个纯Python的技术分析库，比TA-Lib更容易安装 # 安装LLM相关，这里以使用OpenAI API为例 pip install openai # 或者使用开源模型，例如通过Ollama # pip install ollama # 可选：用于美化输出和日志 pip install rich

接下来是项目结构。一个清晰的结构能让后续开发和维护事半功倍。

easy_investment_crewai/ ├── config/ │ ├── __init__.py │ ├── agents_config.yaml # 智能体角色、目标、背景故事配置 │ └── tasks_config.yaml # 任务描述、期望输出配置 ├── tools/ │ ├── __init__.py │ ├── data_fetcher.py # 数据获取工具 │ ├── indicator_calculator.py # 指标计算工具 │ └── risk_engine.py # 风控引擎工具 ├── crew/ │ ├── __init__.py │ └── investment_crew.py # 组装智能体、任务和流程的核心文件 ├── utils/ │ ├── __init__.py │ └── helpers.py # 通用辅助函数 ├── outputs/ # 存放运行结果、报告 ├── .env # 存放API密钥等敏感信息 ├── main.py # 主程序入口 └── requirements.txt

3.2 核心工具类实现详解

工具（Tools）是智能体的“双手”，它们封装了具体的能力。我们来实现最关键的几个。

tools/data_fetcher.py： 智能的数据抓取者这个工具要健壮，能处理网络异常、数据源变更等问题。

import yfinance as yf import akshare as ak import pandas as pd from datetime import datetime, timedelta import logging from typing import Optional, Union logger = logging.getLogger(__name__) class DataFetcherTool: """金融数据获取工具，支持多种数据源和自动降级策略""" def __init__(self, primary_source='yfinance', fallback_source='akshare'): self.primary_source = primary_source self.fallback_source = fallback_source def fetch_stock_data(self, symbol: str, period: str = "1mo", interval: str = "1d", start_date: Optional[str] = None, end_date: Optional[str] = None) -> Union[pd.DataFrame, str]: """ 获取股票历史数据。 参数: symbol: 股票代码，如 '000001.SZ' 或 'AAPL' period: 数据周期，yfinance格式，如 '1d', '5d', '1mo', '1y' interval: 数据间隔，如 '1d', '1h', '1m' start_date: 开始日期，'YYYY-MM-DD'格式 end_date: 结束日期，'YYYY-MM-DD'格式 返回: 成功: 包含OHLCV数据的Pandas DataFrame 失败: 错误信息字符串 """ try: if self.primary_source == 'yfinance': # 处理A股代码，yfinance需要后缀，如 '000001.SZ' -> '000001.SZ' yf_symbol = symbol if '.' not in symbol and symbol.startswith(('0', '3', '6')): # 简单假设：0/3开头是SZ，6开头是SH exchange = 'SZ' if symbol.startswith(('0', '3')) else 'SH' yf_symbol = f"{symbol}.{exchange}" ticker = yf.Ticker(yf_symbol) if start_date and end_date: df = ticker.history(start=start_date, end=end_date, interval=interval) else: df = ticker.history(period=period, interval=interval) if df.empty: raise ValueError(f"yfinance returned empty data for {symbol}") df.reset_index(inplace=True) # 确保日期列是字符串格式，便于后续JSON序列化 if 'Date' in df.columns: df['Date'] = df['Date'].dt.strftime('%Y-%m-%d') logger.info(f"Successfully fetched data for {symbol} from yfinance, shape: {df.shape}") return df elif self.primary_source == 'akshare': # AKShare针对A股数据更友好 # 这里需要根据akshare的实际接口调整，例如使用 stock_zh_a_hist # 此处仅为示例，实际调用需查阅akshare文档 pass except Exception as e: logger.warning(f"Primary source {self.primary_source} failed for {symbol}: {e}. Trying fallback...") # 降级到备用数据源 return self._fetch_with_fallback(symbol, period, interval, start_date, end_date) def _fetch_with_fallback(self, symbol, period, interval, start_date, end_date): """降级策略实现""" try: if self.fallback_source == 'akshare': # 实现akshare的获取逻辑 # df = ak.stock_zh_a_hist(symbol=symbol, period=period, ...) # return df return f"Fallback source {self.fallback_source} logic to be implemented for {symbol}." else: return f"All data sources failed for {symbol}. Please check the symbol or network." except Exception as e2: logger.error(f"Fallback source also failed: {e2}") return f"Failed to fetch data for {symbol} after trying all sources." # 可以继续添加获取基本面数据、宏观经济数据等方法

tools/indicator_calculator.py： 专业的指标计算器这里我们使用ta库，它足够轻量且功能全面。

import pandas as pd import ta from typing import List, Dict class IndicatorCalculatorTool: """技术指标计算工具""" def __init__(self): self.available_indicators = ['sma', 'ema', 'rsi', 'macd', 'bollinger', 'stoch'] def calculate_indicators(self, df: pd.DataFrame, indicators: List[str], **params) -> pd.DataFrame: """ 为给定的DataFrame计算指定技术指标。 参数: df: 必须包含 'close', 'high', 'low', 'volume' 列 indicators: 需要计算的指标名称列表 **params: 各指标的自定义参数，如 `sma_window=20` 返回: 添加了指标列的DataFrame """ result_df = df.copy() # 确保必要的列存在 required_cols = ['close', 'high', 'low', 'volume'] for col in required_cols: if col not in result_df.columns: raise ValueError(f"Input DataFrame must contain '{col}' column.") # 计算每个指标 for indicator in indicators: if indicator == 'sma': window = params.get('sma_window', 20) result_df[f'SMA_{window}'] = ta.trend.sma_indicator(result_df['close'], window=window) elif indicator == 'ema': window = params.get('ema_window', 12) result_df[f'EMA_{window}'] = ta.trend.ema_indicator(result_df['close'], window=window) elif indicator == 'rsi': window = params.get('rsi_window', 14) result_df['RSI'] = ta.momentum.rsi(result_df['close'], window=window) elif indicator == 'macd': fast = params.get('macd_fast', 12) slow = params.get('macd_slow', 26) signal = params.get('macd_signal', 9) macd_indicator = ta.trend.MACD(result_df['close'], window_slow=slow, window_fast=fast, window_sign=signal) result_df['MACD'] = macd_indicator.macd() result_df['MACD_Signal'] = macd_indicator.macd_signal() result_df['MACD_Diff'] = macd_indicator.macd_diff() elif indicator == 'bollinger': window = params.get('bb_window', 20) window_dev = params.get('bb_dev', 2) bollinger = ta.volatility.BollingerBands(result_df['close'], window=window, window_dev=window_dev) result_df['BB_Upper'] = bollinger.bollinger_hband() result_df['BB_Middle'] = bollinger.bollinger_mavg() result_df['BB_Lower'] = bollinger.bollinger_lband() result_df['BB_Width'] = bollinger.bollinger_wband() # 布林带宽度，衡量波动率 elif indicator == 'stoch': k_window = params.get('stoch_k', 14) d_window = params.get('stoch_d', 3) stoch_indicator = ta.momentum.StochasticOscillator(result_df['high'], result_df['low'], result_df['close'], window=k_window, smooth_window=d_window) result_df['Stoch_%K'] = stoch_indicator.stoch() result_df['Stoch_%D'] = stoch_indicator.stoch_signal() else: print(f"Indicator '{indicator}' is not supported yet. Skipping.") # 生成简单的信号（示例：基于RSI的超买超卖） if 'RSI' in result_df.columns: result_df['Signal_RSI'] = 0 result_df.loc[result_df['RSI'] < 30, 'Signal_RSI'] = 1 # 超卖，潜在买入信号 result_df.loc[result_df['RSI'] > 70, 'Signal_RSI'] = -1 # 超买，潜在卖出信号 if 'MACD' in result_df.columns and 'MACD_Signal' in result_df.columns: result_df['Signal_MACD'] = 0 # MACD线上穿信号线，金叉 result_df.loc[(result_df['MACD'] > result_df['MACD_Signal']) & (result_df['MACD'].shift(1) <= result_df['MACD_Signal'].shift(1)), 'Signal_MACD'] = 1 # MACD线下穿信号线，死叉 result_df.loc[(result_df['MACD'] < result_df['MACD_Signal']) & (result_df['MACD'].shift(1) >= result_df['MACD_Signal'].shift(1)), 'Signal_MACD'] = -1 return result_df def get_indicators_summary(self, df_with_indicators: pd.DataFrame) -> Dict: """生成指标摘要，便于后续智能体分析""" summary = {} latest = df_with_indicators.iloc[-1] if 'RSI' in df_with_indicators.columns: summary['rsi'] = { 'value': round(latest['RSI'], 2), 'status': 'oversold' if latest['RSI'] < 30 else 'overbought' if latest['RSI'] > 70 else 'neutral' } # ... 可以添加其他指标的摘要 return summary

3.3 智能体与任务组装实战

这是CrewAI框架发挥威力的地方。我们在crew/investment_crew.py中创建团队。

import os from crewai import Agent, Task, Crew, Process from langchain_openai import ChatOpenAI # 或其他LLM from tools.data_fetcher import DataFetcherTool from tools.indicator_calculator import IndicatorCalculatorTool # 假设还有其他工具... import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class InvestmentCrew: """投资分析智能体团队""" def __init__(self, stock_symbol='AAPL'): self.stock_symbol = stock_symbol self.llm = ChatOpenAI( model="gpt-4-turbo-preview", # 或 "gpt-3.5-turbo" temperature=0.1, # 投资分析需要低随机性，保持严谨 api_key=os.getenv('OPENAI_API_KEY') ) self.data_fetcher = DataFetcherTool() self.indicator_calculator = IndicatorCalculatorTool() # 初始化智能体 self.data_engineer_agent = None self.quant_analyst_agent = None self.strategy_researcher_agent = None self.risk_manager_agent = None # 初始化任务 self.fetch_data_task = None self.calculate_indicators_task = None self.generate_report_task = None self.make_decision_task = None self._create_agents() self._create_tasks() def _create_agents(self): """定义团队中的各个角色""" # 1. 数据工程师 self.data_engineer_agent = Agent( role='资深金融数据工程师', goal='准确、高效地获取并清洗指定金融标的的原始市场数据，确保数据完整、格式规范，为后续分析提供可靠基础。', backstory="""你在一家顶级对冲基金的数据部门工作了十年，对全球各大金融数据源的API了如指掌。 你深知垃圾数据进，垃圾分析出的道理，因此你对数据的准确性和时效性有着近乎偏执的追求。 你擅长处理各种数据异常，并能从多个备用源中获取数据以确保万无一失。""", verbose=True, # 输出详细执行日志 allow_delegation=False, # 数据任务通常不需要委托 llm=self.llm, tools=[self.data_fetcher.fetch_stock_data], # 将工具方法绑定给智能体 ) # 2. 量化分析师 self.quant_analyst_agent = Agent( role='严谨的量化策略分析师', goal='基于清洗后的市场数据，计算一系列关键的技术指标和风险指标，并生成客观、量化的初步信号。', backstory="""你拥有数学和金融工程双学位，痴迷于用数学模型解读市场。 你精通各类技术指标（趋势、动量、波动率）的计算和内在逻辑，不相信任何“万能指标”， 但坚信在多因子综合视角下能发现概率优势。你的计算从不出错。""", verbose=True, allow_delegation=False, llm=self.llm, tools=[self.indicator_calculator.calculate_indicators, self.indicator_calculator.get_indicators_summary], ) # 3. 策略研究员 self.strategy_researcher_agent = Agent( role='洞察深刻的投资策略研究员', goal='综合量化信号、当前市场宏观环境（如市场情绪、行业新闻），撰写一份逻辑清晰、论据扎实、包含明确操作建议的投资分析报告。', backstory="""你曾是明星卖方分析师，后转型为买方研究员。你不仅看数字，更关注数字背后的逻辑和市场情绪。 你擅长将复杂的量化信号翻译成易懂的投资故事，同时始终保持独立思考，不随波逐流。 你的报告以结构严谨、风险提示充分而著称。""", verbose=True, allow_delegation=True, # 研究员可以要求量化分析师重新计算某些指标 llm=self.llm, # 策略研究员可能不需要具体工具，主要依靠LLM的分析能力，但可以给他访问新闻摘要的工具 ) # 4. 风控经理 self.risk_manager_agent = Agent( role='保守谨慎的风险控制经理', goal='严格审核投资建议报告，依据公司风控条例（如最大回撤、仓位集中度、黑名单）做出最终执行或否决的决策，并记录完整决策日志。', backstory="""你经历了多次市场牛熊转换，亲眼见过因风控失效而崩溃的基金。 你的信条是“活下去比赚得多更重要”。你手中有一本厚厚的风控手册，任何投资建议都必须通过它的审查。 你冷酷无情，只认规则，不认人情。""", verbose=True, allow_delegation=False, llm=self.llm, tools=[], # 风控规则可以内置在Agent的prompt中，或作为一个独立的工具 ) def _create_tasks(self): """定义团队要完成的具体任务，并建立依赖关系""" # 任务1：获取数据 self.fetch_data_task = Task( description=f"""获取股票代码为 {self.stock_symbol} 的最新市场数据。 要求获取过去3个月的日线级别数据，包含开盘价、最高价、最低价、收盘价和成交量。 如果主要数据源失败，请尝试备用数据源。请确保数据没有缺失值，并将日期列格式化。""", expected_output="""一个干净的Pandas DataFrame（以JSON字符串形式返回）， 至少包含以下列：Date, Open, High, Low, Close, Volume。 如果失败，请返回清晰的错误信息。""", agent=self.data_engineer_agent, output_file='outputs/raw_data.json' # 可选：将输出保存到文件 ) # 任务2：计算指标 (依赖于任务1的输出) self.calculate_indicators_task = Task( description="""基于数据工程师获取的原始价格和成交量数据，计算一套核心的技术指标。 必须计算的指标包括：20日简单移动平均线(SMA)、12日指数移动平均线(EMA)、14日相对强弱指数(RSI)、 标准参数(12,26,9)的MACD及其信号线、20日布林带(2倍标准差)。 基于RSI和MACD，生成初步的量化信号（例如，RSI<30为超卖信号1，RSI>70为超买信号-1；MACD金叉为1，死叉为-1）。""", expected_output="""一个增强的DataFrame（JSON字符串），在原始数据基础上新增了计算出的指标列和信号列。 同时，提供一份关于最新指标数值和信号状态的文本摘要，例如：'最新收盘价XX，RSI为YY，处于超买区域；MACD出现金叉...'。""", agent=self.quant_analyst_agent, context=[self.fetch_data_task], # 关键：此任务需要上一个任务的输出作为上下文 output_file='outputs/data_with_indicators.json' ) # 任务3：生成投资报告 (依赖于任务2的输出) self.generate_report_task = Task( description=f"""你是一名资深策略研究员。请基于量化分析师提供的指标数据和分析摘要，为股票 {self.stock_symbol} 撰写一份专业的投资分析报告。 报告需包含以下部分： 1. 概述：简要说明分析标的和时间范围。 2. 技术面分析：解读关键指标（如趋势指标MA、动量指标RSI/MACD、波动率指标布林带）的当前状态和含义。 3. 综合判断：结合各指标信号，给出看多、看空或中性的整体观点。 4. 操作建议：给出具体的建议，如“买入”、“持有”、“卖出”或“观望”。如果是买入/卖出，可以建议一个参考仓位（如总资金的5%-10%）。 5. 风险提示：明确指出当前决策所依赖的假设和潜在风险（如指标钝化、市场突发新闻等）。 报告应逻辑清晰，论据引用具体数据，语言专业且简洁。""", expected_output="""一份结构完整、论据充分的投资分析报告，字数在300-500字之间。 报告最后必须用【最终建议：XXX】的格式明确输出操作建议。""", agent=self.strategy_researcher_agent, context=[self.calculate_indicators_task], output_file='outputs/investment_report.md' ) # 任务4：风控审核与决策 (依赖于任务3的输出) self.make_decision_task = Task( description=f"""你作为最终风控负责人，需要审核策略研究员为 {self.stock_symbol} 提交的投资报告。 请依据以下风控条例进行审核： - 条例1：如果标的近期（过去5个交易日）最大日内跌幅超过10%，则禁止任何买入建议，维持中性或卖出。 - 条例2：如果RSI指标连续3日处于超买区（>70），则买入建议的仓位不得超过总资金的5%。 - 条例3：如果报告中没有提及任何风险提示，或风险提示过于敷衍，则打回报告要求重写。 你的工作： 1. 仔细阅读投资报告。 2. 检查报告中的分析和数据是否与量化分析师提供的指标摘要相符。 3. 逐条对照风控条例，判断报告建议是否合规。 4. 做出最终决策：通过、修改后通过、或否决。 5. 如果否决或修改，必须给出详细理由。""", expected_output="""一份风控决策记录，包含： - 审核的标的和报告日期。 - 风控条例检查结果（逐条说明）。 - 最终决策：【通过】/【修改后通过】/【否决】。 - 决策详细理由。 - 如果通过，最终的、带风控约束的操作指令（例如：“批准执行买入，但仓位限制在5%以内”）。""", agent=self.risk_manager_agent, context=[self.generate_report_task, self.calculate_indicators_task], # 需要报告和原始数据来检查风控条例 output_file='outputs/risk_decision.md' ) def run(self): """组建团队并启动工作流""" crew = Crew( agents=[ self.data_engineer_agent, self.quant_analyst_agent, self.strategy_researcher_agent, self.risk_manager_agent ], tasks=[ self.fetch_data_task, self.calculate_indicators_task, self.generate_report_task, self.make_decision_task ], process=Process.sequential, # 顺序执行，最符合当前分析流程 verbose=2, # 输出详细的团队执行日志 ) logger.info(f"开始执行投资分析流程，标的：{self.stock_symbol}") result = crew.kickoff() logger.info("流程执行完毕。") return result

最后，在main.py中启动整个流程：

import sys from dotenv import load_dotenv from crew.investment_crew import InvestmentCrew load_dotenv() # 加载 .env 文件中的 OPENAI_API_KEY def main(): if len(sys.argv) > 1: symbol = sys.argv[1] else: symbol = 'AAPL' # 默认分析苹果公司 print(f"启动对 {symbol} 的智能体协同分析...") crew = InvestmentCrew(stock_symbol=symbol) final_result = crew.run() print("\n" + "="*50) print("分析流程最终输出:") print("="*50) print(final_result) if __name__ == "__main__": main()

运行命令：python main.py 000001.SZ即可开始分析平安银行（A股）。

4. 核心环节：智能体协作流程与信息传递剖析

4.1 顺序流程（Sequential Process）下的协作细节

在Crew的Process.sequential模式下，任务会严格按照定义的顺序执行。但这并不意味着智能体之间是孤立的。context参数是连接任务的关键桥梁。

以我们的流程为例：

1. fetch_data_task完成：数据工程师智能体调用DataFetcherTool，成功获取到数据，输出一个DataFrame的JSON字符串。这个输出被自动保存在任务的output属性中。
2. calculate_indicators_task启动：量化分析师智能体开始工作。因为该任务设置了context=[self.fetch_data_task]，CrewAI框架会自动将fetch_data_task的完整输出（即那个JSON字符串）作为“上下文”注入到量化分析师智能体的提示词（Prompt）中。智能体的LLM会看到类似这样的提示：“这是上一个任务（数据工程师）的输出：{...JSON数据...}。你的任务是：基于这些数据计算指标...”。然后，智能体再调用IndicatorCalculatorTool，将JSON数据解析为DataFrame进行计算。
3. 信息流继续：generate_report_task的上下文是[self.calculate_indicators_task]，因此策略研究员能看到指标计算结果和摘要。make_decision_task的上下文是[self.generate_report_task, self.calculate_indicators_task]，因此风控经理既能拿到最终报告，也能回溯查看原始指标数据来验证风控条例（例如检查“过去5日最大跌幅”）。

这种基于上下文的传递，模拟了真实团队中“上一环节交付物作为下一环节输入”的工作模式。关键在于，智能体接收到的是一段结构化的文本信息（JSON字符串），它需要理解这段信息并从中提取所需数据来调用工具。这就要求我们在设计任务expected_output时，要尽量让输出格式规整、信息完整，便于后续解析。

4.2 提示词（Prompt）设计的实战技巧

智能体的表现很大程度上取决于其角色（role）、目标（goal）和背景故事（backstory），它们共同构成了系统提示词（System Prompt）。此外，任务描述（description）是用户提示词（User Prompt）。设计好这些文本是成功的关键。

1. 角色与背景故事要具体、有“人设”

• 差的例子：role='数据分析员',backstory='你负责分析数据。'
• 好的例子：如上文所示，赋予其专业领域、工作年限、性格特点（如“严谨”、“保守”、“痴迷于数学模型”）。这能引导LLM生成更符合该角色思维模式的文本。例如，风控经理的“冷酷无情，只认规则”的人设，会使其在输出决策时更倾向于否决有风险的提议。

2. 任务描述要清晰、可操作、包含约束

• 模糊指令：“分析一下这只股票。”
• 清晰指令：“基于提供的指标数据，撰写报告，必须包含概述、技术面分析、综合判断、操作建议、风险提示五个部分，其中操作建议必须明确为‘买入’、‘持有’或‘卖出’，并在报告末尾以【最终建议：XXX】格式输出。” 清晰的指令能极大减少LLM的随机性，得到格式稳定、内容完整的输出。

3. 善用“期望输出”（expected_output）expected_output不仅用于给人看，也会影响LLM。描述你期望的输出格式（如“JSON字符串”、“包含X部分的Markdown报告”），LLM会努力向这个格式靠拢。这对于后续自动化处理至关重要。

4. 为工具调用提供明确指引在任务描述中，可以提示智能体使用哪个工具。例如，在数据工程师的任务中说“请使用fetch_stock_data工具获取数据”。虽然CrewAI会自动将可用工具列表给LLM，但明确的指引能提高工具调用的准确率。

5. 性能优化、常见问题与排查指南

5.1 成本、延迟与稳定性优化

1. 模型选型与成本控制

• 分层使用LLM：并非所有智能体都需要GPT-4。数据工程师和量化分析师的任务相对结构化，可以使用更便宜、更快的模型，如gpt-3.5-turbo。而策略研究员和风控经理需要进行复杂推理和文本生成，则使用GPT-4。在CrewAI中，可以为每个Agent单独指定llm参数。
• 设置最大Token数：在LLM初始化时设置max_tokens，防止生成过于冗长的内容，尤其是对于数据分析摘要类的任务。
• 缓存结果：对于相同标的、相同分析周期的请求，可以将中间结果（如原始数据、指标数据）缓存到本地文件或数据库，避免重复调用数据API和LLM，这是降低成本和延迟最有效的手段之一。

2. 异步执行提升速度如果任务间没有严格的先后依赖（例如，可以同时分析多只不相关的股票），可以考虑使用Process.hierarchical或其他支持并行的流程，或者自行使用asyncio封装多个Crew的执行。对于顺序流程，单个任务的工具执行（如网络请求）是同步的，这里可能成为瓶颈，可以考虑将工具内部实现为异步。

3. 工具设计的健壮性

• 重试与降级：如DataFetcherTool所示，必须有完善的异常处理和备用方案。
• 超时设置：所有网络请求必须设置超时，避免进程卡死。
• 输入验证：在工具函数内部，严格检查输入参数的有效性，防止错误数据导致后续流程崩溃。

5.2 常见问题与解决方案速查表

5.3 扩展方向与进阶思考

这个基础框架可以沿多个方向深化：

1. 增加智能体类型

• 宏观研究员智能体：专门分析宏观经济新闻、央行政策对整体市场的影响。
• 舆情分析智能体：爬取并分析社交媒体、财经新闻中对特定标的的情绪倾向。
• 复盘分析师智能体：在每日收盘后，对比分析预测与实际走势，总结策略得失，用于迭代优化。

2. 引入更复杂的流程

• 分层审议流程：策略研究员生成初版报告后，不是直接交给风控经理，而是先由一个“资深投资委员会”智能体（由更强大的LLM驱动）进行评议和修改，再提交风控。这模拟了更真实的投研流程。
• 循环优化流程：如果风控经理否决了方案，可以将报告打回给策略研究员，并附上修改意见，形成一个反馈循环。

3. 与实盘系统集成

• 模拟回测：将智能体生成的每日信号接入回测框架（如Backtrader,Zipline），评估长期收益。
• 半自动执行：在风控经理通过后，将决策指令（如“买入5%仓位”）发送到券商的模拟交易API或通过消息通知人工执行。

4. 智能体的持续学习

• 保存历史决策与市场结果：构建一个知识库，记录每次分析的报告、决策、以及之后一段时间的股价表现。
• 定期复盘：让一个专门的智能体学习这些历史数据，分析哪些因子、哪些表述方式更可能导致成功决策，并尝试优化其他智能体的提示词或工具使用策略。

这个easy_investment_Agent_crewai项目就像一个乐高积木的起点，它展示了用多智能体构建复杂应用的强大潜力和相对低的入门门槛。真正的挑战和乐趣，在于如何根据你自己的投资理念和风险偏好，去设计、调试和扩展这个“AI投资团队”，让它从一個玩具，逐渐成长为一个值得信赖的辅助决策伙伴。