团队

了解并使用 crewAI 框架中的团队，涵盖全面的属性与功能。

概述

crewAI 中的团队代表一组协作的智能体，共同完成一系列任务。每个团队都定义了任务执行、智能体协作以及整体工作流的策略。

团队属性

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

TIP

团队最大 RPM：max_rpm 属性设置团队每分钟可执行的最大请求数，以避免触发限流；一旦设置，它将覆盖各个智能体的 max_rpm 设置。

创建团队

在 CrewAI 中创建团队有两种常见方式：使用 JSONC 项目配置（推荐用于新团队），或直接在代码中定义。

JSONC 配置（推荐）

使用 crewai create crew <name> 创建的新项目使用 crew.jsonc 存放团队级设置和任务，并在 agents/ 中为每个智能体放置一个文件。

crewai run 会自动检测 crew.jsonc 或 crew.json，加载所引用的智能体，提示填写缺失的占位符，并启动团队。

`crew.jsonc` 示例

jsonc

{
  "name": "Market Research Crew",
  "agents": ["researcher", "analyst"],
  "tasks": [
    {
      "name": "research",
      "description": "Research {topic} and collect the most relevant facts.",
      "expected_output": "Structured research notes about {topic}.",
      "agent": "researcher"
    },
    {
      "name": "analysis",
      "description": "Analyze the research and write a concise report.",
      "expected_output": "A markdown report with findings and recommendations.",
      "agent": "analyst",
      "context": ["research"],
      "output_file": "output/report.md"
    }
  ],
  "process": "sequential",
  "verbose": true,
  "memory": true,
  "inputs": {
    "topic": "AI Agents"
  }
}

agents 中的每个字符串会先解析为 agents/<name>.jsonc，再解析为 agents/<name>.json。

jsonc

{
  "role": "{topic} Senior Researcher",
  "goal": "Find accurate and current information about {topic}.",
  "backstory": "You are a careful researcher who cites clear evidence.",
  "llm": "openai/gpt-4o",
  "tools": ["SerperDevTool"]
}

NOTE

当 process 为 "sequential" 时，任务按其在 tasks 中出现的顺序运行。

对于层级团队，请设置 "process": "hierarchical" 并提供 manager_llm 或 manager_agent。manager_agent 可以引用未包含在顶层 agents 列表中的 agents/<name>.jsonc 文件。

JSON 团队定义支持团队级字段，例如 process、verbose、memory、cache、max_rpm、planning、planning_llm、manager_llm、manager_agent、function_calling_llm、output_log_file、stream、tracing、before_kickoff_callbacks 和 after_kickoff_callbacks。

Python 回调和自定义类使用 {"python": "module.attribute"}。自定义工具使用 "custom:<name>"，并在运行时加载 tools/<name>.py。

WARNING

只从你信任的来源运行 JSON 团队项目。custom:<name> 工具和 {"python": "module.attribute"} 引用会在团队加载时执行本地 Python 代码。

经典 Python/YAML 配置

使用 crewai create crew <name> --classic 创建的经典项目使用 crew.py、config/agents.yaml、config/tasks.yaml 以及 @CrewBase、@agent、@task 和 @crew 装饰器。该模式仍然受支持，文档见使用注解。

直接在代码中定义（替代方式）

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

python

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
        )

如何运行上述代码：

python

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

在此示例中：

智能体和任务直接在类中定义，无需装饰器。
我们手动创建和管理智能体与任务列表。
这种方式提供更多控制，但在较大的项目中可维护性较差。

团队输出

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

团队输出属性

属性	参数	类型	描述
原始输出	`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 类提供了多种与该输出交互并呈现它的方式。

示例

python

# 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，则不会生成日志。

python

# 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

检查点

检查点功能让团队在关键事件（例如任务完成）后自动保存其状态，这样长时间运行或被中断的运行可以精确地从上次中断处恢复，而无需重新执行已完成的任务。

快速上手

传入 checkpoint=True 即可使用合理的默认值启用检查点（在每个任务后保存到 .checkpoints/）：

python

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 可对位置、触发事件、存储后端和保留策略进行细粒度控制：

python

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() 继续运行：

python

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

NOTE

从检查点恢复时，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`	保留的最大检查点数量。每次写入后会裁剪最旧的检查点。`None` 表示保留全部。

记忆利用

团队可以利用记忆（短期记忆、长期记忆和实体记忆）来增强其执行能力，并随着时间推移不断学习。此功能允许团队存储和召回执行记忆，从而辅助决策与任务执行策略。

缓存利用

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

团队使用指标

在团队执行完成后，你可以访问 usage_metrics 属性，查看团队执行的所有任务的语言模型（LLM）使用指标。这有助于了解运营效率以及可改进之处。

total_tokens 是计费总量（prompt_tokens + completion_tokens）。诸如 cached_prompt_tokens 和 cache_creation_tokens 这样的细分字段描述的是已包含在这些总量中的子集，不会额外累加到 total_tokens 之上。完整约定请参阅 Flows 概念文档中的 UsageMetrics 字段语义 一节。

python

# 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() 方法启动工作流。这会按照已定义的流程开始执行。

python

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

启动团队的不同方式

团队组建完成后，使用合适的 kickoff 方法启动工作流。CrewAI 提供了多种方法，以便更好地控制启动过程。

同步方法

kickoff()：按照已定义的流程开始执行。
kickoff_for_each()：针对集合中提供的每个输入事件或条目依次执行任务。

异步方法

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

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

NOTE

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

python

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

这些方法让你在管理并执行团队内任务时拥有灵活性，可按需定制同步与异步工作流。详细的异步示例请参阅异步启动团队指南。

流式传输团队执行

如需实时了解团队执行情况，可以启用流式传输，在输出生成时即刻接收：

python

# 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。

现在 kickoff 会在本地保存最近几次 kickoff 返回的任务输出，以便你从中重放。

使用 CLI 从特定任务重放

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

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

要查看最近几次 kickoff 的任务 ID，请使用：

shell

crewai log-tasks-outputs

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

shell

crewai replay -t <task_id>

这些命令可让你从最近几次 kickoff 任务重放，同时保留先前已执行任务的上下文。

团队

概述

团队属性

创建团队

JSONC 配置（推荐）

crew.jsonc 示例

经典 Python/YAML 配置

直接在代码中定义（替代方式）

团队输出

团队输出属性

团队输出的方法与属性

访问团队输出

示例

访问团队日志

检查点

快速上手

使用 CheckpointConfig 实现完全控制

从检查点恢复

CheckpointConfig 属性

记忆利用

缓存利用

团队使用指标

团队执行流程

启动团队

启动团队的不同方式

同步方法

异步方法

流式传输团队执行

从特定任务重放

使用 CLI 从特定任务重放

友情链接

`crew.jsonc` 示例

使用 `CheckpointConfig` 实现完全控制

`CheckpointConfig` 属性