团队

概览

在 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`	启用流式输出以在团队执行期间接收实时更新。返回一个可迭代获取 chunk 的 `CrewStreamingOutput` 对象。默认值为 `False`。
聊天 LLM (可选)	`chat_llm`	用于协调 `crewai chat` CLI 与团队交互的语言模型。可接受模型名字符串或 `LLM` 实例。默认值为 `None`。
Kickoff 前回调 (可选)	`before_kickoff_callbacks`	在团队启动前执行的一组可调用函数。每个回调都会接收并可修改 `inputs` 字典。与 `@before_kickoff` 装饰器不同。默认值为 `[]`。
Kickoff 后回调 (可选)	`after_kickoff_callbacks`	在团队完成后执行的一组可调用函数。每个回调都会接收并可修改 `CrewOutput`。与 `@after_kickoff` 装饰器不同。默认值为 `[]`。
追踪 (可选)	`tracing`	控制团队的 OpenTelemetry 追踪。`True` = 始终启用，`False` = 始终禁用，`None` = 继承环境/用户设置。默认值为 `None`。
技能 (可选)	`skills`	适用于团队内所有智能体的一组 `Path` 对象（技能搜索目录）或已预加载的 `Skill` 对象。默认值为 `None`。
安全配置 (可选)	`security_config`	管理团队指纹和身份的 `SecurityConfig` 实例。默认值为 `SecurityConfig()`。
检查点 (可选)	`checkpoint`	启用自动检查点。传入 `True` 可使用合理默认值，传入 `CheckpointConfig` 可完全控制，传入 `False` 可关闭，传入 `None` 可继承。见下方检查点章节。默认值为 `None`。

创建团队

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

JSONC 配置（推荐）

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

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

`crew.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。

{
  "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"]
}

对于分层团队，请设置 "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。

经典 Python/YAML 配置

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

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

你也可以在不使用 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`	`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(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

检查点

检查点允许团队在关键事件（例如任务完成）后自动保存状态，这样长时间运行或中断的执行可以准确地从离开的地方继续，而无需重新执行已完成的任务。

快速开始

传入 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()

`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）指标。这有助于洞察运行效率和改进空间。

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

启动团队的不同方式

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

同步方法

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

异步方法

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

方法	类型	描述
`akickoff()`	原生 async	在整个执行链中使用真正的 async/await
`akickoff_for_each()`	原生 async	对列表中的每个输入执行原生异步
`kickoff_async()`	基于线程	使用 `asyncio.to_thread` 包装同步执行
`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 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 项目所在目录。
运行以下命令：

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

crewai log-tasks-outputs

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

crewai replay -t <task_id>

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