在 AI 应用开发领域，单一智能体已难以应对复杂任务需求。CrewAI 作为开源多智能体编排框架，通过模拟现实团队协作模式，让不同角色智能体自主协同完成任务，显著提升 AI 系统处理复杂工作的能力。本文将从框架核心概念出发，结合实战案例，带大家掌握 CrewAI 的使用与开发技巧。

一、CrewAI 框架核心解析

1.1 什么是 CrewAI

CrewAI 是由 João Moura 开发的开源多智能体编排框架，建立在 LangChain 基础之上，兼容多种现有工具。它通过协调多个 AI 智能体的协作来完成复杂任务，模拟现实世界中的工作团队，让不同角色的智能体能够自主地相互委派任务和交流，实现比单一语言模型更强大的性能表现。其官网为https://www.crewai.com/，GitHub 仓库地址为https://github.com/crewAIInc/crewAI。

1.2 核心组件详解

CrewAI 的核心功能围绕五大组件展开，各组件分工明确又相互配合：

Agent（代理）

Agent 是具有特定角色、目标和能力的 AI 实体，就像团队中的成员，有明确的职责定位。关键属性包括：

• role：代理的角色，如 “研究员”“作家”，明确其在团队中的身份。
• goal：代理要实现的目标，指导其工作方向。
• backstory：代理的背景故事，帮助 LLM（大语言模型）更好地理解角色，提升角色代入感。
• tools：代理可以使用的工具列表，是其完成任务的 “武器”。
• llm：使用的语言模型，为代理提供语言理解和生成能力。
• memory：是否启用记忆功能，决定代理能否记住之前的交互信息。

以下是创建一个检索代理的代码示例，该代理主要负责从可用资源中检索与用户查询相关的信息：

from crewai import Agent, LLM
from tools import ddgs_text_search, ddgs_news_search

# 定义LLM初始化函数
def deepseek_llm(**kwargs):
    return LLM(
        model="deepseek/deepseek-chat",
        api_key='sk-',  # 此处需替换为实际API密钥
        api_base='https://api.deepseek.com/v1',
        model_kwargs={"llm_provider": "deepseek"}
    )

# 创建检索代理
retriever_agent = Agent(
    role="检索相关信息以回答用户查询: {query}",
    goal="从可用资源中检索与用户查询最相关的信息: {query},始终优先使用pdf搜索工具。如果无法从pdf搜索工具中检索到信息，则尝试使用网络搜索工具。",
    backstory="你是一位细致入微的分析师，以敏锐的洞察力著称。你以理解用户查询: {query} 并从最合适的知识库中检索信息的能力而闻名。",
    llm=deepseek_llm(),
    tools=[ddgs_text_search, ddgs_news_search],  # 为代理分配网络搜索工具
    verbose=True  # 启用详细日志，便于调试
)

Task（任务）

Task 是代理需要完成的具体工作单元，明确了代理的工作内容和预期成果。关键属性有：

• description：任务的详细描述，让代理清楚要做什么。
• agent：负责执行任务的代理，明确任务的执行者。
• expected_output：期望的输出格式，规定任务完成后的成果形式。
• context：依赖的其他任务，即任务依赖，确保任务按合理顺序执行。
• output_file：输出文件路径，指定任务成果的保存位置。

例如，定义一个检索任务，指定由上述创建的检索代理执行：

from crewai import Task

retrieval_task = Task(
    description="从可用资源中检索与用户查询最相关的信息: {query}",
    expected_output="以文本形式从资源中检索到的最相关信息。",
    agent=retriever_agent  # 指定检索代理执行该任务
)

Crew（团队）

Crew 是代理和任务的组织结构，负责协调整个工作流程，相当于团队的 “管理者”。关键属性包括：

• agents：团队中的代理列表，集合完成任务所需的所有 “成员”。
• tasks：要执行的任务列表，明确团队需要完成的 “工作内容”。
• process：执行流程（顺序 / 分层），规定任务和代理的协作方式。
• verbose：是否显示详细执行日志，方便监控和调试。
• memory：是否启用团队记忆，让团队能共享和利用历史信息。

下面代码创建一个包含检索代理和响应合成代理的团队，并分配相应任务：

from crewai import Crew

# 先创建响应合成代理（代码略，类似检索代理创建）
# ...

# 定义响应合成任务（代码略，类似检索任务定义）
# ...

# 创建团队
crew = Crew(
    agents=[retriever_agent, response_synthesizer_agent],  # 团队成员：检索代理和响应合成代理
    tasks=[retrieval_task, response_task],  # 团队任务：检索任务和响应合成任务
    verbose=True
)

Flow（流程）

Flow 提供更精细的工作流控制，支持条件逻辑、循环和状态管理，像团队工作的 “流程规划师”。其特性有：

• @start：定义流程的起点，标记流程开始的方法。
• @listen：监听特定事件，当被监听的任务完成时触发相应方法。
• @router：根据条件路由，实现流程的动态分支。
• 状态持久化和恢复：确保流程在中断后能继续执行。

以下是一个简单的流程示例，包含流程起点、路由和事件监听：

import random
from crewai.flow.flow import Flow, listen, router, start
from pydantic import BaseModel

# 定义状态模型
class ExampleState(BaseModel):
    success_flag: bool = False

# 定义流程类
class RouterFlow(Flow[ExampleState]):
    # 流程起点
    @start()
    def start_method(self):
        print("Starting the structured flow")
        random_boolean = random.choice([True, False])
        self.state.success_flag = random_boolean

    # 条件路由
    @router(start_method)
    def second_method(self):
        if self.state.success_flag:
            return "success"
        else:
            return "failed"

    # 监听"success"事件
    @listen("success")
    def third_method(self):
        print("Third method running")

    # 监听"failed"事件
    @listen("failed")
    def fourth_method(self):
        print("Fourth method running")

# 实例化并启动流程
flow = RouterFlow()
flow.kickoff(inputs={"success_flag": True})

Tool（工具）

Tool 是智能体用来高效执行任务的工具，如 RAG（检索增强生成）、text2sql、第三方功能等，为代理提供额外的能力支持。CrewAI 支持自定义工具封装，以下是一个使用 DDGS（DuckDuckGo Search）进行网络文本搜索的工具封装示例：

from crewai.tools import tool
from ddgs import DDGS

@tool("DDGS Text Search")
def ddgs_text_search(query: str, max_results: int = 10) -> str:
    """
    使用 DDGS 进行网络文本搜索
    Args:
        query: 搜索关键词
        max_results: 最大结果数量
    Returns:
        格式化的搜索结果字符串
    """
    ddgs = DDGS()
    try:
        results = ddgs.text(
            query=query,
            backend="auto",
            max_results=max_results,
            region='zh-cn'
        )
        output = []
        for i, result in enumerate(results, 1):
            output.append(f"\n结果 {i}:")
            output.append(f"标题: {result.get('title', 'N/A')}")
            output.append(f"链接: {result.get('href', 'N/A')}")
            output.append(f"摘要: {result.get('body', 'N/A')[:200]}...")
            output.append("-" * 50)
        return "\n".join(output)
    except Exception as e:
        return f"搜索失败: {str(e)}"

二、环境搭建与基础使用

2.1 系统要求

• Python 3.10 或更高版本
• pip 或 uv 包管理器

2.2 模块安装

通过 pip 命令即可完成 CrewAI 及其工具模块的安装：

pip install crewai
pip install crewai-tools

2.3 两种实现方式

方式一：完全基于 Python 代码实现

这种方式灵活性高，适合需求多变的场景。以实现一个简单的信息检索与响应系统为例，完整代码如下：

# 导入必要模块
from crewai import Agent, LLM, Task, Crew
from tools import ddgs_text_search, ddgs_news_search  # 导入自定义工具

# 定义LLM初始化函数
def deepseek_llm(**kwargs):
    return LLM(
        model="deepseek/deepseek-chat",
        api_key='sk-',  # 替换为实际API密钥
        api_base='https://api.deepseek.com/v1',
        model_kwargs={"llm_provider": "deepseek"},
        timeout=60
    )

# 创建检索代理
retriever_agent = Agent(
    role="检索相关信息以回答用户查询: {query}",
    goal="从可用资源中检索与用户查询最相关的信息: {query},始终优先使用pdf搜索工具。如果无法从pdf搜索工具中检索到信息，则尝试使用网络搜索工具。",
    backstory="你是一位细致入微的分析师，以敏锐的洞察力著称。你以理解用户查询: {query} 并从最合适的知识库中检索信息的能力而闻名。",
    llm=deepseek_llm(),
    tools=[ddgs_text_search, ddgs_news_search],
    verbose=True
)

# 创建响应合成代理
response_synthesizer_agent = Agent(
    role="用户查询的响应合成器: {query}",
    goal="根据用户查询: {query} 将检索到的信息综合成简洁连贯的响应。如果无法检索到信息，则回应“抱歉，我找不到您要的信息。”",
    backstory="你是一位熟练的沟通者，擅长将复杂的信息转化为清晰简洁的响应。",
    llm=deepseek_llm(),
    verbose=True
)

# 定义检索任务
retrieval_task = Task(
    description="从可用资源中检索与用户查询最相关的信息: {query}",
    expected_output="以文本形式从资源中检索到的最相关信息。",
    agent=retriever_agent
)

# 定义响应合成任务
response_task = Task(
    description="为用户查询合成最终响应: {query}",
    expected_output="基于从正确来源检索到的信息，为用户查询: {query} 提供简洁连贯的响应。如果无法检索到信息，则回应“抱歉，我找不到您要的信息。”",
    agent=response_synthesizer_agent
)

# 创建团队并执行任务
crew = Crew(
    agents=[retriever_agent, response_synthesizer_agent],
    tasks=[retrieval_task, response_task],
    verbose=True
)

# 执行任务并输出结果
crew_output = crew.kickoff(inputs={"query": "大模型如何提高编程的效率?"})
print(crew_output)

方式二：Python + YAML 文件配置实现（推荐）

这种方式将配置与逻辑分离，便于维护和复用。首先通过命令创建项目：

researcher_agent:
  role: >
    Research Agent
  goal: >
    收集关于{topic}和{chapter_title}的全面信息，用于增强章节内容。
    以下是作者对书籍和章节期望目标的一些额外信息:
    {goal}
    以下是章节大纲的描述:
    {chapter_description}
  backstory: >
    你是一位经验丰富的研究员，擅长寻找任何给定主题中最相关和最新的信息。
    你的工作是提供有见地的数据，以支持和丰富章节的写作过程。

writer_agent:
  role: >
    Chapter Writer
  goal: >
    根据提供的章节标题、目标和大纲，撰写一个结构良好的书籍章节。
    章节应以Markdown格式编写，并包含约3,000字。
  backstory: >
    你是一位出色的作家，以创作引人入胜、研究充分且信息丰富的内容而闻名。
    你擅长将复杂的想法转化为可读性强且组织良好的章节。

tasks.yaml用于定义任务描述、分配 Agent 及工具覆盖规则，示例如下：

research_chapter:
  description: >
    研究提供的章节主题、标题和大纲，以收集对撰写章节有帮助的额外内容。
    确保专注于可靠、高质量的信息来源。
    以下是作者对书籍和章节期望目标的一些额外信息:
    {goal}
    以下是章节大纲的描述:
    {chapter_description}
    在研究时，请考虑以下关键点:
    - 你正在研究的章节需要与书籍中的其他章节很好地契合。
    - 你需要收集足够的信息来撰写一个3,000字的章节
  expected_output: >
    以下是整本书的大纲:
    {book_outline}
    一组可用于撰写章节的额外见解和信息。
  agent: researcher_agent

write_chapter:
  description: >
    根据章节标题、目标和大纲描述，撰写一个结构良好的章节。
    每个章节应以Markdown格式编写，并包含约3,000字。
    以下是书籍的主题: {topic}
    以下是章节的标题: {chapter_title}
    以下是章节大纲的描述:
    {chapter_description}
    重要注意事项:
    你正在撰写的章节需要与书籍中的其他章节很好地契合。
    以下是整本书的大纲:
    {book_outline}
  expected_output: >
    一个以Markdown格式编写的约3,000字的章节，涵盖提供的章节标题和大纲描述。
  agent: writer_agent

三、常见问题与资源获取

5.1 常见问题（FAQ）

Q1：CrewAI 和 LangChain 有什么区别？

A：CrewAI 专注于多代理协作，提供更高层次的抽象，方便构建多智能体协同系统；LangChain 是更通用的 LLM 应用框架，提供了丰富的组件和工具。CrewAI 可以使用 LangChain 的组件，但不依赖它。

Q2：CrewAI 支持哪些 LLM？

A：支持所有主流 LLM，包括 OpenAI（GPT-3.5、GPT-4、GPT-4o）、Anthropic Claude、Google Gemini、Azure OpenAI，以及通过 Ollama 等部署的本地模型。

Q3：如何降低 API 成本？

A：可通过以下方式降低 API 成本：

• 启用缓存功能，避免重复调用。
• 对简单任务使用更便宜的模型。
• 设置max_rpm（每分钟最大请求数）和max_iter（最大迭代次数）限制。
• 优化 prompt，减少 token 使用量。