Crews

在 crewAI 框架中理解和使用 crew，包括其完整属性与功能。

概览

在 crewAI 中，crew 表示一组协作工作的 agents，它们共同完成一组任务。每个 crew 都定义了任务执行策略、agent 协作方式以及整体工作流。

Crew 属性

创建 Crews

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

YAML 配置（推荐）

使用 YAML 配置可以让 crew 的定义更清晰、更易维护，并且与 CrewAI 项目中 agent 和 task 的定义方式保持一致。

在按照 Installation 章节创建 CrewAI 项目后，你可以在一个继承自 CrewBase 的类中定义 crew，并通过装饰器定义 agents、tasks 和 crew 本身。

使用装饰器的 Crew 类示例

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:
    """你的 crew 描述"""
    agents: List[BaseAgent]
    tasks: List[Task]
    # YAML 配置文件路径
    # 如需查看 YAML 中定义的 agent 和 task 示例，请参考：
    # - 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):
        # 在 crew 启动前修改输入
        inputs['additional_data'] = "Some extra information"
        return inputs
    @after_kickoff
    def process_output(self, output):
        # 在 crew 完成后修改输出
        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,  # 由 @agent 装饰器自动收集
            tasks=self.tasks,    # 由 @task 装饰器自动收集
            process=Process.sequential,
            verbose=True,
        )

运行上述代码的方法：

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

CrewBase 类以及这些装饰器会自动收集 agents 和 tasks，从而减少手动管理的工作量。

来自 annotations.py 的装饰器概览

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

• @CrewBase：将类标记为 crew 基类。
• @agent：标记一个返回 Agent 对象的方法。
• @task：标记一个返回 Task 对象的方法。
• @crew：标记一个返回 Crew 对象的方法。
• @before_kickoff： （可选）标记一个在 crew 启动前执行的方法。
• @after_kickoff： （可选）标记一个在 crew 完成后执行的方法。

这些装饰器有助于组织 crew 的结构，并自动收集 agents 和 tasks，而无需手动逐一列出。

直接代码定义（可选方式）

另一种方式是不使用 YAML 配置文件，而是直接在代码中定义 crew。

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={})

在这个例子中：

• agents 和 tasks 直接在类中定义，不使用装饰器。
• 我们手动创建并管理 agent 和 task 的列表。
• 这种方式提供了更高的控制力，但在大型项目中可维护性通常较差。

Crew 输出

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

Crew 输出属性

Crew 输出的方法与属性

访问 Crew 输出

当 crew 执行完成后，可以通过 Crew 对象的 output 属性访问其输出。CrewOutput 类提供了多种与输出交互和展示的方式。

示例

# crew 执行示例
crew = Crew(
    agents=[research_agent, writer_agent],
    tasks=[research_task, write_article_task],
    verbose=True
)
crew_output = crew.kickoff()
# 访问 crew 输出
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}")

访问 Crew 日志

你可以通过将 output_log_file 设置为 True(Boolean) 或 file_name(str) 来查看 crew 执行的实时日志。支持将事件记录为 file_name.txt 和 file_name.json 两种格式。 当设置为 True(Boolean) 时，会保存为 logs.txt。

如果 output_log_file 设置为 False(Boolean) 或 None，则不会生成日志。

# 保存 crew 日志
crew = Crew(output_log_file = True)  # 日志将保存为 logs.txt
crew = Crew(output_log_file = file_name)  # 日志将保存为 file_name.txt
crew = Crew(output_log_file = file_name.txt)  # 日志将保存为 file_name.txt
crew = Crew(output_log_file = file_name.json)  # 日志将保存为 file_name.json

Memory 的使用

Crews 可以使用 memory（短期记忆、长期记忆和实体记忆）来增强执行效果，并随着时间推移不断学习。这个特性允许 crews 存储和回忆执行记忆，从而帮助决策和任务执行策略的制定。

Cache 的使用

可以使用缓存来存储工具执行结果，从而减少重复执行相同任务的需要，提高整体效率。

Crew 使用指标

在 crew 执行完成后，你可以访问 usage_metrics 属性，以查看该 crew 中所有任务的语言模型（LLM）使用指标。这有助于分析运行效率并发现优化空间。

# 访问 crew 的 usage metrics
crew = Crew(agents=[agent1, agent2], tasks=[task1, task2])
crew.kickoff()
print(crew.usage_metrics)

Crew 执行流程

• Sequential Process：任务按顺序依次执行，形成线性工作流。
• Hierarchical Process：由一个 manager agent 协调 crew，在继续之前负责委派任务并验证结果。注意：这种流程需要提供 manager_llm 或 manager_agent，并且它对于验证流程执行至关重要。

启动 Crew

当你的 crew 组装完成后，可通过 kickoff() 方法启动工作流。这会根据所定义的流程方式开始执行。

# 启动 crew 的任务执行
result = my_crew.kickoff()
print(result)

启动 Crew 的不同方式

当你的 crew 组装完成后，可以使用合适的 kickoff 方法来启动。CrewAI 提供了多种方法，以更灵活地控制 kickoff 过程。

同步方法

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

异步方法

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

# 启动 crew 的任务执行
result = my_crew.kickoff()
print(result)
# 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)
# 使用 akickoff 的原生异步示例
inputs = {'topic': 'AI in healthcare'}
async_result = await my_crew.akickoff(inputs=inputs)
print(async_result)
# 使用 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)
# 使用 kickoff_async 的线程异步示例
inputs = {'topic': 'AI in healthcare'}
async_result = await my_crew.kickoff_async(inputs=inputs)
print(async_result)
# 使用 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)

这些方法让你可以更灵活地管理和执行 crew 中的任务，从而根据需求选择同步或异步工作流。更多异步示例，请参见 Kickoff Crew Asynchronously 指南。

流式 Crew 执行

如果你希望实时查看 crew 的执行过程，可以启用流式输出，在内容生成时即时接收：

# 启用流式输出
crew = Crew(
    agents=[researcher],
    tasks=[task],
    stream=True
)
# 迭代流式输出
streaming = crew.kickoff(inputs={"topic": "AI"})
for chunk in streaming:
    print(chunk.content, end="", flush=True)
# 获取最终结果
result = streaming.result

更多内容请参见 Streaming Crew Execution 指南。

从特定任务重新回放

你现在可以使用 CLI 命令 replay 从某个特定任务开始重新回放。

CrewAI 的 replay 功能允许你通过命令行接口（CLI）从指定任务开始重新执行。运行命令 crewai replay -t <task_id>，即可指定用于回放的 task_id。

现在，kickoff 会将最近一次 kickoff 返回的任务输出保存在本地，以便你能够从该任务继续回放。

使用 CLI 从特定任务重新回放

使用 replay 功能的步骤如下：

1. 打开终端或命令提示符。
2. 进入你的 CrewAI 项目所在目录。
3. 运行以下命令：

查看最近一次 kickoff 的任务 ID：

crewai log-tasks-outputs

然后，从指定任务开始回放：

crewai replay -t <task_id>

这些命令允许你从最近一次 kickoff 的任务开始回放，同时保留此前已执行任务的上下文。