Crewai Community Version(四)——Crew
目录
- 1. Crew总览
- 2. Crew属性
- 3. 与Crew相关的装饰器
- 4. Crew输出
- 5. CrewOutput属性
- 6. Crew使用指标
- 7. Crew执行流程
- 8. 启动Crew的方式
- 9. 重放特定任务
- 10. 创建Crew
- 10.1 YAML配置
- 10.2 直接代码定义
- 10.3 运行结果
- 参考
1. Crew总览
CrewAI框架中的Crew是一组协作的代理，它们一起工作以完成一组任务。每个小组定义任务执行、智能体协作和整个工作流程的策略。
2. Crew属性
| 属性 | 参数 | 含义 |
|---|---|---|
| 任务 | tasks | 分配给crew的任务清单 |
| 智能体 | agents | crew中智能体的名单 |
| 过程 | process | crew遵循的处理流程，默认为sequential |
| 详细日志 | verbose | 执行期间记录的详细程度，默认为False |
| 大模型协调官 | manager_llm | 在hierarchical过程必须使用的大语言模型 |
| 函数调用大语言模型 | function_calling_llm | 如果设置，crew将使用该大语言模型来对所有智能体的工具进行函数调用。每个智能体可以有自己的函数调用大语言模型，它会覆盖crew的函数调用大语言模型进行函数调用 |
| 配置 | config | 可选的crew配置，格式为json或dict |
| 每分钟最大请求数 | max_rpm | 在执行过程中，crew每分钟的最大请求数，默认值为None |
| 记忆 | memory | 用于存储执行记忆 |
| 记忆配置 | memory_config | crew使用的记忆配置 |
| 缓存 | cache | 指定是否使用缓存来存储工具执行的结果，默认值为True |
| 嵌入器 | embedder | crew使用的嵌入器配置，目前主要由记忆使用，默认值为{“provider": “openai”} |
| 步骤回调 | step_callback | 在每个智能体进行一个步骤之后调用的函数，可用于记录智能体的操作或执行其他操作。它不会覆盖特定于智能体的step_callback |
| 任务回调 | task_callback | 在每个任务完成后调用的函数，用于监视或任务执行后的其他操作 |
| 分享crew | share_crew | 您是否希望与CrewAI团队共享完整的crew信息和执行，以使库更好，并允许我们训练模型 |
| 输出日志文件 | output_log_file | 设置为True，表示将日志保存为当前目录下的logs.txt或提供文件路径。如果文件名以.JSON结尾，日志将是JSON格式，否则是txt。默认值为None |
| 协调智能体 | manager_agent | 用作协调智能体的自定义智能体 |
| 提示文件 | prompt_file | crew使用的JSON提示文件 |
| 计划 | planning | 为crew增加规划能力。在每次crew迭代之前激活，所有crew数据将发送给AgentPlanner，后者将规划任务，并将此计划添加到每个任务描述中 |
| 用于规划的大语言模型 | planning_llm | AgentPlanner在规划过程中使用的语言模型 |
3. 与Crew相关的装饰器
CrewAI在annotations.py文件中提供了几个装饰器，用于标记船员类中的方法以进行特殊处理：
| 装饰器 | 作用 |
|---|---|
| @CrewBase | 将该类标记为crew基类 |
| @agent | 表示返回一个Agent对象的方法 |
| @task | 表示返回一个Task对象的方法 |
| @crew | 表示返回一个Crew对象的方法 |
| @before_kickoff | 标记一个方法在crew启动前执行 |
| @after_kickoff | 标记一个方法在crew完成后执行 |
4. Crew输出
CrewAI将crew的输出封装在CrewOutput类中，该类提供了一种结构化的方式来访问crew执行的结果，包括各种格式，如原始字符串、JSON和Pydantic模型。CrewOutput包括来自最终任务输出、词元使用情况和单个任务输出的结果。此外，一旦执行了crew，就可以通过Crew对象的output属性访问它的输出。
5. CrewOutput属性
| 属性 | 参数 | 类型 | 含义 |
|---|---|---|---|
| 原始字符串 | raw | str | crew的原始输出，这是输出的默认格式 |
| Pydantic模型 | pydantic | Optional[BaseModel] | 一个Pydantic模型对象，表示crew的结构化输出 |
| JSON字典 | json_dict | Optional[Dict[str, Any]] | 表示crew JSON输出的字典 |
| 任务输出 | tasks_output | List[TaskOutput] | TaskOutput对象的列表，每个对象表示crew中一个任务的输出 |
| 词元使用情况 | token_usage | Dict[str, Any] |
6. Crew使用指标
在crew执行后，您可以访问usage_metrics属性，以查看crew执行所有任务时语言模型使用指标，这提供了对操作效率和改进领域的见解。
7. Crew执行流程
1. 顺序流程：任务一个接一个地执行，允许线性工作流。
2. 分级流程：协调智能体协调crew，委派任务并在继续之前验证结果。注意：此流程需要manager_llm或manager_agent，这对于验证流程至关重要。
8. 启动Crew的方式
CrewAI提供了几种更好地控制启动过程的方法：
| 启动方式 | 具体流程 |
|---|---|
| kickoff() | 按照定义的进程流启动执行流程 |
| kickoff_for_each() | 为集合中提供的每个输入事件或项依次执行任务 |
| kickoff_async() | 异步启动工作流 |
| kickoff_for_each_async() | 利用异步处理，为每个提供的输入事件或项并发地执行任务 |
9. 重放特定任务
在CrewAI项目所在目录下，执行如下命令：crewai log-tasks-outputs，但是该项目至少执行过一次才能有输出。示例如下：
接着通过上面获取的task_id，重放特定任务：crewai replay -t <task_id>。
10. 创建Crew
10.1 YAML配置
通过crewai create crew <your-project-name>后，在<your-project-name>/src/<your-project-name>/config/agents.yaml中以如下的形式定义Agents：
news_reporter:role: >新闻播报员goal: >从科技、人文、军事等方面寻找当前时间{time}范围内的新闻backstory: >深入基层实地寻找过新闻，能辨别虚假消息，具备良好的新闻写作能力
在<your-project-name>/src/<your-project-name>/config/tasks.yaml中以如下的形式定义Tasks：
summary_task:description: >总结当前时间{time}范围内的新闻expected_output: >一份简练但又内容丰富的新闻稿子agent: news_reporter
;在<your-project-name>/src/<your-project-name>/crew.py创建一个继承CrewBase的类：
from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
from crewai.agents.agent_builder.base_agent import BaseAgent
from typing import List
# If you want to run a snippet of code before or after the crew starts,
# you can use the @before_kickoff and @after_kickoff decorators
# https://docs.crewai.com/concepts/crews#example-crew-class-with-decorators@CrewBase
class NewsReporter():"""NewsReporter crew"""agents: List[BaseAgent]tasks: List[Task]# Learn more about YAML configuration files here:# Agents: https://docs.crewai.com/concepts/agents#yaml-configuration-recommended# Tasks: https://docs.crewai.com/concepts/tasks#yaml-configuration-recommended# If you would like to add tools to your agents, you can learn more about it here:# https://docs.crewai.com/concepts/agents#agent-tools@agentdef news_reporter(self) -> Agent:return Agent(config=self.agents_config['news_reporter'], # type: ignore[index]verbose=True)# To learn more about structured task outputs,# task dependencies, and task callbacks, check out the documentation:# https://docs.crewai.com/concepts/tasks#overview-of-a-task@taskdef summary_task(self) -> Task:return Task(config=self.tasks_config['summary_task'], # type: ignore[index])@crewdef crew(self) -> Crew:"""Creates the NewsReporter crew"""# To learn how to add knowledge sources to your crew, check out the documentation:# https://docs.crewai.com/concepts/knowledge#what-is-knowledgereturn Crew(agents=self.agents, # Automatically created by the @agent decoratortasks=self.tasks, # Automatically created by the @task decoratorprocess=Process.sequential,verbose=True,planning=True,planning_llm="deepseek/deepseek-chat"# process=Process.hierarchical, # In case you wanna use that instead https://docs.crewai.com/how-to/Hierarchical/)在<your-project-name>/src/<your-project-name>/main.py中修改inputs为{”time": str(datetime.now())}：
#!/usr/bin/env python
import sys
import warningsfrom datetime import datetimefrom news_reporter.crew import NewsReporterwarnings.filterwarnings("ignore", category=SyntaxWarning, module="pysbd")# This main file is intended to be a way for you to run your
# crew locally, so refrain from adding unnecessary logic into this file.
# Replace with inputs you want to test with, it will automatically
# interpolate any tasks and agents informationdef run():"""Run the crew."""inputs = {'time': str(datetime.now())}try:NewsReporter().crew().kickoff(inputs=inputs)except Exception as e:raise Exception(f"An error occurred while running the crew: {e}")def train():"""Train the crew for a given number of iterations."""inputs = {'time': str(datetime.now())}try:NewsReporter().crew().train(n_iterations=int(sys.argv[1]), filename=sys.argv[2], inputs=inputs)except Exception as e:raise Exception(f"An error occurred while training the crew: {e}")def replay():"""Replay the crew execution from a specific task."""try:NewsReporter().crew().replay(task_id=sys.argv[1])except Exception as e:raise Exception(f"An error occurred while replaying the crew: {e}")def test():"""Test the crew execution and returns the results."""inputs = {'time': str(datetime.now())}try:NewsReporter().crew().test(n_iterations=int(sys.argv[1]), eval_llm=sys.argv[2], inputs=inputs)except Exception as e:raise Exception(f"An error occurred while testing the crew: {e}")10.2 直接代码定义
from datetime import datetime
from crewai import Agent, Crew, Process, Tasknews_reporter = Agent(role="""新闻播报员""",goal="""从科技、人文、军事等方面寻找当前时间{time}范围内的新闻""",backstory="""深入基层实地寻找过新闻，能辨别虚假消息，具备良好的新闻写作能力""",verbose=True
)summary_task = Task(description="""总结当前时间{time}范围内的新闻""",expected_output="""一份简练但又内容丰富的新闻稿子""",agent=news_reporter
)NewsReporter = Crew(agents=[news_reporter],tasks=[summary_task],process=Process.sequential,verbose=True,planning=True,planning_llm="deepseek/deepseek-chat"
)NewsReporter.crew().kickoff(inputs={"time": str(datetime.now())}
)
10.3 运行结果
运行结果如下：
由于启动了planning，因此planning_llm在任务的description新增了一些内容，以使任务顺利完成。
参考
https://docs.crewai.com/concepts/crews