用 CrewAI 搭一个基于 GLM-5.3-Flash 的多智能体股票分析研究员
云朵君约 13 分钟如果让一个 Agent 一口气去查财务数据、算技术指标、搜新闻，最后再给出一条 BUY、HOLD 或 SELL 信号，它当然也能干。但很快就会碰到一个很现实的问题：查数据、解释数据、处理互相矛盾的结论，全压在一个 Agent 身上，出了问题到底该从哪里查？
所以今天云朵君想拿一个很具体的例子，和大家完整拆一遍：怎么用 GLM-5.3-Flash + CrewAI 搭一个多智能体股票研究员。我们会从 Agent、Tool、Task、Crew 到底是什么讲起，再看为什么数值计算最好交给 Python，三位专业分析师和一位投资组合经理又是怎么把结果一层层传下去的。
先把边界说在前面：这套示例更适合用来学习多智能体的任务拆分和工程设计，不是一个可以直接拿去做投资决策的系统。文中涉及的价格、Benchmark、运行时长和截图，都是 2026 年 8 月 31 日发布时的资料口径，不是云朵君这次重新实测的数据；CrewAI、Z.AI 和 Tavily 的基础概念则按当前官方文档做了核对。
01先把四个角色摆在桌面上
初学者最容易把“多智能体”理解成“多开几个聊天窗口”。其实不一样。
Agent 是一个有角色、目标、背景故事、模型和工具权限的执行者。 Tool 是 Agent 可以调用的确定性能力，例如查询财务数据、计算 RSI，或者搜索新闻。 Task 是交给 Agent 的具体工作，通常包含任务描述、期望输出和负责它的 Agent。 Crew 是把多个 Agent 和 Task 组织起来的团队； Process则决定这些任务按什么方式执行。
这套股票研究示例把工作拆成四个角色：
股票代码
├── 基本面分析师：Finnhub → 财务健康、估值、盈利能力
├── 技术分析师：Yahoo 历史价格 → SMA、RSI、MACD、趋势
├── 新闻分析师：Tavily → 最近 30 天的重要新闻和催化因素
└── 投资组合经理：读取前三份报告 → BUY / HOLD / SELL
拆分的价值不在于“人数更多”，而在于每个 Agent 的输入边界更清楚。基本面分析师不用自己搜索新闻，技术分析师也不用凭语言模型记忆计算指标。模型主要负责解释和组织，数据读取与数值计算交给工具。
我觉得这套设计里最值得记住的工程取舍就是：**先把任务拆成可检查的职责，再让模型处理需要语言理解的部分。**它不会自动消除错误，但至少能让错误更容易定位。
02先准备 API Key 和 Python 环境
这套示例会用到三个外部服务：Z.AI 提供 GLM API，Finnhub 提供基本面数据，Tavily 负责新闻搜索。Z.AI 的官方文档提供了常规 API 端点、Bearer Key 认证和 OpenAI Python SDK 的调用方式；这里要注意，常规 API 和 Coding Plan 是不同用途，不能混用。
项目先放一个 .env 文件：
ZAI_API_KEY="your_zai_api_key"
FINNHUB_API_KEY="your_finnhub_api_key"
TAVILY_API_KEY="your_tavily_api_key"
API Key 不要写进 Notebook，也不要提交到 Git。对于刚开始学习的人，Jupyter Notebook 是一个合适的起点：每一步的返回值都能单独检查，先确认工具输出，再接入 Agent，排错会比直接启动完整 Crew 简单。
依赖先装这些：
!pip install -q crewai crewai-tools finnhub-python tavily-python pandas numpy requests python-dotenv openai
然后加载库和环境变量：
import os
import time
import finnhub
import numpy as np
import pandas as pd
import requests
from dotenv import load_dotenv
from tavily import TavilyClient
from crewai import Agent, Crew, LLM, Process, Task
from crewai_tools import TavilySearchTool
from crewai.tools import tool
load_dotenv()
os.environ.setdefault("CREWAI_TRACING_ENABLED", "false")
os.environ.setdefault("OTEL_SDK_DISABLED", "true")
TICKER = "NVDA"
TICKER 只是示例股票代码，可以换成 AAPL、MSFT 或 TSLA。但股票代码能换，不代表结果就自动可靠：数据日期、缺失字段、币种和接口返回口径仍然要一起检查。
接着初始化两个数据服务和模型：
finnhub_client = finnhub.Client(
api_key=os.environ["FINNHUB_API_KEY"]
)
tavily_client = TavilyClient(
api_key=os.environ["TAVILY_API_KEY"]
)
llm = LLM(
model="openai/glm-5.3-flash",
api_key=os.environ["ZAI_API_KEY"],
base_url="https://api.z.ai/api/paas/v4/",
temperature=0.1,
)
这里的 base_url 体现了 OpenAI-compatible API 的用法：CrewAI 仍然使用自己的 LLM 封装，但请求通过 Z.AI 的兼容接口发送。temperature=0.1 是这里针对金融研究场景采用的配置，目的是让输出更一致、更聚焦，不是“低温度就更准确”的保证。当前官方文档展示的模型名和这里的字符串可能存在版本差异，真正运行前要以当前模型目录和账号权限为准。
03先让 Python 算指标，别让 Agent 猜数字
技术分析师先要拿到一段历史价格数据。这里调用 Yahoo Finance 的公共 chart endpoint，抓取每日开盘价、最高价、最低价、收盘价和成交量，再在本地计算指标。
HTTP = requests.Session()
HTTP.headers.update({"User-Agent": "glm-stock-swarm/1.0"})
def get_price_history(ticker: str, days: int = 450) -> pd.DataFrame:
end = int(time.time())
start = end - days * 24 * 60 * 60
response = HTTP.get(
f"https://query1.finance.yahoo.com/v8/finance/chart/{ticker.strip().upper()}",
params={
"period1": start,
"period2": end,
"interval": "1d",
"events": "history",
},
timeout=30,
)
response.raise_for_status()
result = response.json()["chart"]["result"][0]
values = result["indicators"]["quote"][0]
return pd.DataFrame({
"date": pd.to_datetime(result["timestamp"], unit="s", utc=True)
.tz_localize(None),
"open": values["open"],
"high": values["high"],
"low": values["low"],
"close": values["close"],
"volume": values["volume"],
}).dropna(subset=["close"]).sort_values("date").reset_index(drop=True)
这里有一个容易忽略的条件：后面要算 200 日均线，所以数据量至少要覆盖 200 个交易日。代码里请求约 450 天，并在使用前检查行数是否足够。
指标计算可以直接写成一个普通 Python 函数：
def add_indicators(frame: pd.DataFrame) -> pd.DataFrame:
if frame.empty:
return frame.copy()
result = frame.copy()
result["SMA20"] = result["close"].rolling(20).mean()
result["SMA50"] = result["close"].rolling(50).mean()
result["SMA200"] = result["close"].rolling(200).mean()
delta = result["close"].diff()
gain = delta.clip(lower=0)
loss = -delta.clip(upper=0)
avg_gain = gain.ewm(alpha=1 / 14, min_periods=14, adjust=False).mean()
avg_loss = loss.ewm(alpha=1 / 14, min_periods=14, adjust=False).mean()
result["RSI14"] = 100 - (
100 / (1 + avg_gain / avg_loss.replace(0, np.nan))
)
result["EMA12"] = result["close"].ewm(span=12, adjust=False).mean()
result["EMA26"] = result["close"].ewm(span=26, adjust=False).mean()
result["MACD"] = result["EMA12"] - result["EMA26"]
return result
这几个名字可以先这样理解：SMA 是一段时间的平均价格，用来看趋势；RSI 是 0 到 100 的动量指标，这里用 70 和 30 作为超买、超卖的常见参考；MACD 是两条指数移动平均线的差，用来观察趋势和动量变化。它们不是买卖按钮，而是提供给技术分析师解释的结构化输入。
04把数据源封装成 Agent 能调用的 Tool
普通函数只有在 Python 代码里被直接调用时才工作。CrewAI 的 @tool 装饰器，则把它登记成 Agent 可以选择调用的工具。工具的职责是返回数据，Agent 的职责是说明这些数据意味着什么。
先写一个处理缺失值和小数位的辅助函数，再封装基本面工具：
def _number(value, decimals: int = 2) -> str:
if value is None or pd.isna(value):
return "N/A"
return f"{float(value):,.{decimals}f}"
@tool("Get Stock Fundamentals")
def get_fundamentals(ticker: str) -> str:
"""Get current price and company fundamentals from Finnhub."""
ticker = ticker.strip().upper()
quote = finnhub_client.quote(ticker)
metrics = finnhub_client.company_basic_financials(ticker, "all")
m = metrics.get("metric", {})
return f"""Ticker: {ticker}
Source: Finnhub
Current price: {_number(quote.get('c'))}
Previous close: {_number(quote.get('pc'))}
Daily change %: {_number(quote.get('dp'))}
Market cap (USD millions): {_number(m.get('marketCapitalization'))}
Normalized annual P/E: {_number(m.get('peNormalizedAnnual'))}
ROE TTM: {_number(m.get('roeTTM'))}
Net margin TTM: {_number(m.get('netProfitMarginTTM'))}
Revenue growth TTM YoY: {_number(m.get('revenueGrowthTTMYoy'))}"""
N/A 不能当成 0。比如某个字段缺失，零代表“数值确实为零”，而 N/A 代表“这次没有拿到数据”，两者对投资研究是完全不同的信号。
技术工具则复用刚才的计算函数：
@tool("Analyze Stock Technicals")
def get_technicals(ticker: str) -> str:
"""Calculate price trends, returns, moving averages, RSI, and MACD."""
ticker = ticker.strip().upper()
frame = add_indicators(get_price_history(ticker))
if len(frame) < 200:
return f"Technical data unavailable for {ticker}: fewer than 200 observations."
latest = frame.iloc[-1]
return f"""Ticker: {ticker}
Source: Yahoo public chart
Last market date: {latest['date'].date()}
Close: {_number(latest['close'])}
SMA20 / SMA50 / SMA200: {_number(latest['SMA20'])} / {_number(latest['SMA50'])} / {_number(latest['SMA200'])}
RSI14: {_number(latest['RSI14'])}
MACD: {_number(latest['MACD'])}
20-day return: {_number((latest['close'] / frame['close'].iloc[-21] - 1) * 100)}%"""
新闻分析师使用 Tavily 的搜索工具。这里把主题限定为 news，搜索深度设为 advanced，时间范围限制在最近 30 天，最多返回 5 条结果：
web_search = TavilySearchTool(
api_key=os.environ["TAVILY_API_KEY"],
topic="news",
search_depth="advanced",
days=30,
max_results=5,
)
这些参数不是越大越好。新闻范围太宽，经理 Agent 后面要处理的材料就会变多；结果太少，又可能漏掉重要事件。更重要的是，搜索结果必须保留日期和来源链接，不能只把一段摘要交给下游。
05Agent 只拿自己需要的工具
Tool 准备好以后，才创建 Agent。三个专家各自只拿到需要的工具：
fundamental_agent = Agent(
role="Fundamental Analyst",
goal="Evaluate financial health, growth, profitability, valuation, and balance-sheet risk.",
backstory=(
"You are a careful long-term equity analyst. "
"Use the fundamentals tool and never invent a figure. "
"Treat N/A as missing, not as zero."
),
tools=[get_fundamentals],
llm=llm,
allow_delegation=False,
max_iter=4,
verbose=False,
)
technical_agent = Agent(
role="Technical Analyst",
goal="Determine whether the current price setup is bullish, bearish, or neutral.",
backstory=(
"You interpret indicators calculated by Python. "
"Never guess market prices or indicator values."
),
tools=[get_technicals],
llm=llm,
allow_delegation=False,
max_iter=4,
verbose=False,
)
news_agent = Agent(
role="Financial News Analyst",
goal="Find material recent developments and distinguish confirmed reporting from speculation.",
backstory=(
"You are a skeptical financial-news researcher. "
"Prioritize primary and reputable sources, dates, and links."
),
tools=[web_search],
llm=llm,
allow_delegation=False,
max_iter=5,
verbose=False,
)
manager_agent = Agent(
role="Portfolio Manager",
goal="Combine the specialist reports into a balanced, evidence-based research signal.",
backstory=(
"You lead an equity research team. Weigh bullish and bearish evidence, "
"use HOLD when evidence is mixed, and never add unsupported facts."
),
llm=llm,
allow_delegation=False,
max_iter=4,
verbose=False,
)
allow_delegation=False 表示不允许 Agent 再把任务转给别的 Agent，流程会更可预测。max_iter 是一次任务里允许的推理和工具调用循环上限，可以防止某个 Agent 一直调用工具不结束。它们是控制成本和失败面的工程参数，不是质量保证开关。
经理 Agent 没有直接拿到股票工具。它的工作不是重新查数据，而是读取三个专家的报告。这样“取数”和“综合”就分成了两层。
06Task 决定每个 Agent 交什么报告
创建 Agent 只是定义“谁在工作”，Task 才定义“这次要交什么”。例如基本面任务可以要求返回评分、最强优点、最大风险和缺失数据说明：
fundamental_task = Task(
description=(
f"Analyze {TICKER}'s growth, profitability, valuation, balance sheet, "
"and business quality. Use Get Stock Fundamentals. "
"Return a Fundamental Score from 0-100, strongest positive, "
"biggest risk, and note all material missing data."
),
expected_output=(
"A concise fundamental assessment with a 0-100 score, evidence, "
"strongest positive, biggest risk, and missing-data note."
),
agent=fundamental_agent,
)
technical_task = Task(
description=(
f"Analyze {TICKER}'s technical setup. Use Analyze Stock Technicals. "
"Consider SMA20/50/200, RSI14, MACD, recent returns, and trend. "
"Return Technical Score 0-100, Bullish/Neutral/Bearish signal, "
"trend, momentum, and main technical risk."
),
expected_output="A concise technical assessment with score, signal, trend, momentum, data source, and main risk.",
agent=technical_agent,
)
news_task = Task(
description=(
f"Research the most important recent news for {TICKER}, focusing on the last 30 days. "
"Cover earnings, guidance, products, partnerships, M&A, regulation, lawsuits, "
"management, and industry developments. Use Tavily Search. "
"Return News Score 0-100, sentiment, catalysts, dates, and source URLs."
),
expected_output="A sourced recent-news assessment with score, sentiment, catalysts, dates, and clickable URLs.",
agent=news_agent,
)
这里的 description 是任务说明，expected_output 是输出契约，agent 指定负责人。对初学者来说，最重要的不是把提示词写得很长，而是让每份报告的字段足够接近，方便最后比较。
经理任务通过 context 接收前三个任务的输出：
decision_task = Task(
description=(
f"Review all three specialist reports and make the final educational research assessment for {TICKER}. "
"Return exactly one signal: BUY, HOLD, or SELL. "
"Use HOLD when evidence is mixed or confidence is insufficient. "
"Base every claim only on the supplied reports."
),
expected_output=(
"Ticker; Signal; Overall Score /100; Confidence %; "
"Fundamental, Technical, and News scores; Bull Case; Bear Case; "
"Main Catalyst; Main Risk; Final Explanation; Sources; "
"and an educational-not-financial-advice disclaimer."
),
agent=manager_agent,
context=[fundamental_task, technical_task, news_task],
)
context 不等于共享数据库，它更接近“把上游任务的输出放进下游任务的输入”。如果上游报告漏了日期、来源或缺失值说明，经理 Agent 也不应该自己补一个看似合理的数字，所以任务里明确写了“只使用 supplied reports”。
07用 sequential 把团队跑起来
最后把 Agent 和 Task 放进 Crew：
crew = Crew(
agents=[
fundamental_agent,
technical_agent,
news_agent,
manager_agent,
],
tasks=[
fundamental_task,
technical_task,
news_task,
decision_task,
],
process=Process.sequential,
verbose=False,
tracing=False,
)
result = await crew.kickoff_async()
print(result.raw)
Process.sequential 表示任务按顺序执行：三个专家先完成，经理最后汇总。kickoff_async() 只是用异步方式启动这次工作流，不代表内部每一步都并行；实际是否并发，仍由 Crew 的流程配置和任务依赖决定。result.raw 取出最终原始文本，方便先观察输出，再决定是否要解析成 JSON 或保存到数据库。
一条请求可能触发多次模型调用：Agent 要推理、选择工具、阅读工具结果、继续循环，最后才生成报告。可以用一个简单的数量级来理解成本：如果一个 Agent 平均触发约 7 次模型调用，5 个 Agent 就可能产生 35 次以上调用。具体次数取决于配置，但多 Agent 的成本和延迟确实会随着层数一起增加。
所以我不会把“拆成四个 Agent”当成默认答案。任务边界清楚、每个角色有不同数据源时，拆分有帮助；如果只是把同一段 Prompt 分给四个 Agent，最后再让第五个 Agent 重述一遍，得到的可能只是更多 token 和更难排错的上下文。
08从 Notebook 走向终端应用
Notebook 适合验证单个工具和观察每一步输出。Notebook 版本跑通后，还可以把完整工作流收进一个终端界面项目：
git clone https://github.com/kingabzpro/glm-stock-swarm.git
cd glm-stock-swarm
uv sync
uv run glm-stock-swarm
终端界面接收股票代码，依次展示基本面、技术面、新闻和经理汇总；之后的追问则复用已经生成的研究报告，不必每问一次都重新启动四个 Agent。这个设计把一次昂贵的研究和后续便宜的问答分开了。
如果我来把它继续工程化，下一步会先补四件事：锁定依赖版本，给每个外部请求加超时与重试分类，给新闻和价格数据加日期/新鲜度校验，再把最终报告改成结构化 Schema。这样做不是为了让 Demo 看起来更复杂，而是让错误能被确定地发现。
09写在最后
GLM-5.3-Flash 的第三方指标、API 价格、输出速度，以及完整工作流约 4—5 分钟的运行时长，都只能当作这套示例发布时的参考。它们可以帮助我们理解，为什么低价模型适合拿来做多次调用的实验，但价格会变，Benchmark 会变，模型路由和库 API 也会变，不能直接当成当前报价，更不是云朵君这次的实测结果。
同样，最终的 BUY、HOLD、SELL 只是一个输出格式。它不等于经过审计的数据管线，也不等于可执行的交易建议。真实系统至少还需要处理公司行动、停牌、时区、数据延迟、异常值、新闻重复、来源可信度和人工复核。
这套示例真正适合学习的是另一件事：如何把一个“大而模糊的研究任务”拆成几个有工具、有输入、有输出、有边界的步骤。模型负责在这些步骤里做判断，Python 和外部服务负责把数据拿回来，CrewAI 负责把任务连起来。
做到这里，读者已经可以看懂这套 CrewAI 股票研究员的骨架，也知道它离生产系统还差哪些工程门槛。先从一个工具、一个 Agent、一个 Task 跑通，再逐步增加角色，通常比一开始就搭完整“智能研究团队”更容易找到问题。