Crews (团队) - CrewAI 框架

概述

在 crewAI 中，团队（Crew）代表了一组协作智能体（Agents），它们共同努力完成一系列任务。每个团队都定义了任务执行、智能体协作和整体工作流的策略。

团队属性

属性	参数	描述
任务	`tasks`	分配给团队的任务列表。
智能体	`agents`	团队成员的智能体列表。
过程（Process）（可选）	`process`	团队遵循的过程流（例如：顺序执行、层级执行）。默认值为 `sequential`（顺序执行）。
详细模式（Verbose）（可选）	`verbose`	执行期间的日志详细程度。默认值为 `False`。
管理 LLM （可选）	`manager_llm`	在层级过程中由管理智能体使用的语言模型。使用层级过程时为必填项。
函数调用 LLM （可选）	`function_calling_llm`	如果传入，团队将使用此 LLM 为团队中所有智能体的工具执行函数调用。每个智能体可以有自己的 LLM，这会覆盖团队的函数调用 LLM。
配置（Config）（可选）	`config`	团队的可选配置设置，格式为 `Json` 或 `Dict[str, Any]`。
最大 RPM （可选）	`max_rpm`	执行期间团队遵守的最大每分钟请求数。默认值为 `None`。
记忆（Memory）（可选）	`memory`	用于存储执行记忆（短期记忆、长期记忆、实体记忆）。
缓存（Cache）（可选）	`cache`	指定是否使用缓存来存储工具执行的结果。默认值为 `True`。
嵌入器（Embedder）（可选）	`embedder`	团队所用嵌入器的配置。目前主要由记忆功能使用。默认值为 `{"provider": "openai"}`。
步骤回调（Step Callback）（可选）	`step_callback`	在每个智能体的每个步骤之后调用的函数。这可用于记录智能体的操作或执行其他任务；它不会覆盖智能体特有的 `step_callback`。
任务回调（Task Callback）（可选）	`task_callback`	在每个任务完成之后调用的函数。适用于任务执行后的监控或其他操作。
分享团队（Share Crew）（可选）	`share_crew`	是否希望与 crewAI 团队分享完整的团队信息和执行情况，以改进库并协助我们训练模型。
输出日志文件（可选）	`output_log_file`	设置为 True 可将日志保存为当前目录下的 logs.txt，或指定文件路径。如果文件名以 .json 结尾，日志将为 JSON 格式，否则为 .txt。默认值为 `None`。
管理智能体（可选）	`manager_agent`	`manager` 设置一个自定义智能体作为管理者。
提示词文件（可选）	`prompt_file`	团队所使用的提示词 JSON 文件的路径。
规划（Planning）（可选）	`planning`	为团队添加规划能力。激活后，在每次团队迭代前，所有团队数据都会发送给 AgentPlanner 进行任务规划，并将该计划添加到每个任务描述中。
规划 LLM （可选）	`planning_llm`	在规划过程中由 AgentPlanner 使用的语言模型。
知识库源（可选）	`knowledge_sources`	团队层级的知识库源，所有智能体均可访问。
流式传输（Stream）（可选）	`stream`	启用流式输出以在团队执行期间接收实时更新。返回一个可迭代获取数据块的 `CrewStreamingOutput` 对象。默认值为 `False`。
聊天 LLM （可选）	`chat_llm`	用于编排 `crewai chat` CLI 与团队交互的语言模型。接受模型名称字符串或 `LLM` 实例。默认值为 `None`。
启动前回调（可选）	`before_kickoff_callbacks`	在团队启动之前执行的可调用函数列表。每个回调都会接收并可以修改输入字典。不同于 `@before_kickoff` 装饰器。默认值为 `[]`。
启动后回调（可选）	`after_kickoff_callbacks`	在团队执行完毕之后执行的可调用函数列表。每个回调都会接收并可以修改 `CrewOutput`。不同于 `@after_kickoff` 装饰器。默认值为 `[]`。
追踪（Tracing）（可选）	`tracing`	控制团队的 OpenTelemetry 追踪。`True` = 始终启用，`False` = 始终禁用，`None` = 从环境/用户设置中继承。默认值为 `None`。
技能（Skills）（可选）	`skills`	应用于团队中所有智能体的 `Path` 对象（技能搜索目录）或预加载的 `Skill` 对象列表。默认值为 `None`。
安全配置（可选）	`security_config`	管理团队指纹识别和身份的安全配置实例。默认值为 `SecurityConfig()`。
检查点（Checkpoint）（可选）	`checkpoint`	启用自动检查点。传入 `True` 使用合理的默认值，传入 `CheckpointConfig` 以完全控制，传入 `False` 禁用，或传入 `None` 继承。请参阅下方的检查点部分。默认值为 `None`。

团队最大 RPM：max_rpm 属性设置团队每分钟允许执行的最大请求数，以避免速率限制；如果设置了该值，它将覆盖各个智能体的 max_rpm 设置。

创建团队

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

YAML 配置（推荐）

使用 YAML 配置提供了一种更清晰、更易于维护的定义团队方式，并与 CrewAI 项目中定义智能体和任务的方式保持一致。在按照安装部分 outlined 的步骤创建 CrewAI 项目后，您可以在继承自 CrewBase 的类中定义团队，并使用装饰器来定义智能体、任务和团队本身。

使用装饰器的团队类示例

代码

from crewai import Agent, Crew, Task, Process
from crewai.project import CrewBase, agent, task, crew, before_kickoff, after_kickoff
from crewai.agents.agent_builder.base_agent import BaseAgent
from typing import List

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

    agents: List[BaseAgent]
    tasks: List[Task]

    # Paths to your YAML configuration files
    # To see an example agent and task defined in YAML, checkout the following:
    # - Task: https://docs.crewai.com.cn/concepts/tasks#yaml-configuration-recommended
    # - Agents: https://docs.crewai.com.cn/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'], # type: ignore[index]
            verbose=True
        )

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

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

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

    @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,
        )

如何运行上述代码

代码

YourCrewName().crew().kickoff(inputs={"any": "input here"})

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

CrewBase 类连同这些装饰器，自动化了智能体和任务的收集，减少了手动管理的需求。

`annotations.py` 中的装饰器概述

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

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

这些装饰器有助于组织团队结构，并自动收集智能体和任务，无需手动列出。

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

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

代码

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
        )

如何运行上述代码

代码

YourCrewName().crew().kickoff(inputs={})

在此示例中：

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

团队输出

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

团队输出属性

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

团队输出的方法和属性

方法/属性	描述
json	如果输出格式为 JSON，则返回团队输出的 JSON 字符串表示形式。
to_dict	将 JSON 和 Pydantic 输出转换为字典。
**str**	返回团队输出的字符串表示形式，优先级为：Pydantic > JSON > 原始数据。

访问团队输出

一旦团队执行完毕，可以通过 Crew 对象的 output 属性访问其输出。CrewOutput 类提供了与该输出进行交互和呈现的多种方式。

示例

代码

# 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}")

访问团队日志

您可以通过将 output_log_file 设置为 True(布尔值) 或 file_name(字符串) 来查看团队执行的实时日志。支持以 file_name.txt 和 file_name.json 两种格式记录事件。如果设置为 True(布尔值)，将保存为 logs.txt。如果 output_log_file 设置为 False(布尔值) 或 None，则不会生成日志。

代码

# Save crew logs
crew = Crew(output_log_file = True)  # Logs will be saved as logs.txt
crew = Crew(output_log_file = file_name)  # Logs will be saved as file_name.txt
crew = Crew(output_log_file = file_name.txt)  # Logs will be saved as file_name.txt
crew = Crew(output_log_file = file_name.json)  # Logs will be saved as file_name.json

检查点 (Checkpointing)

检查点（Checkpointing）功能允许团队在关键事件（如任务完成）后自动保存状态，以便长时间运行或中断的作业能够从中断处准确恢复，而无需重新执行已完成的任务。

快速入门

传入 checkpoint=True 以启用具有合理默认值的检查点（在每个任务完成后保存到 .checkpoints/）。

代码

from crewai import Crew, Process

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,
    checkpoint=True,  # saves to .checkpoints/ after every task
)

crew.kickoff(inputs={"topic": "AI trends"})

使用 `CheckpointConfig` 实现完全控制

使用 CheckpointConfig 可对位置、触发事件、存储后端和保留策略进行精细控制。

代码

from crewai import Crew, Process
from crewai.state.checkpoint_config import CheckpointConfig

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,
    checkpoint=CheckpointConfig(
        location="./.checkpoints",       # directory for JSON files (default)
        on_events=["task_completed"],    # trigger after each task (default)
        max_checkpoints=5,               # keep only the 5 most recent checkpoints
    ),
)

crew.kickoff(inputs={"topic": "AI trends"})

从检查点恢复

使用 Crew.from_checkpoint() 从保存的检查点文件还原团队，然后调用 kickoff() 继续执行。

代码

# Resume from the most recent checkpoint
crew = Crew.from_checkpoint(".checkpoints/latest.json")
crew.kickoff()

从检查点还原时，checkpoint_inputs、checkpoint_train 和 checkpoint_kickoff_event_id 会自动重建，您无需手动设置。

`CheckpointConfig` 属性

属性	类型	默认值	描述
`location`	`str`	`"./.checkpoints"`	存储目标。对于 `JsonProvider`，这是一个目录路径；对于 `SqliteProvider`，这是一个数据库文件路径。
`on_events`	`list[str]`	`["task_completed"]`	触发检查点写入的事件类型。使用 `["*"]` 可在每个事件上创建检查点。
`provider`	`JsonProvider \| SqliteProvider`	`JsonProvider()`	存储后端。默认值为 `JsonProvider`（纯 JSON 文件）。
`max_checkpoints`	`int \| None`	`无`	保留的最大检查点数量。每次写入后最旧的检查点将被清除。`None` 表示保留所有检查点。

记忆利用

团队可以利用记忆（短期、长期和实体记忆）来增强其执行能力和随时间的学习效果。此功能允许团队存储和调取执行记忆，从而辅助决策和任务执行策略。

缓存利用

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

团队使用指标

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

代码

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

团队执行过程

顺序执行（Sequential Process）：任务一个接一个地执行，实现线性工作流。
层级执行（Hierarchical Process）：管理智能体协调团队，在继续之前委派任务并验证结果。注意：此过程需要 manager_llm 或 manager_agent，这对于验证流程至关重要。

启动团队（Kickoff）

组建团队后，使用 kickoff() 方法启动工作流。这将根据定义的流程启动执行过程。

代码

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

启动团队的不同方式

组建团队后，使用适当的启动方法开始工作流。CrewAI 提供了几种方法以便更好地控制启动过程。

同步方法

kickoff()：根据定义的流程启动执行过程。
kickoff_for_each()：为每个提供的输入事件或集合中的项目按顺序执行任务。

异步方法

CrewAI 提供两种异步执行方法：

方法	类型	描述
`akickoff()`	原生异步	整个执行链中的真正 async/await
`akickoff_for_each()`	原生异步	列表中每个输入的原生异步执行
`kickoff_async()`	基于线程	将同步执行封装在 `asyncio.to_thread` 中
`kickoff_for_each_async()`	基于线程	列表中每个输入的基于线程的异步执行

对于高并发工作负载，建议使用 akickoff() 和 akickoff_for_each()，因为它们在任务执行、记忆操作和知识检索中使用原生异步。

代码

# 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 native async with akickoff
inputs = {'topic': 'AI in healthcare'}
async_result = await my_crew.akickoff(inputs=inputs)
print(async_result)

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

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

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

这些方法为管理和执行任务提供了灵活性，允许根据需要定制同步和异步工作流。有关详细的异步示例，请参阅异步启动团队指南。

流式团队执行

为了实时了解团队执行情况，您可以启用流式传输，在输出生成时接收它们。

代码

# Enable streaming
crew = Crew(
    agents=[researcher],
    tasks=[task],
    stream=True
)

# Iterate over streaming output
streaming = crew.kickoff(inputs={"topic": "AI"})
for chunk in streaming:
    print(chunk.content, end="", flush=True)

# Access final result
result = streaming.result

在流式团队执行指南中了解更多信息。

从特定任务重播

您现在可以使用我们的 CLI 命令 replay 从特定任务重播。 CrewAI 中的重播功能允许您使用命令行界面 (CLI) 从特定任务进行重播。通过运行命令 crewai replay -t <task_id>，您可以指定重播过程的 task_id。启动命令现在会将最新返回的任务输出在本地保存，以便您能够从中重播。

使用 CLI 从特定任务重播

要使用重播功能，请执行以下步骤：

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

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

crewai log-tasks-outputs

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

crewai replay -t <task_id>

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

文档索引

​概述

​团队属性

​创建团队

​YAML 配置（推荐）

​使用装饰器的团队类示例

​annotations.py 中的装饰器概述

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

​团队输出

​团队输出属性

​团队输出的方法和属性

​访问团队输出

​示例

​访问团队日志

​检查点 (Checkpointing)

​快速入门

​使用 CheckpointConfig 实现完全控制

​从检查点恢复

​CheckpointConfig 属性

​记忆利用

​缓存利用

​团队使用指标

​团队执行过程

​启动团队（Kickoff）

​启动团队的不同方式

​同步方法

​异步方法

​流式团队执行

​从特定任务重播

​使用 CLI 从特定任务重播

概述

团队属性

创建团队

YAML 配置（推荐）

使用装饰器的团队类示例

`annotations.py` 中的装饰器概述

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

团队输出

团队输出属性

团队输出的方法和属性

访问团队输出

示例

访问团队日志

检查点 (Checkpointing)

快速入门

使用 `CheckpointConfig` 实现完全控制

从检查点恢复

`CheckpointConfig` 属性

记忆利用

缓存利用

团队使用指标

团队执行过程

启动团队（Kickoff）

启动团队的不同方式

同步方法

异步方法

流式团队执行

从特定任务重播

使用 CLI 从特定任务重播