​
什么是船员？

在 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用于存储执行记忆（短期、长期、实体记忆）。
记忆配置 (可选)memory_config船员使用的记忆提供程序的配置。
缓存 (可选)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_agentmanager 设置一个自定义代理作为管理器。
提示文件 (可选)prompt_file用于船员的提示 JSON 文件的路径。
规划 (可选)planning为船员添加规划能力。在每次船员迭代之前激活时，所有船员数据都会发送给 AgentPlanner，它将规划任务并将此计划添加到每个任务描述中。
规划 LLM (可选)planning_llm在规划过程中由 AgentPlanner 使用的语言模型。

船员最大 RPM：max_rpm 属性设置船员每分钟可执行的最大请求数，以避免速率限制，如果设置了此属性，它将覆盖单个代理的 max_rpm 设置。

​
创建船员

在 CrewAI 中创建船员有两种方式：使用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 包括最终任务输出的结果、token 使用情况以及单个任务输出。

​
船员输出属性

属性参数类型描述
原始rawstr船员的原始输出。这是输出的默认格式。
PydanticpydanticOptional[BaseModel]表示船员结构化输出的 Pydantic 模型对象。
JSON 字典json_dictOptional[Dict[str, Any]]表示船员 JSON 输出的字典。
任务输出tasks_outputList[TaskOutput]TaskOutput 对象的列表，每个对象代表船员中一个任务的输出。
Token 使用情况token_usageDict[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(布尔值) 或 file_name(字符串)，您可以实时查看船员执行日志。支持将事件记录为 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

​
记忆利用

船员可以利用记忆（短期记忆、长期记忆和实体记忆）来增强其执行能力并随时间进行学习。此功能允许船员存储和回忆执行记忆，有助于决策和任务执行策略。

​
缓存利用

可以使用缓存来存储工具的执行结果，通过减少重新执行相同任务的需要来提高流程效率。

​
船员使用指标

船员执行完成后，您可以访问 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 = 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 = 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。

启动现在将保存最新的启动返回的任务输出到本地，以便您能够从中重放。

​
使用 CLI 从特定任务重放

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

  1. 打开您的终端或命令提示符。
  2. 导航到您的 CrewAI 项目所在的目录。
  3. 运行以下命令

要查看最新的启动任务 ID，请使用

crewai log-tasks-outputs

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

crewai replay -t <task_id>

这些命令允许您从最新的启动任务重放，同时保留先前执行的任务的上下文。