CrewAI 中的团队（crew）代表一个协作的智能体（agent）群组，共同完成一系列任务。每个团队都定义了任务执行、智能体协作和整体工作流程的策略。
团队属性
| 属性 | 参数 | 描述 | |
|---|
| 任务 | 任务 (tasks) | 分配给团队的任务列表。 | |
| 智能体 | 智能体 (agents) | 构成团队的智能体列表。 | |
| 流程 （可选） | process | 团队遵循的流程（例如，顺序式、分层式）。默认为 sequential（顺序式）。 | |
| 详细程度 (Verbose) （可选） | verbose | 执行期间日志记录的详细程度。默认为 False。 | |
| 管理者大型语言模型 (Manager LLM) （可选） | manager_llm | 分层流程中管理者智能体使用的大型语言模型。在使用分层流程时必须。 | |
| 函数调用大型语言模型 (Function Calling LLM) （可选） | function_calling_llm | 如果传入此参数，团队将使用此 LLM 为团队中所有智能体的工具进行函数调用。每个智能体可以有自己的 LLM，它将覆盖团队的函数调用 LLM。 | |
| 配置 (Config) （可选） | config | 团队的可选配置设置，格式为 Json 或 Dict[str, Any]。 | |
| 最大每分钟请求数 (Max 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) （可选） | output_log_file | 设置为 True 可将日志保存为当前目录下的 logs.txt，或提供文件路径。如果文件名以 .json 结尾，日志将为 JSON 格式，否则为 .txt。默认为 None。 | |
| 管理者智能体 (Manager Agent) （可选） | manager_agent | manager 设置一个自定义智能体作为管理者。 | |
| 提示文件 (Prompt File) （可选） | prompt_file | 用于团队的提示 JSON 文件的路径。 | |
| 规划 (Planning) （可选） | planning | 为团队添加规划能力。激活后，在每次团队迭代之前，所有团队数据都会发送给 AgentPlanner，它将规划任务，并且此计划将被添加到每个任务描述中。 | |
| 规划大型语言模型 (Planning LLM) （可选） | planning_llm | AgentPlanner 在规划过程中使用的大型语言模型。 | |
| 知识来源 (Knowledge Sources) （可选） | knowledge_sources | 团队级别的可用知识来源，所有智能体均可访问。 | |
| 流式处理 (Stream) （可选） | stream | 启用流式输出以在团队执行期间接收实时更新。返回一个可迭代分块的 CrewStreamingOutput 对象。默认为 False。 | |
团队最大每分钟请求数 (Crew Max RPM)：max_rpm 属性设置团队为避免速率限制而可执行的最大每分钟请求数，如果您设置了它，它将覆盖单个智能体的 max_rpm 设置。
创建团队
在 CrewAI 中创建团队有两种方式：使用 YAML 配置（推荐） 或 直接在代码中 定义它们。
YAML 配置（推荐）
使用 YAML 配置提供了一种更清晰、更易维护的方式来定义团队，并与 CrewAI 项目中定义智能体和任务的方式保持一致。 按照安装部分所述创建 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.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 文件中提供了几个装饰器，用于标记团队类中的方法以进行特殊处理
@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 包括最终任务输出、令牌使用情况和单个任务输出的结果。
团队输出属性
| 属性 | 参数 | 类型 | 描述 |
|---|
| 原始 (Raw) | raw | str | 团队的原始输出。这是输出的默认格式。 |
| Pydantic | pydantic | Optional[BaseModel] | 一个 Pydantic 模型对象，表示团队的结构化输出。 |
| JSON 字典 (JSON Dict) | json_dict | Optional[Dict[str, Any]] | 一个字典，表示团队的 JSON 输出。 |
| 任务输出 (Tasks Output) | tasks_output | List[TaskOutput] | 一个 TaskOutput 对象列表，每个对象表示团队中一个任务的输出。 |
| 令牌使用情况 (Token Usage) | token_usage | Dict[str, Any] | 令牌使用情况摘要，提供对语言模型执行期间性能的洞察。 |
团队输出方法和属性
| 方法/属性 | 描述 |
|---|
| 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(Boolean) 或 file_name(str) 来查看团队执行的实时日志。支持以 file_name.txt 和 file_name.json 两种格式记录事件。如果为 True(Boolean)，则将保存为 logs.txt。 如果 output_log_file 设置为 False(Boolean) 或 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
内存利用
团队可以利用内存（短期、长期和实体记忆）来增强其执行和长期学习能力。此功能允许团队存储和召回执行记忆，有助于决策和任务执行策略。
缓存利用
可以利用缓存来存储工具的执行结果，通过减少重新执行相同任务的需要来提高效率。
团队使用指标
团队执行后，您可以访问 usage_metrics 属性以查看团队执行的所有任务的语言模型 (LLM) 使用指标。这提供了对操作效率和改进领域的洞察。
# 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() 方法启动工作流。这将根据定义的流程开始执行过程。
# 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()：为每个提供的输入事件或项目并发执行任务，利用异步处理。
# 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)
流式团队执行
为了实时查看团队执行情况，您可以启用流式传输以在生成时接收输出
# 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 replay -t <task_id>
