对比5个最需要了解的AI多智能体编排框架【3】：角色扮演系的CrewAI - 文章 - 开发者社区

picture.image

点击上方蓝字关注我们

在了解轻量级的Swam与重量级的LangGraph两个多智能体编排框架后，我们来了解另一个较早推出且专注于多智能体系统的开源框架CrewAI。本篇内容：

快速认识CrewAI
用CrewAI实现多智能体Demo
CrewAI新特性：更强大的Flows
CrewAI特点总结

快速认识CrewAI

picture.image

CrewAI是一个基于角色扮演、天生为构建具备自主能力的多智能体系统而设计的高层Python开发框架。 基于CrewAI你可以快速组建一个多智能体团队，并通过团队内多个智能体成员的自主决策与无缝协作来完成一系列任务。

picture.image

CrewAI不是一个为了通用LLM应用而设计的底层框架，恰恰相反，它是基于底层框架LangChain之上的更高层抽象，提供了一组专用于多智能体系统构建所需要的模块与工具。因此如果你的应用场景不适合多智能体，并不建议使用CrewAI实现。

picture.image

CrewAI的核心机制可以用下图表示，图中框内展示了最核心的CrewAI组件及其协作机制。

picture.image

CrewAI的核心概念是Crew。简单的总结Crew： Crew就是多个智能体组成的小团队。团队中的多个智能体(Agent)根据设定的流程策略(Process)，借助必要的工具(Tools)，自主协作完成一系列任务(Task)。

与Crew相关的核心概念包括：

【Agent：智能体】

代表一个独立的智能体，在CrewAI中通过简单的定义即可创建， 每个智能体有其扮演的角色、目标、背景、可以使用的工具与LLM等 。Agent可以自主作出决策(借助LLM)以执行某个任务(Task)、必要时可以使用某个工具(Tool），还可以与其他智能体通信，比如转发任务。

【Tools：工具】

智能体可以借助工具来拓展自己的能力，比如调用外部函数与算法、访问外部API、访问数据库、检索必要的知识等。

【Task：任务】

任务是输入或拆分出的工作内容。比如“生成xx风格xx主题的小红书po文”，“搜索最新的前五名热点新闻”等。任务必须由某个Agent完成，这个Agent可以是直接指定，也可能由某个管理的Agent派发，你甚至可以指定任务需要的工具。

【Process：流程】

流程用来确保任务按照指定的策略被分配与执行。在CrewAI中，这里的流程（Process）是由CrewAI框架来负责协调，你只能指定流程的策略。最新的CrewAI中流程只能支持 顺序与层级 两种流程策略。

顺序流程 ：多个Agent按顺序执行任务，确保任务有序完成。
层次结构 ：也就是之前例子中的supervisor模式，即由一个管理Agent负责任务的管理与派发，而任务的执行由其他的工作Agent来完成。

picture.image

在CrewAI中， Process与其最新推出的特性Flows是两个概念 。Process是Crew内部的任务分派顺序，只可以指定顺序或者层次；Flows才是可以完全根据需要定制的工作流，可以跨多个Crew，也可与Crew无关。

picture.image

CrewAI实现多智能体Demo

picture.image

现在使用CrewAI来实现我们的简单多智能体Demo：

picture.image

CrewAI是一个比较简单易用的高层框架，你可以非常快捷的创建一个多智能体系统，相比LangGraph的复杂，其更接近于Swam。

1. 创建工具Tool

同Swam与LangGraph一样，创建两个工具，一个用来执行web搜索，一个用来模拟邮件发送：


        
            

          from crewai import Agent, Crew, Process, Task  
from crewai.project import CrewBase, agent, crew, task  
from crewai.tools import tool  
from langchain\_community.tools.tavily\_search import TavilySearchResults  
from crewai import LLM  
  
# email工具  
@tool  
def email\_tool(recipient: str, subject: str, body: str) -> str:  
    """用于向指定收件人发送带有主题和正文的电子邮件。"""  
      
    print(f"Sending email to {recipient} with subject '{subject}'......")  
    return f"Email sent to {recipient} with subject '{subject}' and body '{body}'."  
  
@tool  
def web\_search\_tool(query: str) -> str:  
    """用于使用输入的关键词信息执行网页搜索。"""  
      
    print(f"Searching for: {query}......")  
    search = TavilySearchResults(max\_results=3)  
    results = search.invoke(query)  
    return "\n".join([r["content"] for r in results])

2. 定义三个Agent

非常类似Swam，简单的设定角色、目标及其使用的工具、模型等，即可创建一个Agent。注意这里的管理Agent，由于其完全借助LLM来分派任务，所以无需额外的工具。


        
            

          #web search agent  
web\_searcher = Agent(  
      role="网络搜索者",  
      goal="根据输入的任务，判断搜索关键词，并进行搜索",  
      backstory="你擅长在网上导航，找到最相关和最新的信息。",  
      tools=[web\_search\_tool],   
      verbose=True  
    )  
  
#email agent  
emailer = Agent(  
      role="邮件发送者",  
      goal="根据输入的任务，发送精心编写的电子邮件给指定的联系人。",  
      backstory="你是一个熟练的沟通者，擅长撰写清晰有效的电子邮件，并发送给指定的收件人。",  
      tools=[email\_tool],   
      verbose=True  
    )  
  
#project manager agent  
manager = Agent(  
            role="项目经理",  
            goal="高效管理团队，确保高质量完成任务",  
            backstory="你是一位经验丰富的项目经理，擅长监督复杂项目并指导团队取得成功。你的角色是协调团队成员的努力，确保每项任务按时完成并达到最高标准。",  
            llm='gpt-4o-mini',  
            allow\_delegation=True  
        )

3. 组件Crew并执行任务

有了Agent，现在你需要组建一个团队（Crew），并设定工作任务（Task）以及任务流程的策略(Process）：


          
crew = Crew(      agents=[web_searcher,emailer],       tasks=[Task(                        description="搜索南京的最新天气情况并发送邮件到ycp@test.com",                        expected_output="邮件确认信息"                    )],           
          
      manager_agent=manager,      process=Process.hierarchical,      verbose=True    )crew.kickoff()

由于我们采用Supervisor的层级多智能体系统结构，因此需要指定一个管理Agent（manager_agent）；同时设定任务的策略为hierarchical，这种任务策略下，会由管理Agent来负责进行任务的创建与分派，因此在任务（Task）的定义中无需直接指定某个执行的Agent。

4. 运行任务，观察结果

现在观察任务执行的结果，可以看到任务执行的细节与推理过程：

首先管理Agent判断出需要执行搜索任务，并把任务交给了搜索Agent，这个Agent会使用web_search_tool来执行任务，获得工具的响应，然后输出结果：

picture.image

管理Agent收到搜索Agent的结果后，会再次判断下一个任务及其执行者，这里将任务分派给了邮件Agent，邮件Agent则自主决定使用工具email_tool来完成这个任务：

picture.image

最后，管理Agent收到邮件Agent的执行反馈，进行总结、判断并输出最终的任务结果，结束流程。

picture.image

可以看到，CrewAI中团队的各个Agent遵循ReAct行动范式来自主完成任务执行流程，并最终输出结果。

CrewAI提供了一种直观与简洁的方式来快速地创建一个多智能体团队，完成特定的任务。但很显然，由于这里的流程编排过于薄弱，基本只能依靠Crew内预定义的两种简单流程策略，所以就存在类似早期Langchain（LangGraph推出之前）中Agent相同的问题：

无法支持复杂工作流，比如循环、条件分支等
较大的不可控性，完全依赖于Agent自我决策（本质上是依赖于LLM）

CrewAI新特性：更强大的Flows

picture.image

面对更复杂的LLM应用工作流的需求，CrewAI也在最新的版本推出了重要的新特性： CrewAI Flows ，以用来解决之前版本中对复杂工作流支撑能力的不足。熟悉的朋友可能意识到， 这是一个意义类似于LangChain的LangGraph、LlamaIndex的Workflows的新特性（甚至在使用上也非常相似） 。

picture.image

CrewAI Flows的主要特性有：

简化工作流创建： 轻松将多个团队和任务组装成复杂的AI工作流
灵活的状态管理： 工作流中不同任务之间的状态共享非常简单
事件驱动架构： 事件驱动架构，类似LlamaIndex的Workflows
灵活的控制逻辑： 在工作流中可以实现条件逻辑、循环和分支

用CrewAI Flows来模拟实现上面的Supervisor模式的工作流，又是一种完全不同的方法，这里我们给出一个模拟的实现（借助LLM推理与调用Agent的部分被省略）：


        
            

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

            
#全局状态，非常类似LangGraph中的state
            
class ExampleState(BaseModel):  
    task: str = ""  
    messages: list = []  
    next\_agent: str = ""  
  
class SupervisorFlow(Flow[ExampleState]):  
  
    #管理节点。作为工作流入口，同时，Search与Email两个节点也会返回到这个节点  
    @start(or\_("search\_method", "email\_method"))  
    def supervisor(self):  
        print(f"\n任务：{self.state.task}")  
        print("\n判断下一步需要使用的Agent...")  
  
        #这里随机模拟，实际需要借助LLM来推理下一步的Agent  
        choice = random.choice(['search', 'email', 'finish'])  
        self.state.next\_agent = choice  
  
    #路由节点。根据supervisor节点的返回值，选择下一步的节点  
    @router(supervisor)  
    def route(self):  
        if self.state.next\_agent == 'search':  
            return "search\_agent"  
        elif self.state.next\_agent == 'email':  
            return "email\_agent"  
        else:  
            return "finish"  
  
    #Search节点  
    @listen("search\_agent")  
    def search\_method(self):  
        print("\n调用Search Agent...")  
  
    #Email节点  
    @listen("email\_agent")  
    def email\_method(self):  
        print("\n调用Email Agent...")  
  
    #结束节点。当supervisor节点返回finish时，结束工作流  
    @listen("finish")  
    def finish\_method(self):  
        print("\n任务完成.")  
  
flow = SupervisorFlow()  
flow.kickoff(inputs={'task':'搜索最热门新闻并发送邮件到aa@aa.com'})