在 crewAI 中，一个 crew 代表一个协作的智能体小组，共同努力完成一系列任务。每个 crew 定义了任务执行的策略、智能体协作方式以及整体工作流程。
Crew 属性
| 属性 | 参数 | 描述 | |
|---|
| 任务 | tasks | 分配给 crew 的任务列表。 | |
| 智能体 | agents | 属于该 crew 的智能体列表。 | |
| Process （可选） | process | crew 遵循的流程（例如，顺序式、层级式）。默认为 sequential。 | |
| Verbose （可选） | verbose | 执行期间日志记录的详细程度。默认为 False。 | |
| Manager LLM （可选） | manager_llm | 在层级式流程中由管理者智能体使用的大语言模型。在使用层级式流程时必需。 | |
| Function Calling LLM （可选） | function_calling_llm | 如果传入，crew 将使用此 LLM 为 crew 中的所有智能体执行工具的函数调用。每个智能体可以有自己的 LLM，这将覆盖 crew 的 LLM 用于函数调用。 | |
| Config （可选） | config | crew 的可选配置设置，格式为 Json 或 Dict[str, Any]。 | |
| Max RPM （可选） | max_rpm | crew 在执行期间遵守的每分钟最大请求数。默认为 None。 | |
| Memory （可选） | memory | 用于存储执行记忆（短期、长期、实体记忆）。 | |
| Cache （可选） | cache | 指定是否使用缓存来存储工具执行的结果。默认为 True。 | |
| Embedder （可选） | embedder | crew 使用的嵌入器配置。目前主要由记忆功能使用。默认为 {"provider": "openai"}。 | |
| Step Callback （可选） | step_callback | 在每个智能体的每一步之后调用的函数。可用于记录智能体的行为或执行其他操作；它不会覆盖智能体特定的 step_callback。 | |
| Task Callback （可选） | task_callback | 在每个任务完成后调用的函数。可用于监控或在任务执行后执行额外的操作。 | |
| Share Crew （可选） | share_crew | 是否愿意与 crewAI 团队分享完整的 crew 信息和执行情况，以帮助我们改进库并训练模型。 | |
| Output Log File （可选） | output_log_file | 设置为 True 可将日志保存为当前目录下的 logs.txt，或提供文件路径。如果文件名以 .json 结尾，日志将为 JSON 格式，否则为 .txt 格式。默认为 None。 | |
| Manager Agent （可选） | manager_agent | manager 设置一个自定义智能体作为管理者。 | |
| Prompt File （可选） | prompt_file | 用于 crew 的提示 JSON 文件的路径。 | |
| Planning （可选） | planning | 为 Crew 增加规划能力。激活后，在每次 Crew 迭代之前，所有 Crew 数据将被发送到一个 AgentPlanner，它将规划任务，并且这个计划将被添加到每个任务的描述中。 | |
| Planning LLM （可选） | planning_llm | 在规划流程中由 AgentPlanner 使用的大语言模型。 | |
| Knowledge Sources （可选） | knowledge_sources | 在 crew 级别可用的知识源，所有智能体均可访问。 | |
Crew Max RPM：max_rpm 属性设置了 crew 每分钟可以执行的最大请求数，以避免速率限制，并且如果您设置了它，它将覆盖单个智能体的 max_rpm 设置。
创建 Crews
在 CrewAI 中创建 crews 有两种方式：使用 YAML 配置（推荐） 或 直接在代码中 定义。
YAML 配置（推荐）
使用 YAML 配置提供了一种更清晰、更易于维护的方式来定义 crews，并且与在 CrewAI 项目中定义智能体和任务的方式保持一致。 按照 安装 部分的说明创建您的 CrewAI 项目后，您可以在一个继承自 CrewBase 的类中定义您的 crew，并使用装饰器来定义智能体、任务和 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:
"""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.org.cn/concepts/tasks#yaml-configuration-recommended
# - Agents: https://docs.crewai.org.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 文件中提供了几个装饰器，用于标记您的 crew 类中的方法以进行特殊处理。
@CrewBase：将类标记为 crew 基类。@agent：表示一个返回 Agent 对象的方法。@task：表示一个返回 Task 对象的方法。@crew：表示返回 Crew 对象的方法。@before_kickoff：（可选）标记一个在 crew 开始前执行的方法。@after_kickoff：（可选）标记一个在 crew 完成后执行的方法。
这些装饰器有助于组织您的 crew 结构，并自动收集智能体和任务，而无需手动列出它们。
直接代码定义（替代方案）
或者，您也可以不使用 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={})
- 智能体和任务直接在类中定义，没有使用装饰器。
- 我们手动创建并管理智能体和任务的列表。
- 这种方法提供了更多的控制权，但对于较大的项目来说可能维护性较差。
Crew 输出
在 CrewAI 框架中，crew 的输出被封装在 CrewOutput 类中。该类提供了一种结构化的方式来访问 crew 执行的结果，包括多种格式，如原始字符串、JSON 和 Pydantic 模型。CrewOutput 包括最终任务输出的结果、token 使用情况以及各个任务的输出。
Crew 输出属性
| 属性 | 参数 | 类型 | 描述 |
|---|
| Raw | raw | str | crew 的原始输出。这是输出的默认格式。 |
| Pydantic | pydantic | Optional[BaseModel] | 一个表示 crew 结构化输出的 Pydantic 模型对象。 |
| JSON Dict | json_dict | Optional[Dict[str, Any]] | 一个表示 crew JSON 输出的字典。 |
| Tasks Output | tasks_output | List[TaskOutput] | 一个 TaskOutput 对象列表，每个对象代表 crew 中一个任务的输出。 |
| Token Usage | token_usage | Dict[str, Any] | token 使用情况的摘要，提供对执行期间语言模型性能的洞察。 |
Crew 输出方法和属性
| 方法/属性 | 描述 |
|---|
| json | 如果输出格式是 JSON，则返回 crew 输出的 JSON 字符串表示形式。 |
| to_dict | 将 JSON 和 Pydantic 输出转换为字典。 |
| **str** | 返回 crew 输出的字符串表示形式，优先顺序为 Pydantic、JSON，然后是 raw。 |
访问 Crew 输出
一旦 crew 执行完毕，可以通过 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}")
访问 Crew 日志
您可以通过将 output_log_file 设置为 True(布尔值) 或 file_name(字符串) 来查看 crew 执行的实时日志。支持将事件日志记录为 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
内存利用
Crews 可以利用内存（短期、长期和实体内存）来增强其执行和随时间的学习能力。此功能允许 crews 存储和回忆执行记忆，从而辅助决策和任务执行策略。
缓存利用
可以利用缓存来存储工具执行的结果，通过减少重复执行相同任务的需要来提高流程效率。
Crew 使用指标
在 crew 执行后，您可以访问 usage_metrics 属性来查看由 crew 执行的所有任务的语言模型（LLM）使用指标。这为运营效率和改进领域提供了洞察。
# Access the crew's usage metrics
crew = Crew(agents=[agent1, agent2], tasks=[task1, task2])
crew.kickoff()
print(crew.usage_metrics)
Crew 执行流程
- 顺序流程：任务一个接一个地执行，实现线性工作流。
- 层级流程：一个管理者智能体协调 crew，委派任务并在继续之前验证结果。注意：此流程需要一个
manager_llm 或 manager_agent，并且它对于验证流程至关重要。
启动一个 Crew
一旦您的 crew 组建完成，使用 kickoff() 方法启动工作流。这将根据定义的流程开始执行过程。
# Start the crew's task execution
result = my_crew.kickoff()
print(result)
启动 Crew 的不同方式
一旦您的 crew 组建完成，使用适当的 kickoff 方法启动工作流。CrewAI 提供了几种方法来更好地控制启动过程：kickoff()、kickoff_for_each()、kickoff_async() 和 kickoff_for_each_async()。
kickoff()：根据定义的流程开始执行过程。kickoff_for_each()：为提供的每个输入事件或集合中的每个项目顺序执行任务。kickoff_async()：异步启动工作流。kickoff_for_each_async()：利用异步处理，为提供的每个输入事件或项目并发执行任务。
# 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 = await 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 = await 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>，您可以指定重播过程的 task_id。 现在，kickoff 会将最新返回的任务输出保存在本地，以便您能够进行重播。使用 CLI 从特定任务重播
要使用重播功能，请按照以下步骤操作
- 打开您的终端或命令提示符。
- 导航到您的 CrewAI 项目所在的目录。
- 运行以下命令
要查看最新的 kickoff 任务 ID，请使用
然后，要从特定任务重播，请使用
crewai replay -t <task_id>
