当前位置：首页 > AI > 正文内容

CrewAI 多智能体编排实战教程：构建AI协作团队

廖万里7小时前AI0

CrewAI是当前最热门的多智能体编排框架之一，它让开发者能够像组建团队一样构建AI Agent协作系统。本文将从零开始，带你掌握CrewAI的核心概念、环境搭建、实战案例与最佳实践。

一、核心概念：理解CrewAI的架构设计

在AI Agent开发领域，单智能体系统已经能够处理许多简单任务，但当面对复杂的多步骤工作流时，单智能体往往显得力不从心。CrewAI应运而生，它采用角色扮演和团队协作的理念，让多个AI Agent像真实团队一样协同工作。

1.1 四大核心组件

Agent（智能体）：每个Agent都是一个具有特定角色、目标和工具的独立实体。就像团队中的不同角色——研究员负责收集信息，撰稿人负责内容创作，编辑负责审核润色。在CrewAI中，你可以为每个Agent定义：

Role（角色）：定义Agent的身份定位
Goal（目标）：定义Agent要达成的具体目标
Backstory（背景故事）：为Agent提供上下文，增强决策一致性
Tools（工具集）：Agent可以调用的外部工具，如搜索、代码执行等

Task（任务）：定义具体的工作内容，包括任务描述、预期输出和分配给哪个Agent。每个任务都有明确的输入输出，形成工作流的基本单元。

Crew（团队）：由多个Agent组成的协作团队，定义了团队成员和执行流程。你可以把Crew理解为一个项目组，里面包含了完成项目所需的所有角色。

Process（流程）：定义Agent之间的协作方式。CrewAI支持两种流程模式：

Sequential（顺序执行）：任务按顺序依次执行，前一个任务的输出作为后一个任务的输入
Hierarchical（层级执行）：由一个Manager Agent协调其他Agent的工作，模拟真实团队的层级管理

1.2 工作流程示意

以一个内容创作团队为例，流程如下：

研究Agent接收主题，搜索相关资料
研究Agent输出研究报告
写作Agent基于研究报告撰写初稿
编辑Agent审核修改，输出最终文章

这种流水线式的协作，让每个Agent专注于自己擅长的领域，最终产出高质量成果。

二、环境搭建：从零开始配置CrewAI

2.1 安装依赖

首先确保你的Python版本在3.10以上，然后创建虚拟环境并安装CrewAI：

# 创建虚拟环境
python -m venv crewai-env
source crewai-env/bin/activate  # Windows用: crewai-env\Scripts\activate

# 安装CrewAI核心包
pip install crewai

# 安装CrewAI工具包（包含搜索、文件读写等工具）
pip install "crewai[tools]"

# 验证安装
import crewai
print(crewai.__version__)

2.2 配置LLM后端

CrewAI支持多种大语言模型后端，包括OpenAI、Claude、Ollama等。推荐使用环境变量管理API密钥：

import os

# 方式1：使用OpenAI（推荐）
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"

# 方式2：使用Claude（通过LangChain集成）
os.environ["ANTHROPIC_API_KEY"] = "your-anthropic-api-key"

# 方式3：使用本地模型（Ollama）
# 需要先安装并启动Ollama服务
# ollama serve
# ollama pull llama2

2.3 第一个CrewAI程序

让我们创建一个简单的双Agent协作系统，模拟一个调研报告生成流程：

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool

# 初始化搜索工具
search_tool = SerperDevTool()

# 定义研究员Agent
researcher = Agent(
    role="高级研究员",
    goal="深入研究给定主题，收集全面、准确的信息",
    backstory="""你是一位经验丰富的研究员，擅长从海量信息中筛选出最有价值的内容。
    你有严谨的研究方法论，能够辨别信息的可靠性，并善于总结归纳。""",
    verbose=True,
    allow_delegation=False,
    tools=[search_tool]
)

# 定义撰稿人Agent
writer = Agent(
    role="技术撰稿人",
    goal="将研究结果转化为结构清晰、易于理解的报告",
    backstory="""你是一位资深技术撰稿人，擅长将复杂的技术概念转化为通俗易懂的文字。
    你的文章结构严谨，逻辑清晰，读者反馈一直很好。""",
    verbose=True,
    allow_delegation=True
)

# 定义研究任务
research_task = Task(
    description="研究{topic}的最新发展趋势和技术要点",
    expected_output="一份包含关键发现和引用来源的研究摘要",
    agent=researcher
)

# 定义写作任务
writing_task = Task(
    description="基于研究结果，撰写一份关于{topic}的技术报告",
    expected_output="一篇结构完整、内容丰富的Markdown格式技术报告",
    agent=writer,
    context=[research_task]  # 依赖研究任务的输出
)

# 组建团队
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,
    verbose=True
)

# 执行任务
result = crew.kickoff(inputs={"topic": "AI Agent在企业中的应用"})
print(result)

三、实战案例：构建自动化内容营销团队

现在让我们构建一个更复杂的实战案例——一个能够自动完成内容营销全流程的多智能体系统。这个系统将包含四个Agent：策略师、研究员、撰稿人和SEO专家。

3.1 定义专业工具集

from crewai_tools import (
    SerperDevTool,      # 搜索工具
    ScrapeWebsiteTool,  # 网页抓取工具
    FileReadTool,       # 文件读取工具
    DirectoryReadTool   # 目录读取工具
)

# 初始化工具
search_tool = SerperDevTool()
scrape_tool = ScrapeWebsiteTool()
file_read_tool = FileReadTool()

3.2 创建专业Agent团队

# 1. 内容策略师 - 负责规划内容方向
content_strategist = Agent(
    role="内容策略师",
    goal="分析市场趋势，制定内容策略和主题规划",
    backstory="""你是一位资深内容营销专家，对行业趋势有敏锐的洞察力。
    你擅长识别热门话题，能够从用户需求出发规划内容矩阵。
    你的策略决策都有数据支撑，注重ROI和用户价值。""",
    verbose=True,
    tools=[search_tool, scrape_tool],
    allow_delegation=True
)

# 2. 高级研究员 - 负责深度调研
senior_researcher = Agent(
    role="高级研究员",
    goal="深入研究指定主题，收集高质量参考资料",
    backstory="""你是一位严谨的研究员，拥有丰富的信息检索和分析经验。
    你能够快速定位权威信息源，甄别信息真伪，并提供结构化的研究输出。
    你擅长使用各种搜索技巧和工具来获取深层次信息。""",
    verbose=True,
    tools=[search_tool, scrape_tool, file_read_tool],
    allow_delegation=False
)

# 3. 内容创作者 - 负责撰写内容
content_creator = Agent(
    role="内容创作者",
    goal="创作高质量、原创性强的内容",
    backstory="""你是一位才华横溢的内容创作者，文字功底扎实。
    你擅长将复杂概念简化表达，文章可读性强，富有感染力。
    你的作品曾在多个平台获得高阅读量和好评。""",
    verbose=True,
    tools=[file_read_tool],
    allow_delegation=False
)

# 4. SEO专家 - 负责优化搜索引擎表现
seo_specialist = Agent(
    role="SEO优化专家",
    goal="优化内容的搜索引擎表现，提升可发现性",
    backstory="""你是一位SEO领域的专家，精通搜索引擎算法和优化技巧。
    你了解关键词布局、元描述优化、内链外链策略等SEO最佳实践。
    你的优化工作总能显著提升内容的搜索排名和流量。""",
    verbose=True,
    tools=[search_tool],
    allow_delegation=False
)

3.3 定义任务流程

from datetime import datetime

# 任务1：市场分析
market_analysis_task = Task(
    description="""
    分析{industry}行业的最新内容营销趋势。
    需要识别：
    1. 当前最热门的内容主题
    2. 竞争对手的内容策略
    3. 用户最关心的问题
    4. 内容形式偏好（文章、视频、图文等）
    """,
    expected_output="一份详细的市场分析报告，包含热门主题列表和内容建议",
    agent=content_strategist
)

# 任务2：深度调研
research_task = Task(
    description="""
    基于市场分析结果，选择最有价值的主题进行深度调研。
    收集：
    1. 权威资料来源
    2. 行业专家观点
    3. 数据和案例
    4. 相关技术细节
    """,
    expected_output="一份完整的研究报告，包含所有关键信息和引用来源",
    agent=senior_researcher,
    context=[market_analysis_task]
)

# 任务3：内容创作
content_creation_task = Task(
    description="""
    基于研究报告，创作一篇高质量的技术文章。
    要求：
    1. 标题吸引眼球，体现价值
    2. 结构清晰，层次分明
    3. 内容丰富，信息密度高
    4. 语言流畅，易于理解
    5. 包含实用的代码示例
    """,
    expected_output="一篇完整的Markdown格式文章，字数3000-5000字",
    agent=content_creator,
    context=[research_task]
)

# 任务4：SEO优化
seo_optimization_task = Task(
    description="""
    对文章进行SEO优化：
    1. 优化标题和副标题
    2. 添加关键词元数据
    3. 优化文章结构和内链
    4. 添加吸引人的摘要
    """,
    expected_output="优化后的完整文章，包含SEO元数据",
    agent=seo_specialist,
    context=[content_creation_task]
)

3.4 组建并运行团队

# 创建内容营销团队
content_crew = Crew(
    agents=[content_strategist, senior_researcher, content_creator, seo_specialist],
    tasks=[market_analysis_task, research_task, content_creation_task, seo_optimization_task],
    process=Process.sequential,
    verbose=True,
    memory=True  # 启用记忆功能，Agent可以记住之前的交互
)

# 执行任务
print("开始执行内容营销任务...")
result = content_crew.kickoff(inputs={
    "industry": "AI Agent开发"
})

# 保存结果
output_file = f"content_output_{datetime.now().strftime('%Y%m%d_%H%M%S')}.md"
with open(output_file, "w", encoding="utf-8") as f:
    f.write(str(result))

print(f"内容已保存到: {output_file}")

四、高级特性与最佳实践

4.1 使用层级流程（Hierarchical Process）

当任务复杂度高、需要动态分配时，层级流程可以让Manager Agent自动分配任务：

from crewai import Agent, Task, Crew, Process

# Manager Agent负责协调
manager = Agent(
    role="项目经理",
    goal="协调团队成员，确保项目按时高质量完成",
    backstory="你是一位经验丰富的项目经理，擅长资源调配和进度管理。",
    allow_delegation=True,
    verbose=True
)

# 使用层级流程
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[task1, task2, task3],
    process=Process.hierarchical,
    manager_llm="gpt-4",  # 指定Manager使用的LLM
    verbose=True
)

4.2 自定义工具开发

你可以为Agent开发自定义工具，扩展其能力边界：

from crewai_tools import tool
import requests

@tool("获取股票价格")
def get_stock_price(ticker: str) -> str:
    """获取指定股票代码的实时价格"""
    # 使用免费API获取股票数据
    url = f"https://api.example.com/stock/{ticker}"
    response = requests.get(url)
    if response.status_code == 200:
        data = response.json()
        return f"股票 {ticker} 当前价格: ${data['price']}"
    return f"无法获取股票 {ticker} 的价格"

# 在Agent中使用自定义工具
financial_agent = Agent(
    role="金融分析师",
    goal="分析股票市场并提供投资建议",
    backstory="你是一位专业的金融分析师",
    tools=[get_stock_price]
)

4.3 记忆与上下文管理

CrewAI支持短期记忆和长期记忆，让Agent能够在多次交互中保持上下文：

crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    memory=True,  # 启用记忆
    verbose=True
)

# 记忆类型
# - 短期记忆：存储当前会话的上下文
# - 长期记忆：使用向量数据库持久化存储重要信息