在当今快速发展的技术环境中，人工智能（AI）智能体已成为优化工作流程、提高效率和减少人力成本的关键工具。
谷歌近期发布了AI Agent 白皮书，业内人士甚至直接定义，2025 年是 Agent 时代。
而我最近尝试的Crew.ai，是一个先进的AI智能体平台，通过其独特的工作分配模式，为企业和开发者提供了强大的自动化解决方案。
今天，带着两个案例，跟各位深入探讨Crew.ai的基础概念、优势特性、基础组件以及其代理流程，帮助大家全面理解这一技术如何在实际中应用。
不过开始之前，要先回顾一下 AI 智能体的基础概念和 Crew.ai 对应的支持组件。
AI智能体的基础概念
首先，先来看一下，AI 智能体的组成部分。
智能体（Agent）
AI智能体是模拟人类决策过程的软件实体，能够自主执行任务并作出决策。在Crew.ai中，智能体通过学习和适应环境来优化其性能。
任务（Action）
任务是智能体执行的具体操作，可以是数据处理、图像识别或任何其他自动化操作。Crew.ai智能体能够识别任务类型，并根据任务需求选择合适的工具和方法。
工具（Tools）
智能体依赖于各种工具和技术，如机器学习模型和数据处理工具，来执行任务。这些工具的选择和配置直接影响智能体的效率和效果。
知识库（Memory）
知识库是智能体存储和检索信息的核心，它包含了智能体执行任务所需的所有数据和规则。Crew.ai通过高效的知识管理，确保智能体能够快速准确地访问所需信息。
流程管理（Planning）
流程管理是智能体完成任务的关键。Crew.ai通过定义清晰的流程，确保任务能够按计划执行，同时允许灵活调整以适应变化的需求。
Crew.ai的优势和特性
Crew.ai以其高效性、灵活性和可扩展性脱颖而出。它能够快速处理复杂任务，适应不同的工作环境，并支持大规模部署和集成，使其成为企业自动化流程的理想选择。
Crew.ai的基础组件
代理（Agent）
- 定义：代理是CrewAI中的自主单元，具备执行任务、做出决策和与其他代理通信的能力，能够模拟人类的决策过程。
- 功能：代理通过学习和适应环境来优化性能，能够识别任务类型并选择合适的工具和方法来执行任务，如研究员代理负责收集和分析信息，作家代理专注于创作内容。
任务（Task）
- 定义：任务是代理需要完成的具体工作，包含执行所需的各种细节，如清晰的描述、负责执行的代理、所需工具等。
- 属性：任务具有多种属性，包括描述、代理、预期输出、工具、异步执行、上下文、配置、输出格式、回调函数等，这些属性共同定义了任务的执行细节和期望结果。
工具（Tool）
- 定义：工具是代理在执行任务时所依赖的各种技术和软件，如机器学习模型、数据处理工具等，赋予代理多种能力。
- 特征：工具具有实用性、集成性、可定制性、错误处理和缓存机制等关键特征，能够无缝集成到工作流程中，提升代理的整体能力，实现高效协作。
流程（Process）
- 定义：流程是代理完成任务的一系列步骤和规则，确保任务能够按计划执行，使个体代理能够作为一个有凝聚力的整体运作。
- 类型：流程包括顺序执行（任务按预定义顺序执行）、分级（通过管理者代理监督任务执行）和共识流程（计划中，实现代理之间的协作决策-这个还未实现）。
团队（Crews）
- 定义：团队是由多个代理组成的集合，共同协作完成复杂的任务，通过合理的流程设计，提高整体工作效率。
- 属性：团队具有任务、代理、流程、详细程度、经理LLM、配置、最大RPM、语言、语言文件、内存、缓存、嵌入器、完整输出、步骤回调、任务回调等属性，这些属性共同定义了团队的行为和执行策略。
记忆（Memory）
- 定义：记忆系统是CrewAI中的一个重要组成部分，包括短期记忆、长期记忆、实体记忆和上下文记忆，用于存储和检索信息。
- 功能：记忆系统通过存储和利用过去的经验和信息，增强代理的上下文意识、经验积累和实体理解能力，从而提升代理的决策质量和问题解决能力。
今天，重点聊一下 Crew.ai 的流程 Process。
Crew.ai的代理流程
当前，Crew.ai 支持的流程有两种模式，一种是顺序流程，另一种是层级流程。
简单的用大白话来说，顺序流程就是流水线，而层级流程，就是有一个领导 Agent 分派小弟干活儿。
顺序流程（Sequential Process）
顺序流程是指任务按照预定的顺序依次执行。只有上一个 Agent 的任务完成了，下一个 Agent 的任务才会启动。
例如这个内容撰写团队，需要由内容规划师制定内容计划，再由内容作者撰写文章，最后，交给内容编辑进行内容的编辑优化。
整个内容撰写团队的顺序工作模式示意图：
这里是我写了一个 Gradio 页面，可以让大家更加直观的看到整个智能体团队的输入和输出（您可能留意到了，这篇文章，就是这个智能体团队帮我写的，哈哈哈哈）；
对应的agent 定义、任务分配和工作流程代码如下：
def create\_agents(topic):
"""创建智能体"""
planner = Agent(
role="内容规划师",
goal=f"规划关于 {topic} 的引人入胜且事实准确的内容",
backstory=f"您正在规划一篇主题和内容结构是: {topic} 的技术博客文章。"
"您收集有助于受众学习的"
"信息。"
"您的工作是内容作者撰写该主题文章的基础。",
allow\_delegation=False,
verbose=True,
llm=llm
)
writer = Agent(
role="内容作者",
goal=f"根据规划撰写关于主题和内容结构是: {topic} 的高质量文章",
backstory="您是一位经验丰富的技术文章作者，"
"擅长将复杂的概念转化为易于理解的内容。",
allow\_delegation=False,
verbose=True,
llm=llm
)
editor = Agent(
role="内容编辑",
goal="审查和优化文章质量",
backstory="您是一位资深编辑，"
"确保文章的准确性、可读性和吸引力。",
allow\_delegation=False,
verbose=True,
llm=llm
)
return planner, writer, editor
def create\_tasks(topic, planner, writer, editor):
"""创建任务"""
plan = Task(
description=(
f"1. 优先考虑关于主题和内容结构是: {topic} 的最新趋势、关键参与者和"
"值得注意的新闻。\n"
"2. 确定目标受众，考虑"
"他们的兴趣和痛点。\n"
"3. 根据输入的{topic},优化详细的内容大纲，包括"
"背景和要点。\n"
"4. 包括 SEO 关键词和相关数据或来源。"
),
expected\_output="一份全面的内容计划文档，"
"包含大纲、受众分析、"
"SEO 关键词和资源。",
agent=planner
)
write = Task(
description=(
"根据内容计划撰写文章，确保：\n"
"1. 内容清晰、引人入胜\n"
"2. 使用具体示例和数据支持观点\n"
"3. 适当使用过渡语句\n"
"4. 包含引人入胜的开场和有力的结论"
),
expected\_output="一篇结构完整、内容丰富的文章初稿",
agent=writer
)
edit = Task(
description=(
"审查并优化文章：\n"
"1. 检查事实准确性和逻辑连贯性\n"
"2. 优化标题和小标题\n"
"3. 改进段落结构和文章流畅度\n"
"4. 确保语言清晰简洁\n"
"5. 检查拼写和语法"
),
expected\_output="一篇经过优化的最终文章",
agent=editor
)
return [plan, write, edit]
def generate\_article(topic):
"""生成文章的主要函数"""
try:
# 创建智能体和任务
planner, writer, editor = create\_agents(topic)
tasks = create\_tasks(topic, planner, writer, editor)
# 创建 Crew
crew = Crew(
agents=[planner, writer, editor],
tasks=tasks,
verbose=True
)
# 执行任务流程并获取结果
result = crew.kickoff(inputs={"topic": topic})
# 获取最终的文章内容
try:
# 尝试获取最后一个任务的输出
tasks\_outputs = result.tasks\_outputs
if tasks\_outputs and len(tasks\_outputs) > 0:
final\_content = tasks\_outputs[-1] # 获取最后一个任务的输出
else:
final\_content = str(result)
except AttributeError:
try:
final\_content = str(result)
except:
final\_content = "无法获取文章内容"
return final\_content
except Exception as e:
print(f"发生错误: {str(e)}")
import traceback
print(traceback.format\_exc())
return None
层级流程（Hierarchical Process）
层级流程涉及任务按层级结构分配和执行。 注意，这个流程中，是有领导的。
在项目管理中，从项目规划、任务分配到执行监控，每个层级都有明确的责任和任务。
案例里是一个处理客户问题的团队。客户支持经理是整个团队的管理者，分别管理技术支持专家和账单支持专家。在这里，客户支持经理需要判断客户提问的问题是账单问题还是技术问题。判断完成后，指派对应的专家处理。最终，客户问题报告也是由客户支持经理对外提供的。
以下是当团队接收到一个工单内容为：这个月怎么多扣了 100 元，整个团队的工作过程。
同样的，我也把对应的 Gradio 界面贴出来，大家可以直观的看到对应的团队工作成果。
对应的代码示意如下：
1-agents.yaml
manager:
role: >
客户支持经理
goal: >
监督支持团队，确保及时有效地解决客户咨询
backstory: >
你是一位经验丰富的客户支持经理，具有领导支持团队的广泛经验。
你的主要职责是协调支持代理的工作，确保客户问题得到及时和满意的解决。
你擅长任务分配、绩效监控，并保持高标准的客户服务。
technical\_support:
role: >
技术支持专家
goal: >
及时有效地解决客户报告的技术问题
backstory: >
你是一位技术支持专家，拥有丰富的故障排除软件和硬件问题的背景。
你的主要职责是协助客户解决技术问题，确保他们的满意度和产品的顺利运行。
billing\_support:
role: >
账单支持专家
goal: >
处理与账单、支付和账户管理相关的客户咨询
backstory: >
你是一位经验丰富的账单支持专家，擅长处理客户的账单咨询。
你的主要目标是提供清晰准确的账单流程信息，解决支付问题，并协助账户管理，以确保客户满意。
2-task.yaml
categorize\_tickets:
description: >
根据内容对来电客户支持工单进行分类：'{ticket}'，
以确定它是技术问题还是账单问题。
expected\_output: >
一个标记为'技术'或'账单'的分类工单。
agent: manager
resolve\_technical\_issues:
description: >
解决工单中描述的技术问题：'{ticket}'
expected\_output: >
为每个技术问题提供详细解决方案。
agent: technical\_support
resolve\_billing\_issues:
description: >
解决工单中描述的账单问题：'{ticket}'
expected\_output: >
为每个与账单相关的咨询提供全面的回应。
agent: billing\_support
quality\_assurance\_review:
description: >
审核提供的技术和账单问题的解决方案，以确保准确性和客户满意度。
expected\_output: >
一份确认解决方案质量和准确性的报告。
agent: manager
3-crew.py
from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
from crewai\_tools import SerperDevTool, ScrapeWebsiteTool, FileWriterTool
from langchain\_openai import ChatOpenAI
from dotenv import load\_dotenv
import os
from docx import Document
# 加载环境变量
load\_dotenv()
# 初始化 LLM，使用 Deepseek API
llm = ChatOpenAI(
api\_key=os.getenv('DEEPSEEK\_API\_KEY'),
base\_url=os.getenv('DEEPSEEK\_API\_BASE'),
model\_name="deepseek/deepseek-chat",
temperature=0.7,
)
class CustomFileWriterTool(FileWriterTool):
def \_write\_file(self, content: str, filename: str) -> str:
"""自定义文件写入工具"""
try:
# 确保文件名安全
safe\_filename = "".join([c for c in filename if c.isalnum() or c in (' ', '-', '\_')]).rstrip()
if not safe\_filename:
safe\_filename = "article"
# 添加 .docx 扩展名（如果没有的话）
if not safe\_filename.endswith('.docx'):
safe\_filename += '.docx'
# 创建 Word 文档
doc = Document()
doc.add\_heading(filename, 0)
doc.add\_paragraph(content)
# 保存文档
doc.save(safe\_filename)
return f"文章已成功保存为 {safe\_filename}"
except Exception as e:
return f"保存文件时发生错误: {str(e)}"
@CrewBase
class SequentialProcessExample:
"""ProcessExample crew"""
# 使用相对路径
base\_path = os.path.dirname(os.path.abspath(\_\_file\_\_))
agents\_config = os.path.join(base\_path, 'agents.yaml')
tasks\_config = os.path.join(base\_path, 'tasks.yaml')
@task
def categorize\_tickets(self) -> Task:
return Task(config=self.tasks\_config['categorize\_tickets'], verbose=True)
@task
def resolve\_technical\_issues(self) -> Task:
return Task(config=self.tasks\_config['resolve\_technical\_issues'], verbose=True)
@task
def resolve\_billing\_issues(self) -> Task:
return Task(config=self.tasks\_config['resolve\_billing\_issues'], verbose=True)
@task
def quality\_assurance\_review(self) -> Task:
return Task(config=self.tasks\_config['quality\_assurance\_review'], verbose=True)
@agent
def technical\_support(self) -> Agent:
return Agent(
config=self.agents\_config['technical\_support'],
verbose=True,
llm=llm
)
@agent
def billing\_support(self) -> Agent:
return Agent(
config=self.agents\_config['billing\_support'],
verbose=True,
llm=llm
)
@agent
def manager(self) -> Agent:
return Agent(
config=self.agents\_config['manager'],
verbose=True,
llm=llm
)
@crew
def crew(self) -> Crew:
return Crew(
agents=[self.technical\_support(), self.billing\_support()],
tasks=[
self.categorize\_tickets(),
self.resolve\_technical\_issues(),
self.resolve\_billing\_issues(),
self.quality\_assurance\_review()
],
manager\_agent=self.manager(),
process=Process.hierarchical,
verbose=True,
)
通过本文的介绍，希望您能够对Crew.ai的智能体工作分配模式有一个全面的了解，并考虑如何将其应用于自己的业务中，以实现更高的自动化和效率。
后续将持续推出使用 Crew.ai 进行实际业务支持的案例分享。喜欢就点个再看，么么哒。