CrewAI快速入门

简介

CrewAI 是一个 AI 智能体协作框架，就像组建一个 AI 团队。你可以给每个 AI 分配不同职位（如研究员、分析师），它们会像真实团队一样分工合作完成任务。适合自动化客服、数据分析、市场调研等场景。

组件	描述	关键特性
Crew	最高层级的组织	・管理人工智能代理团队・监督工作流程・确保协作・交付成果
AI Agents	专业团队成员	・拥有特定角色（研究员、撰稿人）・使用指定工具・能够委派任务・自主做出决策
Process	工作流管理系统	・定义协作模式・控制任务分配・管理交互・确保高效执行
Tasks	单个的工作安排	・有明确的目标・使用特定工具・融入更大的流程・产生可执行的结果

各部分如何协同运作

1. Crew是一个团队，包含一个或多个AI Agent，一般是完成一个较大的目标
2. AI Agents是执行任务的最小单位，一般完成一个小目标，可以配备外部工具和上下文记忆
3. 多个Crew通过Process可以整合起来，完整一个最终的目标

Demo

下面是一个CrewAI项目的一般结构，以编写一个查找员工信息的程序为例（crewai版本0.108.0）

一般是先在agents.yaml 和tasks.yaml 里面编写给智能体的Prompt，其所采用的结构是RTGO

在agents.yaml 中编写如下内容：

developer_information_retriever_agent:
  role: >
    人力资源助理
  goal: >
    根据公司开发人员的员工代码检索并提供他们的详细信息。如果员工代码不匹配，则说“员工详细信息不可用。”
  backstory: >
    一位公司人事管理方面的专家，有权访问结构化的开发人员信息。

在tasks.yaml 中编写如下内容

developers_information:
  description: >
    查找并提供员工代码为{employeeCode}的开发人员的信息。
  expected_output: >
    一份包含开发人员姓名、工作职位、所在地区、电话号码和电子邮件地址的详细报告。
  agent: developer_information_retriever_agent

知识库内容

Name: Samantha Brown
Job Role: Lead
Region: AU
Phone Number: 555-1357
Email Address: sbrown@example.com

配置好了Agent和Task，然后就可以开始配置团队了（编写crew.py ）

from pathlib import Path

from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
from crewai.knowledge.source.json_knowledge_source import JSONKnowledgeSource
import os
from dotenv import load_dotenv

load_dotenv()  # 加载环境变量

@CrewBase
class DevelopersCrew:
    """Crew for retrieving developers information."""

    # 获取当前位置路径
    current_dir = Path(__file__).parent
    # 获取当前位置的上一级路径
    parent_dir = current_dir.parent.parent.parent.parent
    # 标准知识库文件
    file_path = parent_dir / "data/Developers.json"

    print(file_path)

    # Create a JSON knowledge source
    developers_json_source = JSONKnowledgeSource(
        file_paths=[file_path]
    )

    # Paths to the YAML configuration files
    agents_config = 'config/agents.yaml'
    tasks_config = 'config/tasks.yaml'

    @agent
    def developer_information_retriever_agent(self) -> Agent:
        return Agent(
            config=self.agents_config['developer_information_retriever_agent'],
            verbose=True,
            knowledge_sources=[self.developers_json_source],
            # embedder={
            #     "provider": "openai",
            #     "config": {
            #         "model": "text-embedding-3-small",
            #         "api_key": os.getenv("OPENAI_API_KEY"),
            #         "api_base": os.getenv("OPENAI_API_BASE"),
            #     }
            # }
            embedder={
                "provider": "ollama",
                "config": {
                    "model": "nomic-embed-text:latest",
                    # "api_key": GEMINI_API_KEY,
                    "api_base": "<http://localhost:11434/v1>",
                }
            }
        )

    @task
    def developers_information(self) -> Task:
        return Task(
            config=self.tasks_config['developers_information'],
        )

    @crew
    def crew(self) -> Crew:
        """Provides Employee details."""
        return Crew(
            agents=self.agents,  # Automatically collected by the @agent decorator
            tasks=self.tasks,  # Automatically collected by the @task decorator
            process=Process.sequential,
            verbose=True,
        )

然后编写main.py ，给出程序的输入内容

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

from src.human_resource_assistance_flow.crews.developers_crew.developers_crew import DevelopersCrew
from src.human_resource_assistance_flow.crews.leads_crew.leads_crew import LeadsCrew

# from crews.developers_crew.developers_crew import DevelopersCrew
# from crews.leads_crew.leads_crew import LeadsCrew
class EmployeeState(BaseModel):
    employeeCode: str = ""
    job_role: str = ""
    employee_details: str = ""

class HumanResourceAssistanceFlow(Flow[EmployeeState]):

    @start()
    def get_employee_details(self):
        # In a real application, these details might come from user input or a database
        self.state.employeeCode = "L209"
        self.state.job_role = "Lead"
        print(f"employeeCode: {self.state.employeeCode}, job_role: {self.state.job_role}")
        return "details_collected"

    @router(get_employee_details)
    def route_based_on_relationship(self):
        """Route based on job role."""
        if self.state.job_role == "Lead":
            return "Lead"
        elif self.state.job_role == "Developer":
            return "Developer"
        else:
            return "NA"

    @listen("Lead")
    def leads_information(self):
        """Retrieve the infomation about the lead"""
        crew = LeadsCrew().crew()
        inputs = {
            "employeeCode": self.state.employeeCode
        }
        result = crew.kickoff(inputs=inputs)
        self.state.employee_details = result.raw
        print(f"Employee Information: {self.state.employee_details}")
        return "details_retrieved"

    @listen("Developer")
    def developers_information(self):
        """Retrieve the infomation about the developer"""
        crew = DevelopersCrew().crew()
        inputs = {
            "employeeCode": self.state.employeeCode,
        }
        result = crew.kickoff(inputs=inputs)
        self.state.employee_details = result.raw
        print(f"Employee Information: {self.state.employee_details}")
        return "details_retrieved"

    

    @listen(or_(leads_information,developers_information))
    def save_details(self):
        with open("employee_details.txt", "w") as file:
            file.write(self.state.employee_details)
        print("Employee details saved to employee_details.txt")

def kickoff():
    employee_flow = HumanResourceAssistanceFlow()
    employee_flow.kickoff()

def plot():
    employee_flow = HumanResourceAssistanceFlow()
    employee_flow.plot()

if __name__ == "__main__":
    kickoff()

最后配置LLM的API文件.env

# ChatGPT
OPENAI_API_BASE="<https://xxxxxxx/v1>"
OPENAI_API_KEY="sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
OPENAI_MODEL_NAME="openai/gpt-4o"

最终结果

补充

可以正常使用知识库的功能，最好在0.108.0以上

知识库的位置：C:\\Users\\用户名\\AppData\\Local\\CrewAI