1 CrewAI 基础
在构建现代人工智能系统的过程中,AI 智能体(AI Agents)展现出相较于传统方法更强大的灵活性与自主性。
1.1 AI 智能体的动机
- RAG 系统的视角:虽然检索增强生成(RAG)技术在信息扩展方面表现优异,但其程序化的流程限制了系统的自主决策能力。相比之下,AI 智能体具备动态推理和规划能力,能够根据任务需求自主选择信息源,实现更高程度的任务自动化。
- 软件开发视角:传统软件依赖于固定输入、预设逻辑和标准化输出,而 AI 智能体则支持多样化的输入格式(如文本、PDF、JSON 等),并基于大语言模型(LLM)进行灵活转换,输出形式也更加丰富,包括代码、结构化数据等,极大提升了系统的适应性和扩展性。
- 自主系统视角:AI 智能体突破了传统 LLM 需要频繁人工干预的交互模式,能够将复杂目标分解为子任务,逐步执行并自我优化,从而显著减少人为参与,提升整体效率。
1.2 AI Agents 的构建模块
- 角色扮演(Role-playing):赋予智能体明确的专业身份,有助于提升响应的专业性和一致性。例如,“企业法律事务律师"比"通用助手"更具针对性。
- 任务聚焦(Focus/Tasks):专业化分工优于多功能集成。每个智能体应专注于单一任务,通过多智能体协作完成复杂工作流,避免功能重叠和冲突。
- 工具集成(Tools):外部工具的使用极大增强了智能体的能力。工具选择应遵循"精简有效"的原则,优先配置完成任务所需的核心工具。
- 协同机制(Cooperation):多智能体系统通过任务划分与上下文共享,实现相互反馈与结果优化,从而提升整体系统的决策质量与执行效率。
- 安全护栏(Guardrails):为防止幻觉、无限循环或错误决策,需设置使用限制、验证检查点及回退机制,确保智能体行为可控可靠。
- 记忆管理(Memory):智能体的记忆可分为短期记忆(如对话历史)、长期记忆(如用户偏好)和实体记忆(如客户信息),合理设计记忆机制有助于提升个性化服务能力和系统连续性。
AI 智能体代表了从传统自动化向真正自主智能系统演进的重要方向。通过科学的角色定义、任务分配、工具整合与协作机制,结合有效的安全控制与记忆系统,AI 智能体能够在复杂环境中实现高效、精准的任务执行,成为未来智能系统构建的关键支柱。
1.3 使用 CrewAI 构建 Agent
构建一个功能性的 AI Agent 系统在 CrewAI 中被清晰地划分为三个核心步骤:定义智能体 (Agents)、分配任务 (Tasks) 和 组建团队 (Crew)。
智能体是您团队中的"执行者”。它不仅仅是一个简单的语言模型调用,而是一个被赋予了特定 角色 (Role)、目标 (Goal) 和 背景故事 (Backstory) 的虚拟专家。这种设定为其行为和决策提供了丰富的上下文,使其输出更具针对性和专业性。
有了智能体,我们还需要为其分配具体的工作,这就是 任务 (Task) 的作用。一个好的任务定义应该清晰、具体,并包含对最终成果的预期。
当智能体和任务都准备就绪后,最后一步就是将它们组织起来,形成一个 团队 (Crew)。Crew 负责管理整个工作流程的执行。
from crewai import Agent, Task, Crew
# 创建智能体
senior_technical_writer = Agent(
role="高级技术写作专家",
goal="创作发布级别的技术文章",
backstory="您是一位经验丰富的研究员,擅长发现AI领域的最新发展。xxx",
tools=[file_write_tool],
verbose=True,
llm=llm # 指定使用的语言模型
)
# 创建任务
writing_task = Task(
description="基于主题 {topic} 撰写一篇全面的技术文章",
agent=senior_technical_writer,
tools=[file_write_tool],
expected_output="一篇结构化的技术文章,包含引言、主要内容和结论"
)
# 组建团队
crew = Crew(
agents=[senior_technical_writer],
tasks=[writing_task],
verbose=True
)
# 执行任务
result = crew.kickoff(inputs={"topic": "AI智能体的未来发展"})
role
(str): 智能体的"职位头衔"。它明确了智能体在团队中的身份和职责,例如"市场分析师"、“资深软件工程师"或"代码审查员”。这直接影响其沟通风格和专注点。goal
(str): 智能体的核心任务目标。这是它所有行动的最终导向,是衡量其工作是否成功的标准。一个清晰的goal
能让智能体更有效地规划步骤。backstory
(str): 这是塑造智能体"性格"和"经验"的关键。一段丰富的背景故事能为 LLM 提供强大的上下文,使其在生成内容时,无论是语气、措辞还是思考深度,都更贴近所扮演的角色。description
(str): 任务的核心描述。这是给智能体的"任务简报"。注意,这里使用了{topic}
占位符,这使得任务可以动态接收输入,极大地提高了复用性。agent
(Agent): 指定执行此任务的智能体实例。通过这个参数,我们将writing_task
与senior_technical_writer
绑定在一起。expected_output
(str): 定义任务的"完成标准 (Definition of Done)"。这部分至关重要,它为智能体提供了明确的交付物形态和质量要求,有效避免了结果的模糊和不可控。描述越具体,产出质量越高。
CrewAI 的真正威力体现在多智能体协作上。你可以定义一个由研究员、写作者、审查员等组成的流水线,让它们接力完成复杂任务。
# 创建多智能体工作组
multi_agent_crew = Crew(
agents=[research_agent, summarization_agent, fact_checking_agent],
tasks=[research_task, summarization_task, fact_checking_task],
process=Process.sequential, # 按顺序执行任务
verbose=True
)
# 执行工作流
result = multi_agent_crew.kickoff(inputs={"topic": "量子计算的最新突破"})
agents
(list): 包含此团队中所有智能体实例的列表。tasks
(list): 包含需要被执行的所有任务实例的列表。process
(Process): 定义任务的执行流程。这是多智能体协作的核心。Process.sequential
(顺序流程): 任务将按照它们在tasks
列表中的顺序依次执行。前一个任务的输出会自动作为后一个任务的上下文,形成一个完美的工作流管道。Process.hierarchical
(层级流程): 更复杂的模式,通常会有一个"经理"智能体来协调和委派任务给其他"员工"智能体。
kickoff(inputs: dict)
: 这是启动整个团队工作的入口方法。inputs
字典用于填充任务描述中定义的占位符(如{topic}
),从而启动整个流程。
1.4 模块化
随着智能体系统复杂度的增加,将所有配置(角色、任务、背景故事等)硬编码在主脚本中会变得混乱且难以维护。为了解决这个问题,我们可以采用一种更优雅、更具扩展性的模块化架构。将智能体和任务的定义从执行逻辑中分离出来,通过独立的 YAML 配置文件进行管理,并使用一个专门的 Python 类来组织和加载整个工作流(Crew)。
我们将通过三个核心步骤实现模块化:
- 分离配置文件:创建独立的 YAML 文件来定义智能体 (
agents.yaml
) 和任务 (tasks.yaml
)。 - 创建模块化 Crew 类:使用
crewai
提供的装饰器 (@CrewBase
,@agent
,@task
,@crew
) 在一个 Python 类中"声明式"地组装工作流。 - 简洁的执行入口:在主程序中,只需实例化该类并调用其
crew()
方法即可启动任务。
@CrewBase
class ResearchCrew:
"""一个负责开展研究、归纳成果并进行事实核查的Crew"""
agents_config = 'config/agents.yaml'
tasks_config = 'config/tasks.yaml'
def __init__(self):
self.search_tool = SerperDevTool()
self.llm = llm
@agent
def research_agent(self) -> Agent:
return Agent(
config=self.agents_config['research_agent'],
tools=[self.search_tool],
llm=self.llm
)
@task
def research_task(self) -> Task:
return Task(
config=self.tasks_config['research_task'],
tools=[self.search_tool],
)
@crew
def crew(self) -> Crew:
return Crew(
agents=self.agents,
tasks=self.tasks,
process=Process.sequential,
)
@CrewBase
装饰器提供的便利使得指定路径 (agents_config
) 就能当字典 (self.agents_config[…]
) 使用。其内部工作流如下:
- 识别配置路径:当
@CrewBase
装饰器处理ResearchCrew
类时,它会扫描类属性,找到了agents_config = 'config/agents.yaml'
和tasks_config = 'config/tasks.yaml'
。 - 自动加载和解析:在类的
__init__
方法被调用之前或期间,@CrewBase
的底层逻辑会自动读取这些路径指向的 YAML 文件。 - 数据转换:它使用一个 YAML 解析库(如 PyYAML)将文件内容转换成 Python 的字典对象。
- 属性覆盖:最后,它将解析后的字典覆盖回同名的实例属性上。
self.agents
和 self.tasks
没有在 __init__
中被显式定义。它们是由 @agent
和 @task
装饰器动态注入到类实例中的。工作流程如下:
- 当你创建实例
crew_instance = ResearchCrew()
时,@CrewBase
的逻辑开始运行。 - 框架会查找所有被
@agent
装饰的方法(这里是research_agent
),并执行它。 research_agent()
方法返回一个Agent
对象。@agent
装饰器捕获这个返回的对象,并将它添加到一个名为agents
的列表中。这个列表被附加到crew_instance
实例上,成为self.agents
。- 对所有
@task
装饰的方法执行完全相同的过程,最终生成self.tasks
列表。
1.5 结构化输出
默认情况下,大语言模型(LLM)就像一位才华横溢但天马行空的艺术家,其输出是自由流动的文本。这在日常对话中表现出色,但在需要程序化处理的自动化工作流中,却是一个巨大的挑战。程序如何才能可靠地使用一个时而是段落、时而是列表的输出呢?
解决方案是建立一份"契约"。我们需要强制 LLM 的输出遵循一个可预测的、结构化的格式,比如 JSON。这就是结构化输出的用武之地,而 Pydantic 正是我们用来定义这份"数据契约"的首选工具。
核心原理: 我们不再仅仅要求 LLM 描述一个答案,而是给它一个数据模板,并要求它填写这份模板。
首先,我们使用 Pydantic 的 BaseModel
来精确定义我们期望的数据结构。这个类就像一个蓝图。至关重要的是,我们为每个字段提供的 description
(描述)会成为给 LLM 的直接指令。
假设我们需要提取货币转换所需的详细信息,我们的契约可以这样定义:
class CurrencyConverterInput(BaseModel):
"""货币转换工具的输入模式"""
amount: float = Field(…, description="要转换的金额")
from_currency: str = Field(…, description="源货币代码(例如:'USD')")
to_currency: str = Field(…, description="目标货币代码(例如:'EUR')")
契约定义好后,我们便可以在任务(Task)中指示智能体(Agent)必须遵循它。这通过 Task
对象的 output_pydantic
参数来实现。
# 这个任务的唯一目标就是将用户的自然语言请求结构化
query_task = Task(
description="理解用户的自然语言查询,提取总金额、源货币和目标货币。用户查询为:'{query}'",
expected_output="对用户查询的结构化响应",
agent=query_analyst,
output_pydantic=CurrencyConverterInput,
)
当这个任务运行时,CrewAI 会指示 LLM 的最终回复不能是一个句子,而必须是一个严格遵循 CurrencyConverterInput
模型的 JSON 对象。一个像"100 美元等于多少人民币?“这样的模糊查询,就被从凌乱的文本转换为了干净、可供机器读取的数据:{'amount': 100.0, 'from_currency': 'USD', 'to_currency': 'CNY'}
。
1.6 构建自定义工具
智能体是强大的思考者,但它们被禁锢在自己的数字世界里。要执行有意义的行动——比如查询数据库、调用 API 或保存文件——它们需要工具。虽然 CrewAI 提供了许多内置工具,但真正的威力在于创建属于你自己的工具。
一个自定义工具,本质上是一个被精心包装的 Python 函数。它拥有明确的 name
(名称)、description
(描述)和输入参数模式(args_schema
),这使得智能体能够理解这个工具能做什么以及如何使用它。
延续上面的例子,让我们构建一个工具,它能接收上一步生成的结构化数据,并调用外部 API 来执行真实的货币转换。
class CurrencyConverterTool(BaseTool):
"""货币转换工具类 - 使用实时汇率进行货币转换"""
name: str = "货币转换工具"
description: str = "使用实时汇率将一种货币的金额转换为另一种货币"
args_schema: Type[BaseModel] = CurrencyConverterInput # 参数格式
api_key: str = os.getenv("EXCHANGE_RATE_API_KEY")
# 工具被调用时执行的函数
def _run(self, amount: float, from_currency: str, to_currency: str) -> str:
"""执行货币转换的核心方法"""
pass
args_schema
:我们将工具直接与CurrencyConverterInput
这个 Pydantic 模型关联。这确保了工具只会接收到它期望格式的数据。name
&description
:这不仅仅是注释。智能体背后的 LLM 会阅读它们,来判断这个工具是否适合用来完成当前的任务。_run
:这是包含你自定义业务逻辑的核心工作方法。
通过结合结构化输出和自定义工具,我们创造了一个强大而可靠的多智能体工作流,其能力远超单个 LLM 提示所能达到的高度。当用户提出查询:”100 美元等于多少人民币?"
- 智能体 1 (查询分析师): 它的任务是理解和结构化。它接收到查询后,由于其任务指定了
output_pydantic=CurrencyConverterInput
,它不会尝试直接回答问题。相反,它输出一个干净的结构化 Pydantic 对象:{'amount': 100.0, 'from_currency': 'USD', 'to_currency': 'CNY'}
。这个输出会作为上下文传递给下一个任务。 - 智能体 2 (货币分析师): 它的任务是行动和执行。它接收到来自智能体 1 的结构化数据。它审视自己的目标和可用的工具,发现了"实时货币转换器",通过阅读描述,它明白这正是完成任务所需的工具。由于传入的数据与工具的
args_schema
完美匹配,它便能可靠地调用_run
方法来获取实时转换结果。
2 CrewAI Flows
在构建 AI 智能体系统时,简单的顺序执行已无法满足复杂应用的需求。本文将深入探讨 CrewAI 的强大功能——Flows,它为构建事件驱动、状态管理和条件分支的复杂 AI 工作流提供了优雅的解决方案。Flows 的核心理念在于,将传统软件开发的确定性逻辑与大语言模型(LLM)的自主推理能力巧妙融合,从而创造出既可靠又智能的全新系统。
2.1 基础构建
Flows 的构建基于两个核心的装饰器,它们像乐高积木一样,是我们搭建工作流的基础:
@start()
: 标记一个方法的"启动"状态。它告诉 Flow:“这是我们旅程的起点。” 当整个流程启动时,所有被@start
标记的方法会最先被执行。@listen(task_name)
: 建立方法之间的依赖关系。它让一个方法"聆听"另一个方法的完成信号。一旦被监听的方法执行完毕,这个"聆听者"就会被自动触发,并能直接接收前者的输出作为自己的输入。
class MovieRecommendationFlow(Flow):
@start
def generate_movie_genre(self):
...
return genre # 返回电影类型
@listen(generate_movie_genre)
def recommend_specific_movie(self, genre):
# 'genre' 参数自动接收了上一步的返回值 "科幻"
# … 执行电影推荐的逻辑 …
通过这种方式,我们建立了一个清晰的、事件驱动的依赖链:generate_movie_genre
完成 -> recommend_specific_movie
自动开始。
2.2 状态管理
在复杂的流程中,信息需要在不同步骤间传递和更新。比如,用户的 ID、偏好设置、或者中间计算结果。Flows 通过 self.state
属性提供了强大的状态管理机制,它就像是整个流程共享的"记事本"。
1. 非结构化状态 (Unstructured State)
最简单直接的方式,就是把 self.state
当作一个普通的 Python 字典来用。它灵活、便捷,非常适合快速原型设计。
@start def initialize_flow(self):
self.state['user_id'] = 12345 # 存入用户ID
user = self.state['user_id'] # 在另一个步骤中轻松读取
# 可以做到任意读写
2. 结构化状态 (Structured State)
当流程变得复杂,随意读写字典容易出错。为了保证数据的类型安全和一致性,我们可以为"记事本"规定一个格式。Flows 通过与 Pydantic 模型集成,完美实现了这一点。
我们首先定义一个 Pydantic 模型作为状态的"模板",然后在定义 Flow 类时声明它。就像给我们的记事本加上了固定的栏目(如"用户 ID"、“电影类型”、“推荐列表”),并规定了每一栏只能填写特定类型的数据(数字、文本等),从而从根本上杜绝了运行时因数据格式错误引发的问题。
class TaskState(BaseModel):
task: str
status: str
class MovieRecommendationFlow(Flow[TaskState]):
# self.state 里面只有两个变量,且类型为 str
2.3 条件流控制
现实世界的流程充满了选择。Flows 提供了强大的条件逻辑工具,让我们的 AI 工作流能够像经验丰富的决策者一样,根据实时结果动态调整执行路径。
1. or_
和 and_
: 处理多个依赖
- or_
: 当一个任务需要等待多个前置任务中的任意一个完成时,使用 or_
。
场景: 无论是用户通过在线聊天提出推荐请求,还是通过邮件系统,我们都需要记录这次请求。@listen(or_(live_chat_request, email_ticket_request))
就能完美实现这一点。
- and_
: 当一个任务必须等待多个前置任务全部完成时,使用 and_
。
场景: 只有当 AI “分析完用户历史偏好“并且”获取了最新的电影库“后,才能开始最终的"个性化推荐"任务。
2. @router
: 智能路由 @router
装饰器是实现复杂分支逻辑的利器。它可以将一个方法的返回值作为"路标”,动态地决定接下来应该走向哪条路径(执行哪个任务)。
class TicketRoutingFlow(Flow[TicketState]):
@start
def classify_ticket(self):
self.state.priority = random.choice(["high", "low"])
return self.state.priority
@router(classify_ticket)
def route_ticket(self, priority):
if priority == "high":
return "urgent_support" # 返回路由值
else:
return "email_support" # 返回路由值
@listen("urgent_support")
def assign_to_agent(self):
print("Ticket assigned to a live agent.")
@listen("email_support")
def send_to_email_queue(self):
print("Ticket sent to the email support queue.")
2.4 在 Flows 中集成 Crews
Flows 的真正威力在于它能够编排由多个智能体组成的、功能强大的 Crews。你可以将一个复杂的 Crew(比如一个专门负责撰写影评的团队)看作是 Flow 中的一个独立、可调用的"超级任务”。
通过 crewai new flow <flow_name>
命令,我们可以快速生成一个标准化的项目结构。在这个结构中,每个 Crew 都有自己独立的配置文件和脚本,而主流程文件 main.py
则负责扮演"总指挥"的角色。
在 main.py
的 Flow 中,你可以像调用普通函数一样,实例化并启动在其他文件中定义的任何 Crew,将其作为一个步骤无缝地集成到更宏大的工作流中。这使得我们可以构建出层次分明、高度模块化且极其强大的 AI 系统。
2.5 社交媒体内容撰写 FLow
随着 AI Agentic 系统的不断发展,单一智能体或单一 Crew 已无法满足复杂实际业务需求。CrewAI Flows 通过支持多 Crew 协作,极大提升了 AI 工作流的灵活性、可扩展性与模块化。
3 CrewAI 健壮性
当我们的 AI 智能体从执行简单的单步任务,走向处理复杂、多阶段的真实世界问题时,健壮性就从一个"加分项"变成了"必需品"。一个真正可用的 AI 系统,不仅要聪明,更要稳定、可靠、可控。如果智能体的输出像脱缰的野马,或者流程中的一个环节出错就导致整个系统崩溃,那么它离生产环境就还有很远的距离。
3.1 Guardrails(护栏)
大语言模型(LLM)的创造力是其核心优势,但有时也会导致"过度发挥",比如产生不符合格式要求的内容,或者陷入内容幻觉。Guardrails(护栏) 机制就像是为智能体的行为设定了明确的规则和边界,确保其输出始终在我们期望的轨道上。
在我们的博客写作流程中,第一步是让一个"研究员"智能体总结最新的行业动态。我们要求摘要必须精炼,不能超过 150 个词。如果不对其加以约束,它可能会生成一篇冗长的报告。这时,护栏就派上了用场。
我们可以定义一个简单的校验函数,并在任务中引用它:
def validate_summary_length(task_output) -> Tuple[bool, Any]:
if len(str(task_output).split()) > 150:
return False, "summary_length = xxx,should < 150"
return True, task_output
summary_task = Task(
description="你需要总结一篇关于卷积神经网络的研究论文,控制在200词以内。",
expected_output="一个精炼、准确的200词以内的研究论文摘要。",
agent=summary_agent,
guardrail=validate_summary_length, # 添加输出校验机制
max_retries=3 # 最大重试次数
)
当 summary_agent
完成任务时,validate_summary_length
函数会自动被调用。如果摘要过长,护栏会"拦截"这次输出,并将错误信息反馈给智能体。智能体在收到"摘要过长"的指令后,会理解问题所在并尝试重新生成一个更短的版本。这个自我修正的过程最多会进行 2 次,极大地提高了输出的可靠性。
3.2 上下文传递(Context)
一个好的博客写作流程,不是孤立任务的堆砌,而是一个连贯的创作过程。研究员的发现需要被分析师理解,分析师的洞察又要成为写作者的素材。上下文传递(Context Passing) 机制确保了信息在任务间的无缝流动,让后一步的智能体能够"站在前人的肩膀上"工作。
默认情况下,CrewAI 的顺序流程会自动将前一个任务的输出作为后一个任务的上下文。但当一个任务需要综合多个前置任务的结果时,我们就需要明确指定 context
。
我们的流程包含三个核心任务:
research_task
: 研究行业趋势。analysis_task
: 从研究结果中提炼关键洞察。blog_writing_task
: 结合研究和分析,撰写最终的博客文章。
最后的写作任务,显然需要同时参考研究和分析的结果。
# 定义任务:研究最新 AI 进展
research_task = Task(
...
)
# 定义任务:分析研究成果并提取关键点
analysis_task = Task(
...
)
# 写作任务需要前两个任务的上下文
blog_writing_task = Task(
description="撰写一篇详细的 AI 趋势博客文章",
expected_output="一篇结构清晰、内容详实的博客文章",
agent=writer_agent,
context=[research_task, analysis_task] # 引用前面的研究与分析结果
)
通过 context=[research_task, analysis_task]
,我们告诉 writer_agent
:“在开始写作前,请务必阅读并理解这两份材料。” 这使得信息流不再是简单的线性传递,而形成了一个汇合点,确保了最终产出的完整性和深度。
3.3 异步(Asynchronous)
在我们的博客写作流程中,“研究技术趋势"和"分析竞争对手动态"这两个任务之间并没有直接的依赖关系,它们完全可以同时进行。如果让它们按顺序执行,无疑会浪费宝贵的时间。异步任务执行(Asynchronous Task Execution) 解决了这个问题,它允许无依赖的任务并行处理,从而显著提升整体效率。
我们可以让两个不同的研究员智能体同时开始工作,一个研究 AI 技术,另一个研究市场法规。
# 任务1:研究AI技术(异步)
research_ai_task = Task(
...
async_execution=True # 异步执行,提高效率
)
# 任务2:研究市场法规(异步)
research_regulation_task = Task(
...
async_execution=True # 异步执行,与 AI 研究并行处理
)
# 任务3:生成综合 AI 报告(依赖前两个任务的结果)
generate_report_task = Task(
...
context=[research_ai_task, research_regulation_task]
# 使用前两个任务的输出作为上下文,实现同步
)
只需在任务定义中加入 async_execution=True
,这两个研究任务就会被同时启动。而 generate_report_task
由于 context
中指定了对它们两者的依赖,会自动等待这两个并行任务全部完成后,才开始执行。这就像一个高效的项目经理,将能并行的工作分派出去,然后在汇合点收集所有成果,从而大大缩短了项目的总周期。
3.4 回调机制(Callbacks)
当一篇博客最终由 writer_agent
完成后,工作并没有结束。我们可能需要将文章自动保存到内容管理系统(CMS)、通过 Slack 通知编辑团队进行审核,或者记录一条完成日志。回调机制(Callbacks) 允许我们在任务成功完成后,自动触发一个预定义的函数,实现流程的自动化闭环。
# 定义一个回调函数,用于处理最终的博客文章
def save_and_notify(task_output):
...
# 在写作任务中设置回调
blog_writing_task = Task(
...,
agent=writer_agent,
callback=save_and_notify # 指定任务完成后的动作
)
现在,每当 blog_writing_task
成功执行完毕,save_and_notify
函数就会被自动调用,并接收到任务的输出结果。这使得我们的 AI 工作流不再是一个孤立的系统,而是能与外部世界(文件系统、API、消息队列等)进行交互的、真正自动化的解决方案。
3.5 分层智能体(Hierarchical)
在构建多智能体(Multi-Agent)系统时,我们常常会遇到一个挑战:如何从简单的线性任务链,进化到能够模拟真实世界复杂组织协作的模式?CrewAI 的 分层流程(Hierarchical Process) 正是为此而生。它引入了"管理者"角色,将传统的顺序执行转变为一个结构化的、由上至下的任务管理与协同体系。
默认情况下,CrewAI 的任务以 顺序流程(Sequential Process) 执行,即一个智能体完成任务后,接力棒式地传给下一个。这种模式适用于简单的、一步接一步的工作流。然而,现实世界的项目,如撰写一份市场研究报告,通常需要一个项目经理来:
- 分解目标:将"完成报告"这个大目标分解为市场分析、风险评估、竞品调研等子任务。
- 分配任务:根据团队成员的专长,将子任务分配给最合适的人。
- 监督与整合:跟进各部分进度,审核产出质量,并最终将所有部分整合成一份逻辑连贯的最终报告。
分层流程正是对这种真实组织结构的精妙模拟。它引入了 管理者智能体(Manager Agent),使其成为整个团队的"大脑"和"指挥官”,从而实现更高效、更智能的协作。
[!NOTE] 核心思想 分层流程通过引入管理者角色,将任务执行从"流水线"升级为"项目团队",极大提升了系统的结构化、可扩展性和最终产出的质量。
在分层工作流中,智能体被划分为两类:
- 管理者智能体 (Manager Agent):
- 职责:作为团队的领导者,负责理解全局目标,进行任务分解、规划与分配,并对下属智能体的产出进行审核与整合。
- 关键特征:必须具备委派任务的权限,即
allow_delegation=True
。 - 角色类比:项目经理、总指挥、报告总编。
- 专业智能体 (Specialized Agents):
- 职责:作为团队的执行专家,专注于完成其被分配到的特定任务,如数据分析、内容写作或代码实现。
- 关键特征:通常不具备再次委派任务的权限,即
allow_delegation=False
。 - 角色类比:市场分析师、技术文案、前端工程师。
工作流如下:
- 管理者接收到一个高阶任务(例如,“分析 AI 在教育行业的应用前景”)。
- 管理者进行思考和规划,将任务分解为:“市场现状研究”、“关键技术分析”、“未来趋势预测"等几个子步骤。
- 管理者依次将这些子任务委派给对应的专业智能体(如研究员、技术分析师)。
- 专业智能体执行任务并返回结果。
- 管理者审查返回的结果,如果不满意可能会要求返工,如果满意则接受。
- 最后,管理者将所有审核通过的部分整合起来,形成一份完整、高质量的最终报告。
要启用分层流程,你需要在创建 Crew
时进行明确配置。以下是两种主要的实现方式:
3.5.1 方式一:指定管理者大模型(manager_llm
)- 自动创建管理者
这是最直接的方式。你只需告诉 CrewAI 使用哪个大语言模型(LLM)来扮演管理者的角色,CrewAI 将在后台自动为你创建一个具备管理能力的智能体。对于需要复杂规划和协调的任务,推荐为管理者使用能力更强的大模型(如 GPT-4、Claude 3 Opus 等),以确保任务分解和决策的质量。
from crewai import Agent, Task, Crew, Process
from myllm import llm # 假设你已经配置好了你的大模型
# 1. 定义你的专业智能体(执行者)
researcher = Agent(
role='高级市场研究员',
goal='深入分析特定行业的市场趋势与数据',
backstory='你是一位经验丰富的数据分析专家,擅长从海量信息中挖掘核心洞见。',
llm=llm,
allow_delegation=False # 专业智能体不进行委派
)
writer = Agent(
role='科技内容作家',
goal='将复杂的技术和市场分析转化为引人入胜的内容',
backstory='你是一位才华横溢的作家,能够用清晰、有吸引力的语言讲述技术故事。',
llm=llm,
allow_delegation=False
)
# 2. 定义任务列表
tasks = [
# 在这里定义需要被管理者分配的一系列任务
]
# 3. 组建带有自动管理者的 Crew
project_crew = Crew(
agents=[researcher, writer],
tasks=tasks,
process=Process.hierarchical, # 关键:明确指定使用分层流程
manager_llm=llm, # 关键:为自动创建的管理者指定一个LLM
)
3.5.2 方式二:自定义管理者智能体(manager_agent
)- 精细化控制
如果你希望对管理者的角色、背景故事、甚至工具进行更精细的控制,可以先创建一个自定义的 Agent
,然后将其指定为 manager_agent
。
# (接上文的 researcher 和 writer 定义)
# 1. 创建一个自定义的管理者智能体
manager = Agent(
role="项目研究经理",
goal="高效统筹整个项目研究流程,确保按时高质量完成输出",
backstory="""你是一位经验丰富的项目经理,对市场、风险和财务有全面的理解。
你擅长分解复杂问题,并将任务精确地分配给团队成员,
最后由你亲自审核并整合所有内容,形成最终的决策报告。""",
llm=llm,
allow_delegation=True # 管理者必须允许委派
)
# 2. 组建使用自定义管理者的 Crew
project_crew = Crew(
agents=[researcher, writer], # 执行者列表
tasks=tasks,
process=Process.hierarchical, # 指定分层流程
manager_agent=manager, # 关键:使用你自定义的管理者智能体
)
3.6 人类参与(Human-in-Loop)
尽管完全自动化的智能体系统展现了惊人的潜力,但在处理多步骤、高风险的复杂任务时,单一环节的失误就可能导致整个流程的失败。为了解决这一痛点,CrewAI 引入了 人类参与(Human-in-the-Loop, HITL) 机制,将人类的智慧与判断力无缝集成到自动化工作流中,从而极大地提升了系统的鲁棒性、可控性和可信度。
[!NOTE] 核心价值 HITL 机制允许我们在自动化流程的关键检查点(Checkpoints)暂停,由人类专家进行验证、修正或提供额外输入,确保每一步都走在正确的轨道上。
在 CrewAI 中实现 HITL 非常简单,只需在定义 Task
时,将 human_input
参数设置为 True
即可。智能体完成它的研究任务,并生成初步的总结。此时,CrewAI 的执行会暂停,并在终端(或你配置的任何输入/输出接口)中打印出研究员的产出,并提示用户输入。人类专家(用户)可以直接确认或者提供反馈或修正。
3.7 多模态(Multimodal)
现实世界的信息是丰富多彩的,远不止文本。为了让智能体能够理解和处理图像、音频等多种数据格式,CrewAI 内置了强大的 多模态能力。这使得 AI 的应用边界从纯文本分析,扩展到了产品质检、医学影像解读、设计稿审核等更广阔的领域。
在 CrewAI 中为智能体赋予视觉能力同样轻而易举。你只需在定义 Agent
时,将 multimodal
参数设置为 True
。CrewAI 会自动为该智能体配备处理图像的能力,通常是利用像 GPT-4o、GPT-4V 或 Gemini 等原生支持多模态的大模型。我这边用的 qwen-plus-latest
似乎不支持多模态功能,修改成了 qwen-vl-max-latest
也不行。
一旦智能体具备了多模态能力,你就可以在任务描述中引导它处理图像。通常,图像会通过上下文(Context) 或 工具(Tools) 传递给任务。
4 CrewAI 知识库
如果说工具(Tools) 赋予了智能体行动的能力,那么知识库(Knowledge Base) 则赋予了它们思考的深度。当智能体需要超越实时搜索,去理解和推理特定领域的复杂信息时,例如,消化一份数百页的财务报告、遵循内部编码规范或掌握产品的所有技术参数,知识库就成了其不可或缺的"第二大脑”。
4.1 工具 vs. 知识源
在 CrewAI 的设计哲学中,工具与知识源有着明确的界定,理解其差异是构建高效智能体的关键。
- 工具 (Tools):是主动的、动作导向的。它们是智能体用来与外部世界交互并执行任务的手段,例如使用搜索引擎查询实时新闻、调用 API 获取天气数据或在文件系统中保存文件。智能体需要有明确的意图去"调用"一个工具。
- 知识源 (Knowledge Sources):是被动的、认知驱动的。它们为智能体提供一个可供检索和推理的背景信息库。智能体不会主动"调用"知识源,而是在思考和形成答案时,隐式地从中汲取信息,如同人类专家在决策时会参考自己掌握的专业知识一样。
[!NOTE] 设计原则 “要让智能体做事,用工具;要让智能体知道什么,用知识源。”
4.2 字符串知识源
对于小规模、非结构化的知识,如一段公司政策、项目简介或常见问题解答(FAQ),使用字符串知识源 (StringKnowledgeSource
) 是最快捷的方式。它允许你直接将文本内容灌输给智能体,非常适合快速原型设计和测试。
from crewai.knowledge import StringKnowledgeSource
# 定义一段结构化的文本作为知识
company_policy_text = """
我们的休假政策规定,员工每年享有15天带薪年假。
新入职员工第一年按比例计算。
所有休假申请需提前两周通过内部HR系统提交。
"""
# 将其封装成一个知识源对象
policy_knowledge_source = StringKnowledgeSource(content=company_policy_text)
4.3 知识源的作用范围
信息并非总是需要对所有人可见。如同在真实团队中存在信息隔离一样,CrewAI 允许你精细地控制知识的访问权限:
- Agent 级别:将知识源绑定到特定的智能体。这适用于需要专业知识的场景,例如,只有"财务分析师"智能体才能访问机密的财务报表知识库。
- Crew 级别:将知识源赋予整个团队。这适用于需要共享通用知识的场景,例如,公司的产品手册、市场定位、品牌指南等,需要销售、市场和客服团队的所有成员都能访问。
[!TIP] 设计建议 模拟真实世界的团队协作模式。如果信息是需要跨部门共享的"公共知识",则设为 Crew 级别;如果是某个岗位的"专业技能包",则设为 Agent 级别。
4.4 无限扩展:构建自定义知识源
当你的数据存储在专有数据库、内部 API 或其他非标准系统中时,自定义知识源就显示出其强大的威力。通过继承 CrewAI 的 BaseKnowledgeSource
基类,你可以编写自己的逻辑,将任何数据源无缝接入智能体的知识体系。
这为你打开了无限可能:让智能体连接到实时销售数据库来回答业绩问题,或者接入内部 Jira 系统来报告项目进度。你只需重写 load_content()
方法,负责抓取数据并将其转换为文本格式,CrewAI 会自动处理后续的分块、嵌入和检索。
4.5 智能消化:分块策略(Chunking)
对于大型文档(如一本完整的书籍或一份长篇报告),直接将其作为一个整体提供给模型是低效且不可靠的。LLM 的注意力是有限的,过长的文本会稀释关键信息。因此,分块 (Chunking) 是必不可少的预处理步骤。
CrewAI 会自动将大型文档切分成更小的、语义相关的片段。一个关键的最佳实践是设置块间重叠(Chunk Overlap),确保在切分点不会割裂完整的语义,从而提升检索的准确性。
4.6 隐私与性能:嵌入策略定制
嵌入 (Embedding) 是将文本知识转化为向量表示、使其可被机器检索的关键技术。CrewAI 提供了灵活的嵌入策略,让你能够在性能、成本和隐私之间找到最佳平衡。虽然 CrewAI 默认会使用与你配置的 LLM 相匹配的嵌入模型(例如,使用 OpenAI LLM 时,默认用其 text-embedding-ada-002
),但你完全可以指定任何你偏好的模型。例如,为了数据隐私和成本控制,你可以选择在本地通过 Ollama 运行一个开源嵌入模型,如 nomic-embed-text
。
- 全离线部署:在对数据安全要求极高的场景下,使用本地嵌入模型结合本地 LLM,实现完全的数据私有化。
- 成本/性能优化:针对不同类型的数据选择最合适的嵌入模型,或在成本敏感的场景下切换到更经济的嵌入服务。
下面的例子展示了如何组建一个 Crew,为其提供知识源,并指定使用 Google 的嵌入模型,而不是依赖默认选项。
hr_crew = Crew(
agents=[hr_info_agent],
tasks=[org_inquiry_task],
process=Process.sequential,
knowledge_sources=[org_knowledge_source],
embedder={
"provider": "google",
"config": {
"model": "text-embedding-004",
"api_key": os.getenv("GOOGLE_API_KEY")
}
},
verbose=True
)
5 CrewAI 记忆
在之前的章节中,我们构建的智能体(Agent)虽然能力强大,但它们都有一个共同的"健忘症"。每一次与它们交互,都像是在和一位没有记忆的陌生人对话。你刚刚告诉过它的信息,转头它便忘得一干二净。这种"无状态"(stateless)的特性,是阻碍 AI 从一个简单的"执行器"进化为真正"智能伙伴"的最大障碍。
本节,我们将聚焦于为智能体植入"记忆"(Memory),探索如何让它们拥有保持上下文、跨会话学习和个性化交互的能力,从而变得真正"有智慧"。
5.1 记忆的核心价值
想象一下,如果你的私人助理每天都认不出你,每次都需要你重新介绍自己和你的偏好,这将是多么低效和令人沮丧的体验。无记忆的 AI 智能体正是如此。引入内存机制,将赋予它们四种核心的、改变游戏规则的能力:
- 上下文保持:在多轮对话或一个复杂的多步任务中,智能体能够记住前文,理解对话的来龙去脉,让交流如丝般顺滑。
- 个性化服务:通过记住你的偏好、历史选择,智能体能提供为你量身定制的建议和服务,而不是千篇一律的通用回答。
- 持续学习:智能体可以在一次次交互中积累经验,从错误中学习,不断优化未来的决策,变得越来越"懂你"。
- 高效协作:在团队中,一个智能体可以利用另一个智能体先前储存在共享记忆中的发现,避免重复劳动,实现 1+1>2 的效果。
5.2 记忆系统初探
我们创建一个标准的 Crew,然后告诉它一个信息:
用户:“我最喜欢的颜色是蓝色。”
接着,在同一次对话中,我们向它提问:
用户:“我最喜欢的颜色是什么?” 智能体:“抱歉,我不知道您最喜欢的颜色是什么。”
它忘记了。这正是"无状态"的表现。
现在,我们在创建 Crew
时,只需加入一个简单的参数 memory=True
,就像按下一个神奇的开关。
crew = Crew(
agents=[assistant],
tasks=[task],
process=Process.sequential,
memory=True, # 开启记忆
embedder=embedder,
verbose=True
)
user_input = "我最喜欢的颜色是蓝色,我最爱的框架是CrewAI。"
result = crew.kickoff(inputs={"user_task": user_input})
user_input = "我最喜欢的颜色是什么?"
result = crew.kickoff(inputs={"user_task": user_input})
5.3 CrewAI 记忆工作原理
当你启用记忆时,CrewAI 并没有简单地创建一个记录本,而是自动初始化并管理了一套复杂且分工明确的三层记忆系统。当智能体执行任务时,它会从这三层记忆中检索与当前问题最相关的信息,并将其注入到最终的提示(Prompt)中,让大语言模型(LLM)在做决策时能够"看到"这些关键的记忆片段。
5.3.1 短期记忆 (Short-Term Memory, STM)
- 核心作用:作为智能体的"工作记忆",负责存储当前会话或任务序列中的近期交互,确保对话的连贯性。
- 内部机制:它基于检索增强生成 (RAG) 技术。简单来说,就是将近期的对话内容转化为一种数学表达(向量),并存入一个专门的数据库。当有新问题时,它会快速找出语义上最接近的几条历史对话,作为参考。
- 适用场景:所有需要连贯对话的场景,例如客服问答、多步骤指令执行等。
5.3.2 长期记忆 (Long-Term Memory, LTM)
- 核心作用:实现跨会话的持久化记忆,用于积累高层次的见解、经验和学习成果。它记录的不是"说了什么",而是"学到了什么"。
- 内部机制:它更关注任务的整体描述(task.description),并检索历史上与当前任务宏观目标相似的成功或失败经验,为决策提供战略性参考。
- 适用场景:需要智能体从过去的成败中学习、避免重复犯错、随时间积累领域知识的复杂应用。
5.3.3 实体记忆 (Entity Memory)
- 核心作用:专门捕获和组织关于特定实体(如人物、组织、产品)的信息,为每个实体建立一个微型知识图谱。
- 内部机制:它同样基于 RAG,但将信息与检测到的实体名称进行了绑定。这使得智能体在处理多个实体时,能够精确地检索到"关于 A 公司的信息",而不会与"B 公司的信息"相混淆。
- 适用场景:研究助理需要收集多家公司信息、客服智能体需同时处理多个客户、故事创作需保持角色设定一致性等。
5.3.4 上下文记忆 (Contextual Memory)
- 核心作用:它并非一种独立的存储类型,而是一个智能检索层。它的职责是在智能体执行任务前,智能地融合短期、长期和实体记忆中的信息。
- 内部机制:它会同时向上述三种记忆发出查询,然后将检索到的、按相关性排序后的信息片段整合起来,共同注入到 LLM 的提示(Prompt)中。它确保了智能体总能基于最全面的上下文进行思考。
5.3.5 用户记忆 (User Memory)
- 核心作用:实现深度个性化,将所有记忆(短期、长期、实体)的范围都限定在单个用户。
- 内部机制:通过将所有记忆与一个唯一的
user_id
关联来实现。这可以确保张三的记忆和李四的记忆完全隔离,既保护了数据隐私,也使智能体能为不同用户提供量身定制的交互体验。 - 适用场景:所有面向多用户的生产级应用,如个人助理、定制化推荐系统、专属客服等。这是从"通用 AI"迈向"专属 AI"的关键一步。
5.4 重置记忆
在开发、测试或需要清除过时/敏感信息的场景下,对内存进行重置至关重要。CrewAI 提供了简单直接的方法。
- 方法:调用
crew.reset_memories()
方法。 - 精确控制:你可以选择性地清空特定类型的记忆,例如,只清空当前会话的短期记忆,同时保留宝贵的长期经验。
# 清空所有记忆
crew.reset_memories(memory_type='all')
# 只清空短期记忆
crew.reset_memories(memory_type='short')
通过这套精妙的多层次记忆系统,CrewAI 的智能体不再是一个被动的指令执行者。它拥有了回顾过去(短期记忆)、总结经验(长期记忆)和理解世界(实体记忆)的能力,并能通过用户记忆实现真正的个性化,向着"智能伙伴"迈出了坚实的一步。
[!NOTE] CrewAI 默认的记忆功能全部使用 RAG 来实现,因此响应速度会有一些影响。
6 Agent 设计范式
6.1 ReAct 模式
构建高级 AI 智能体(Agent)时,经常会遇到超出大语言模型(LLM)本身知识或能力范围的问题,如需要实时检索、复杂计算或 API 交互。当 AI 单靠内部参数无法解答时,传统一次性生成答案的方法力有未逮。ReAct 模式(Reasoning + Acting)应运而生,为 LLM 打造了结构化的问题求解框架。ReAct 模式将智能体决策流程分为三个可循环迭代的阶段:
- 思辨 (Thought):模型基于已有上下文和已获得信息,生成内部的中间推理。这是真实的"思考过程",帮模型分解任务和规划行动。
- 行动 (Action):根据推理结果,模型输出一个结构化的指令 (如调用某个工具或 API),以获取它所需的外部信息或完成某项动作。通常格式规范,便于机器解析。
- 观察 (Observation):外部系统自动捕获行动的结果,并以"观察"形式反馈给 LLM。LLM 据此调整思路,进入新一轮思辨。
该循环精准复刻了人类解决复杂问题时的"想→做→看→再想"的认知流程。一次只做一步,直至模型有把握输出最终答案。
system_prompt = """
你在一个循环中运行,每步只做一件事:
1. Thought(思考):描述你对输入问题的想法。
2. PAUSE(暂停):思考要采取的行动。
3. Action(行动):明确行动及参数。
4. PAUSE(暂停):等待行动结果。
5. Observation(观察):接收执行反馈。
可用操作:
**math(数学计算)**
**lookup_population(查找人口)**
循环直至有足够信息输出答案。
"""
这一机制广泛用于处理开放域问答、数据查找、动态决策等场景,兼具解释性、透明性和可调试性。
6.2 Planning 模式
ReAct 模式强调"边想边试",极适合即时反馈和高不确定性任务,但在结构化、多步骤、复合目标的复杂任务上,规划模式(Planning Pattern) 更具优势。其技术要点在于,将长期目标分解为可控的子任务,形成计划后再有序执行,显著提高效率、全局性和可解释性。规划模式结构分明,分为如下阶段:
- 规划(Planning Phase):
- 智能体首先分析整体任务,细致地将问题拆解为有序子步骤,生成详细计划(Plan)。
- 计划通常为列表或结构化数据(如 JSON),有助于后续自动执行和追踪。
- 执行(Execution Phase):
- 智能体依次执行计划中的每一步,可调用算数、检索、API 等外部工具。每一步执行类似 ReAct 中的"行动 - 观察"模式。
- 结果汇总与聚合(Aggregation & Response):
- 完成所有步骤后,集中整理各步观察,融合为综合性输出——既可返还用户,也便于系统自我评估。
- 动态调整与再规划(Re-planning, 可选):
- 若中间步骤遇到异常,智能体可主动重返"规划阶段",根据环境与需求弹性调整后续步骤。
你是一个智能的规划代理(Planning Agent)。
每次迭代只执行一项:
1) PLAN(计划)——罗列回答问题必需的步骤。
2) Execute(执行)——按顺序执行计划步骤,仅执行一个。
3) Observation(观察)——获取执行输出。
4) Collect(汇总)——收集每步结果。
5) Answer(回答)——基于汇总信息输出解答。
只能先输出PLAN,再逐步执行,最终给出FINAL ANSWER。
可用工具:
**math(数学计算)**
**lookup_population(查找人口)**
规划模式方法论得到了丰富研究与前沿技术的支持,包括:
- Plan-and-Solve Prompting:引导 LLM 先列计划后求解,降低遗漏环节风险,提升复杂推理准确率。
- Tree-of-Thought (ToT):非线性规划,允许智能体并行探索多条任务路径,利用分支搜寻最优解,极大提升效果。
- Reflexion:系统引入自省、自我复查机制,执行后反思潜在错误,自动调整未来计划,真正实现持续自适应优化。
- OpenAI 函数调用 (Function calling):为结构化工具调用、计划分解与复合执行提供基础设施支撑,提高成功率和解析安全性。
综上,规划模式极适合多步骤数据分析、流程拆解、模板化业务执行等场景,补足了 ReAct 模式在全局把控与高效协作上的短板。
6.3 Multi-Agent 模式
当单个智能体难以胜任复杂任务时,将多位专业化 AI 组建为协作团队(Multi-Agent System)成为自然选择。多智能体模式不仅模拟了人类组织分工,还极大提升了大规模问题求解的灵活性、可维护性与透明性。多智能体具有如下优势:
- 模块化与可维护性:任务按子模块分解,每个智能体聚焦自身领域,系统变为易于开发、升级与调试的"组件拼装"。
- 专业化与高效性:分别为每个智能体量身定制角色、背景知识、工具集,使其对口子任务游刃有余,效率与质量双重提升。
- 过程透明与可解释性:每位智能体的决策与中间产出均可追踪审计,利于用户信任和系统自查。
- 鲁棒与错误隔离:问题只影响相应子任务或个体,易于定位与修正,提升整体鲁棒性。
- 易于系统化扩展:完全沿用人类"流水线"协作直觉,可并行、串行甚至分叉建立复杂任务流。
多智能体系统由以下关键组件组成:
- 智能体(Agent)
由 LLM 驱动、具有独立 Prompt、角色设定及专业工具权限的自主单元。它们结合 ReAct 或 Planning 模式,高度角色化、独立作业。 - 工具(Tool)
为智能体提供计算、检索、外部服务调用等接口。不同智能体获得专属的"工具包",高内聚、低耦合。 - 编排与协作(Crew/Orchestrator)
整体系统的流程控制器,定义智能体协作关系(如串行、并行、分支决策),协调任务流与上下文信息传递,保障团队高效协作。
多智能体通常采用智能体流水线(Pipeline),信息与中间结果环环相扣,逐步完善目标。可进一步引入异步任务、分工复核、“智能体对话"等协同策略,使团队智能体互补其长,动态调整方案。
综上,ReAct 模式擅长即时反馈、探索式任务,规划模式优化全局分解与结构化执行,多智能体模式则将"分工协作的专业团队"理念推向极致。三者往往可组合创新,形成性能强大、易于扩展和维护的 AI 复杂系统。面对千变万化的现实应用需求,灵活运用这些范式,将是每一位 AI 系统设计师的关键能力。