Crews - CrewAI

什么是Crew？

在crewAI中，一个团队代表了一组协作的代理，他们共同合作以完成一系列任务。每个团队定义了任务执行的策略、代理之间的协作以及整体工作流程。

船员属性

属性	参数	描述
任务	`tasks`	分配给团队的任务列表。
代理	`agents`	属于团队的一部分的代理列表。
流程 (可选)	`process`	团队遵循的流程（例如，顺序的、分层的）。默认是 `sequential`。
详细 (可选)	`verbose`	执行期间日志记录的详细程度。默认为 `False`。
Manager LLM (可选)	`manager_llm`	在分层过程中，管理代理使用的语言模型。在使用分层过程时必需。
函数调用LLM (可选)	`function_calling_llm`	如果传入，团队将使用此LLM为团队中所有代理的工具进行函数调用。每个代理可以有自己的LLM，这将覆盖团队的LLM进行函数调用。
配置 (可选)	`config`	可选的工作组配置设置，格式为`Json`或`Dict[str, Any]`。
最大RPM (可选)	`max_rpm`	执行期间团队遵守的每分钟最大请求数。默认为 `None`。
语言 (可选)	`language`	用于机组人员的语言，默认为英语。
语言文件 (可选)	`language_file`	用于团队的语言文件的路径。
内存 (可选)	`memory`	用于存储执行记忆（短期、长期、实体记忆）。
内存配置 (可选)	`memory_config`	用于配置crew使用的内存提供者。
缓存 (可选)	`cache`	指定是否使用缓存来存储工具执行的结果。默认为 `True`。
嵌入器 (可选)	`embedder`	用于配置crew使用的嵌入器。目前主要用于内存。默认值为`{"provider": "openai"}`。
完整输出 (可选)	`full_output`	团队是否应返回包含所有任务输出的完整输出，还是仅返回最终输出。默认为 `False`。
步骤回调 (可选)	`step_callback`	一个在每个代理的每一步之后调用的函数。这可以用于记录代理的操作或执行其他操作；它不会覆盖特定代理的`step_callback`。
任务回调 (可选)	`task_callback`	一个在每个任务完成后调用的函数。用于监控或任务执行后的额外操作。
共享团队 (可选)	`share_crew`	是否希望与crewAI团队共享完整的团队信息和执行情况，以改进库，并允许我们训练模型。
输出日志文件 (可选)	`output_log_file`	是否希望有一个包含完整输出和执行记录的文件。你可以通过设置为True来启用，它将默认保存在当前文件夹中，并命名为logs.txt，或者传递一个包含文件完整路径和名称的字符串。
管理代理 (可选)	`manager_agent`	`manager` 设置一个自定义代理，该代理将用作管理器。
提示文件 (可选)	`prompt_file`	用于团队的提示JSON文件的路径。
规划 (可选)	`planning`	为团队添加规划能力。当在每次团队迭代之前激活时，所有团队数据将被发送到一个AgentPlanner，该AgentPlanner将规划任务，并将此计划添加到每个任务描述中。
规划LLM (可选)	`planning_llm`	AgentPlanner在规划过程中使用的语言模型。

团队最大RPM: max_rpm 属性设置了团队每分钟可以执行的最大请求数，以避免速率限制，并且如果您设置了它，它将覆盖各个代理的 max_rpm 设置。

创建团队

在CrewAI中有两种创建团队的方式：使用YAML配置（推荐）或直接在代码中定义。

YAML 配置（推荐）

使用YAML配置提供了一种更清晰、更易于维护的方式来定义团队，并且与CrewAI项目中定义代理和任务的方式一致。

在按照安装部分创建您的CrewAI项目后，您可以在继承自CrewBase的类中定义您的团队，并使用装饰器来定义代理、任务和团队本身。

使用装饰器的示例船员类

code
from crewai import Agent, Crew, Task, Process
from crewai.project import CrewBase, agent, task, crew, before_kickoff, after_kickoff


@CrewBase
class YourCrewName:
    """Description of your crew"""

    # Paths to your YAML configuration files
    # To see an example agent and task defined in YAML, checkout the following:
    # - Task: https://docs.crewai.com/concepts/tasks#yaml-configuration-recommended
    # - Agents: https://docs.crewai.com/concepts/agents#yaml-configuration-recommended
    agents_config = 'config/agents.yaml' 
    tasks_config = 'config/tasks.yaml' 

    @before_kickoff
    def prepare_inputs(self, inputs):
        # Modify inputs before the crew starts
        inputs['additional_data'] = "Some extra information"
        return inputs

    @after_kickoff
    def process_output(self, output):
        # Modify output after the crew finishes
        output.raw += "\nProcessed after kickoff."
        return output

    @agent
    def agent_one(self) -> Agent:
        return Agent(
            config=self.agents_config['agent_one'],
            verbose=True
        )

    @agent
    def agent_two(self) -> Agent:
        return Agent(
            config=self.agents_config['agent_two'],
            verbose=True
        )

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

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

    @crew
    def crew(self) -> Crew:
        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,
        )

任务将按照它们定义的顺序执行。

CrewBase 类以及这些装饰器自动收集代理和任务，减少了手动管理的需求。

来自 `annotations.py` 的装饰器概述

CrewAI 在 annotations.py 文件中提供了多个装饰器，用于标记你的 crew 类中的方法以进行特殊处理：

@CrewBase: 将类标记为船员基类。
@agent: 表示一个返回Agent对象的方法。
@task: 表示一个返回Task对象的方法。
@crew: 表示返回Crew对象的方法。
@before_kickoff: （可选）标记一个在团队开始前执行的方法。
@after_kickoff: （可选）标记一个在团队完成后执行的方法。

这些装饰器有助于组织您的团队结构，并自动收集代理和任务，而无需手动列出它们。

直接代码定义（替代方案）

或者，您可以直接在代码中定义crew，而不使用YAML配置文件。

code
from crewai import Agent, Crew, Task, Process
from crewai_tools import YourCustomTool

class YourCrewName:
    def agent_one(self) -> Agent:
        return Agent(
            role="Data Analyst",
            goal="Analyze data trends in the market",
            backstory="An experienced data analyst with a background in economics",
            verbose=True,
            tools=[YourCustomTool()]
        )

    def agent_two(self) -> Agent:
        return Agent(
            role="Market Researcher",
            goal="Gather information on market dynamics",
            backstory="A diligent researcher with a keen eye for detail",
            verbose=True
        )

    def task_one(self) -> Task:
        return Task(
            description="Collect recent market data and identify trends.",
            expected_output="A report summarizing key trends in the market.",
            agent=self.agent_one()
        )

    def task_two(self) -> Task:
        return Task(
            description="Research factors affecting market dynamics.",
            expected_output="An analysis of factors influencing the market.",
            agent=self.agent_two()
        )

    def crew(self) -> Crew:
        return Crew(
            agents=[self.agent_one(), self.agent_two()],
            tasks=[self.task_one(), self.task_two()],
            process=Process.sequential,
            verbose=True
        )

在这个例子中：

代理和任务直接在类中定义，无需使用装饰器。
我们手动创建和管理代理和任务的列表。
这种方法提供了更多的控制，但对于较大的项目可能较难维护。

团队输出

在CrewAI框架中，一个团队的输出被封装在CrewOutput类中。该类提供了一种结构化的方式来访问团队执行的结果，包括各种格式，如原始字符串、JSON和Pydantic模型。 CrewOutput包括最终任务输出的结果、令牌使用情况以及各个任务的输出。

船员输出属性

属性	参数	类型	描述
原始	`raw`	`str`	团队的原始输出。这是输出的默认格式。
Pydantic	`pydantic`	`Optional[BaseModel]`	一个Pydantic模型对象，表示crew的结构化输出。
JSON 字典	`json_dict`	`Optional[Dict[str, Any]]`	表示团队输出的 JSON 的字典。
任务输出	`tasks_output`	`List[TaskOutput]`	一个`TaskOutput`对象的列表，每个对象代表团队中一个任务的输出。
令牌使用情况	`token_usage`	`Dict[str, Any]`	令牌使用情况的摘要，提供语言模型在执行期间性能的洞察。

船员输出方法和属性

方法/属性	描述
json	如果输出格式为JSON，则返回crew输出的JSON字符串表示。
to_dict	将JSON和Pydantic输出转换为字典。
**str**	返回机组输出的字符串表示，优先使用Pydantic，然后是JSON，最后是原始数据。

访问船员输出

一旦一个crew被执行，它的输出可以通过Crew对象的output属性访问。CrewOutput类提供了多种方式来交互和展示这个输出。

示例

Code
# Example crew execution
crew = Crew(
    agents=[research_agent, writer_agent],
    tasks=[research_task, write_article_task],
    verbose=True
)

crew_output = crew.kickoff()

# Accessing the crew output
print(f"Raw Output: {crew_output.raw}")
if crew_output.json_dict:
    print(f"JSON Output: {json.dumps(crew_output.json_dict, indent=2)}")
if crew_output.pydantic:
    print(f"Pydantic Output: {crew_output.pydantic}")
print(f"Tasks Output: {crew_output.tasks_output}")
print(f"Token Usage: {crew_output.token_usage}")

内存利用率

团队可以利用记忆（短期记忆、长期记忆和实体记忆）来随着时间的推移增强他们的执行和学习能力。此功能允许团队存储和回忆执行记忆，有助于决策和任务执行策略。

缓存利用率

缓存可以用来存储工具执行的结果，通过减少重新执行相同任务的需求，使过程更加高效。

机组使用指标

在团队执行后，您可以访问usage_metrics属性，查看由团队执行的所有任务的语言模型（LLM）使用指标。这提供了对操作效率和改进领域的洞察。

Code
# Access the crew's usage metrics
crew = Crew(agents=[agent1, agent2], tasks=[task1, task2])
crew.kickoff()
print(crew.usage_metrics)

团队执行流程

顺序过程：任务一个接一个地执行，允许工作的线性流动。
分层流程：一个管理者代理协调团队，分配任务并在继续之前验证结果。注意：此流程需要manager_llm或manager_agent，并且对于验证流程流至关重要。

启动一个团队

一旦你的团队集结完毕，使用kickoff()方法启动工作流程。这将根据定义的过程流程开始执行过程。

Code
# Start the crew's task execution
result = my_crew.kickoff()
print(result)

启动团队的不同方式

一旦您的团队组建完成，使用适当的启动方法启动工作流程。CrewAI 提供了几种方法以便更好地控制启动过程：kickoff()、kickoff_for_each()、kickoff_async() 和 kickoff_for_each_async()。

kickoff(): 根据定义的流程开始执行过程。
kickoff_for_each(): 为每个代理单独执行任务。
kickoff_async(): 异步启动工作流。
kickoff_for_each_async(): 以异步方式为每个代理单独执行任务。

Code
# Start the crew's task execution
result = my_crew.kickoff()
print(result)

# Example of using kickoff_for_each
inputs_array = [{'topic': 'AI in healthcare'}, {'topic': 'AI in finance'}]
results = my_crew.kickoff_for_each(inputs=inputs_array)
for result in results:
    print(result)

# Example of using kickoff_async
inputs = {'topic': 'AI in healthcare'}
async_result = my_crew.kickoff_async(inputs=inputs)
print(async_result)

# Example of using kickoff_for_each_async
inputs_array = [{'topic': 'AI in healthcare'}, {'topic': 'AI in finance'}]
async_results = my_crew.kickoff_for_each_async(inputs=inputs_array)
for async_result in async_results:
    print(async_result)

这些方法为您在团队中管理和执行任务提供了灵活性，允许根据您的需求定制同步和异步的工作流程。

从特定任务重放

您现在可以使用我们的CLI命令replay从特定任务重放。

CrewAI中的重放功能允许您使用命令行界面（CLI）从特定任务重放。通过运行命令crewai replay -t ，您可以指定重放过程的task_id。

Kickoffs 现在会保存最新返回的任务输出，以便您能够从中重放。

使用CLI从特定任务重放

要使用回放功能，请按照以下步骤操作：

打开您的终端或命令提示符。
导航到您的CrewAI项目所在的目录。
运行以下命令：

要查看最新的启动任务ID，请使用：

crewai log-tasks-outputs

然后，要从特定任务重放，请使用：

crewai replay -t <task_id>

这些命令允许您从最新的启动任务重新开始，同时保留之前执行任务的上下文。

开始使用

核心概念

操作指南

工具

遥测

船员

什么是Crew？

船员属性

创建团队

YAML 配置（推荐）

使用装饰器的示例船员类

来自 `annotations.py` 的装饰器概述

直接代码定义（替代方案）

团队输出

船员输出属性

船员输出方法和属性

访问船员输出

示例

内存利用率

缓存利用率

机组使用指标

团队执行流程

启动一个团队

启动团队的不同方式

从特定任务重放

使用CLI从特定任务重放

开始使用

核心概念

操作指南

工具

遥测

​什么是Crew？

​船员属性

​创建团队

​YAML 配置（推荐）

​使用装饰器的示例船员类

​来自 annotations.py 的装饰器概述

​直接代码定义（替代方案）

​团队输出

​船员输出属性

​船员输出方法和属性

​访问船员输出

​示例

​内存利用率

​缓存利用率

​机组使用指标

​团队执行流程

​启动一个团队

​启动团队的不同方式

​从特定任务重放

​使用CLI从特定任务重放

什么是Crew？

船员属性

创建团队

YAML 配置（推荐）

使用装饰器的示例船员类

来自 `annotations.py` 的装饰器概述

直接代码定义（替代方案）

团队输出

船员输出属性

船员输出方法和属性

访问船员输出

示例

内存利用率

缓存利用率

机组使用指标

团队执行流程

启动一个团队

启动团队的不同方式

从特定任务重放

使用CLI从特定任务重放