我在 2024 年初次接触 CrewAI 时，被它“多个 AI 代理协作”的概念深深吸引。但当我尝试让两个代理互相传递消息时，遇到了消息丢失、格式错误、延迟过高等一系列问题。今天，我决定把这半年来踩过的坑整理成教程，帮助和我一样的初学者少走弯路。
一、为什么 CrewAI 需要消息队列
想象一下，你让一个“研究员”代理去搜集资料，再让一个“作家”代理根据资料写文章。如果它们之间没有有效的通信机制，作家只能干等着，研究员也不知道什么时候该开始工作。CrewAI 的消息队列就是解决这个问题的——它让代理之间可以异步、高效地传递信息。
在实际业务场景中，比如构建一个自动化客服系统，我们需要：
- 意图识别代理负责判断用户问题类型
- 知识库查询代理根据意图从数据库获取答案
- 回复生成代理将答案组织成自然语言
这三个代理之间必须有序协作，而 HolySheep AI 的 API 凭借国内直连延迟低于 50ms 的优势，能让整个通信过程几乎无感知，非常适合这种实时性要求高的场景。
二、CrewAI 消息通信基础架构
2.1 消息队列核心组件
CrewAI 的消息队列主要由三部分组成：生产者（消息发送方）、消费者（消息接收方）和消息代理（Message Broker）。理解这三个角色的关系，是设计稳定通信协议的前提。
在 CrewAI 中，每个代理既是生产者也是消费者。当代理 A 需要代理 B 完成某项任务时，A 会将任务封装成消息放入队列，B 从队列中取出消息并处理。这个过程就像快递员把包裹放进驿站，收件人自己去取——双方不需要同时在线。
2.2 消息格式设计
标准的 CrewAI 消息包含以下字段：
from crewai import Agent, Task, Crew
from pydantic import BaseModel
from typing import Optional, Dict, Any
from datetime import datetime
class CrewMessage(BaseModel):
"""CrewAI 标准消息格式"""
message_id: str # 全局唯一标识
sender: str # 发送方代理名称
receiver: str # 接收方代理名称
content: str # 消息正文
metadata: Dict[str, Any] # 元数据（如优先级、截止时间）
timestamp: datetime # 创建时间
message_type: str # 消息类型：task/request/response/notification
创建一条示例消息
demo_message = CrewMessage(
message_id="msg-20240115-001",
sender="researcher",
receiver="writer",
content="已完成关于人工智能发展趋势的研究，共收集了12篇文献",
metadata={"source_count": 12, "confidence": 0.95},
timestamp=datetime.now(),
message_type="response"
)
print(f"消息ID: {demo_message.message_id}")
print(f"发送方: {demo_message.sender}")
print(f"内容: {demo_message.content}")三、动手实战：搭建第一个多代理通信系统
3.1 环境准备与配置
首先，我们需要安装必要的依赖包。建议使用 Python 3.10 或更高版本，以获得完整的类型提示支持。
# 创建虚拟环境（推荐）
python -m venv crewai_env
source crewai_env/bin/activate # Windows 用户使用 crewai_env\Scripts\activate
安装核心依赖
pip install crewai langchain-openai pydantic redis python-dotenv
验证安装
python -c "import crewai; print(crewai.__version__)"3.2 配置 HolySheep AI API
接下来是关键步骤——配置我们的 API 密钥。我强烈推荐使用 HolySheep AI，因为它不仅支持微信和支付宝充值（这对国内开发者太友好了），而且汇率是 ¥1=$1，对比官方 ¥7.3=$1 的汇率，能节省超过 85% 的成本。
import os
from dotenv import load_dotenv
加载环境变量
load_dotenv()
方式一：直接从环境变量读取
os.environ["OPENAI_API_KEY"] = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
方式二：在代码中直接设置（仅用于演示，生产环境请使用环境变量）
强烈建议：先在 https://www.holysheep.ai/register 注册获取 Key
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # 替换为你的真实 Key
API_BASE = "https://api.holysheep.ai/v1"
print(f"API Base URL: {API_BASE}")
print(f"Key 前5位: {API_KEY[:5]}..." if len(API_KEY) > 5 else "请配置正确的 API Key")配置完成后，你的 CrewAI 应用会通过 HolySheep AI 的代理服务器转发请求。实测国内直连延迟低于 50ms，对于多代理协作场景来说，这个延迟完全可以接受。
3.3 定义代理与任务
现在让我们定义两个简单的代理：一个负责接收用户问题，一个负责分析问题并给出建议。
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI
from crewai.tools import BaseTool
from typing import List
初始化 LLM，连接到 HolySheep AI
llm = ChatOpenAI(
model="gpt-4o",
openai_api_base="https://api.holysheep.ai/v1",
openai_api_key="YOUR_HOLYSHEEP_API_KEY",
temperature=0.7
)
定义问题分析代理
analyzer = Agent(
role="技术问题分析专家",
goal="快速准确地分析用户提出的技术问题",
backstory="你是一位拥有10年经验的技术顾问，擅长将复杂问题分解为简单步骤",
verbose=True,
allow_delegation=False,
llm=llm
)
定义解决方案生成代理
solver = Agent(
role="解决方案工程师",
goal="根据分析结果生成清晰的解决方案",
backstory="你是一位热情的工程师，喜欢用代码和实例帮助他人解决问题",
verbose=True,
allow_delegation=False,
llm=llm
)
定义任务
analysis_task = Task(
description="分析用户的技术问题，识别关键信息：问题类型、难度等级、所需技能",
expected_output="一份结构化的问题分析报告，包含问题分类和解决思路",
agent=analyzer
)
solution_task = Task(
description="基于分析报告，生成详细、可操作的解决方案",
expected_output="包含步骤说明、代码示例和注意事项的完整方案",
agent=solver,
context=[analysis_task] # 关键：此任务依赖分析任务的结果
)
创建crew并执行
crew = Crew(
agents=[analyzer, solver],
tasks=[analysis_task, solution_task],
process=Process.sequential, # 顺序执行，确保 solver 能拿到 analyzer 的结果
verbose=True
)
启动任务
result = crew.kickoff(inputs={"user_question": "如何在 Python 中实现异步消息队列？"}