Multi AI Agent Systems with CrewAI 学习笔记

> 课程来源：DeepLearning.AI Short Course > 课程链接：https://www.deeplearning.ai/short-courses/multi-ai-agent-systems-with-crewai/ > 讲师：João Moura（CrewAI 创始人兼 CEO） > 时长：约 2 小时 41 分钟 > 课程节数：18 节视频 + 7 个代码示例 > 学习时间：2026-04-12

---

一、课程概述

1.1 课程目标

本课程由 CrewAI 创始人 João Moura 亲自主讲，核心目标是：

- 超越单一 LLM 性能：通过设计多个 AI Agent 团队进行协作，解决复杂多步骤任务 - 自动化业务流程：使用开源框架 CrewAI 实现简历定制、活动策划、内容创作等常见业务流程自动化 - 构建智能 Agent 团队：通过定义 Agent 的角色（Role）、目标（Goal）和背景故事（Backstory），将复杂任务分配给专门的 Agent

1.2 课程适用人群

- 已学习过 Prompt Engineering 课程 - 具备基础编程能力 - 希望将 LLM 融入专业工作流程的开发者

1.3 课程大纲

模块	内容
AI Agent 基础	Agent 定义、角色扮演、工具集成
多 Agent 协作	Sequential、Hierarchical、Consensual 三种协作模式
任务定义	如何设计清晰、可执行的任务
实战项目	简历定制、内容创作、客户支持、活动策划、财务分析

1.4 升级课程推荐

DeepLearning.AI 还提供了进阶课程 《Design, Develop, and Deploy Multi-Agent Systems with CrewAI》：

- 4 个模块，约 20 小时 - 涵盖：ReAct 循环、记忆系统、MCP 协议、Guardrails、生产级部署 - 包含代码审查器、会议助手、深度研究员等实战项目

---

二、核心概念解析

2.1 Agent（智能体）

是什么： Agent 是 CrewAI 的核心执行单元，相当于"团队中的具体角色"。每个 Agent 有明确的职责和能力边界。 为什么重要： - Agent 是多 Agent 系统的基础构建块 - 通过角色定义，LLM 可以生成更贴合场景的输出 - Agent 可以自主决策、协作、委派任务 怎么用：

from crewai import Agent
from crewai_tools import SerperDevTool
创建研究员 Agent
researcher = Agent(
    role="资深AI情报研究员",           # 角色：定义身份
    goal="获取最新AI技术动态",          # 目标：Agent 追求的方向
    backstory="""                   # 背景故事：塑造 Agent 行为
        你精通情报搜集，能从官网、X平台、新闻中提炼可靠信息。
        你总是引用信息来源，确保数据可追溯。
    """,
    tools=[SerperDevTool()],        # 工具：扩展 Agent 能力
    verbose=True,
    allow_delegation=False           # 是否允许委派任务
)

Agent 的关键属性：

属性	说明	重要性
`role`	Agent 的身份角色	⭐⭐⭐⭐⭐
`goal`	Agent 追求的目标	⭐⭐⭐⭐⭐
`backstory`	背景故事，塑造行为	⭐⭐⭐⭐
`tools`	工具集（搜索、爬虫等）	⭐⭐⭐⭐
`llm`	使用的语言模型	⭐⭐⭐⭐
`memory`	是否启用记忆	⭐⭐⭐
`allow_delegation`	是否允许委派	⭐⭐⭐

2.2 Task（任务）

是什么： Task 是分配给 Agent 的具体工作，包含任务描述和预期输出。 为什么重要： - 任务定义了"做什么"和"做到什么程度" - 清晰的任务定义是多 Agent 协作的基础 - 任务之间可以通过 context 参数共享结果 怎么用：

from crewai import Task
research_task = Task(
    description="""
        研究 {topic} 领域的最新技术动态。
        需要找到：
        1. 主要参与者和贡献
        2. 技术创新和突破
        3. 挑战和限制
        4. 未来方向
    """,
    expected_output="""
        一个包含 10 个要点的列表，覆盖 {topic} 最重要的发现，
        强调与开发者相关的技术细节。
    """,
    agent=researcher,                # 指定执行者
    context=[previous_task]          # 可选：依赖的前置任务
)
writing_task = Task(
    description="基于研究报告撰写博客",
    expected_output="1000字博客草稿（Markdown格式）",
    agent=writer,
    context=[research_task]          # 接收研究报告的输出
)

2.3 Crew（团队）

是什么： Crew 是协调多个 Agent 协作的容器，负责管理任务分配和执行流程。 为什么重要： - Crew 是多 Agent 系统的"管理层" - 决定 Agent 之间的协作模式（顺序、层级、共识） - 管理任务流程和结果输出 怎么用：

from crewai import Crew, Process
crew = Crew(
    agents=[researcher, analyst, writer],  # 团队成员
    tasks=[research_task, analysis_task, writing_task],  # 任务列表
    process=Process.sequential,              # 协作模式
    verbose=True,
    memory=True                             # 启用记忆
)
result = crew.kickoff(inputs={"topic": "AI Agents"})

2.4 Tools（工具）

是什么： Tools 是扩展 Agent 能力的实用程序，如搜索引擎、数据库连接、API 调用等。 为什么重要： - Agent 本身只有语言理解能力，Tools 赋予它"行动能力" - 通过工具集成，Agent 可以访问实时信息 - 解决 LLM"知识滞后"的问题 怎么用：

from crewai_tools import SerperDevTool, ScrapeWebsiteTool, RagTool
内置工具
search_tool = SerperDevTool()
scrape_tool = ScrapeWebsiteTool()
rag_tool = RagTool()
自定义工具
from crewai.tools import BaseTool
from pydantic import Field
class MyCustomTool(BaseTool):
    name: str = "my_custom_tool"
    description: str = "描述工具用途，LLM 会根据此决定何时使用"
    
    def _run(self, input: str) -> str:
        # 工具执行逻辑
        return "工具执行结果"
custom_tool = MyCustomTool()

2.5 Memory（记忆系统）

是什么： Memory 是 CrewAI 的统一记忆系统，提供短期、长期和实体记忆能力。 为什么重要： - 让 Agent 能够在多次任务执行中学习和改进 - 保持对话上下文的一致性 - 支持跨 Crew 运行的知识积累 怎么用：

# 启用记忆（默认短期记忆）
crew = Crew(
    agents=[...],
    tasks=[...],
    memory=True  # 启用记忆功能
)
高级：使用统一 Memory API
from crewai import Memory
memory = Memory()
LLM 自动推断存储位置
memory.remember("我们选择了 PostgreSQL 作为用户数据库")
memory.remember("Sprint 速度是 42 点", scope="/team/metrics")
按范围检索
agent_memory = memory.scope("/agent/researcher")
matches = agent_memory.recall("相关论文")

2.6 Flow（流程）

是什么： Flow 是 CrewAI 的事件驱动工作流引擎，提供精确的状态管理和流程控制。 为什么重要： - 实现 Crews 和外部代码的深度集成 - 支持条件分支、循环、状态持久化 - 是生产级部署的推荐架构 怎么用：

from crewai.flow.flow import Flow, start, listen
from pydantic import BaseModel
class AppState(BaseModel):
    user_input: str = ""
    research_results: str = ""
    final_report: str = ""
class ProductionFlow(Flow[AppState]):
    @start()
    def gather_input(self):
        self.state.user_input = "AI Agents 最新动态"
    
    @listen(gather_input)
    def run_research(self):
        # 调用 Crew
        crew = ResearchCrew()
        result = crew.kickoff(inputs={"topic": self.state.user_input})
        self.state.research_results = result.raw
    
    @listen(run_research)
    def write_report(self):
        # 继续处理...
        pass
持久化
@persist
class PersistedFlow(Flow[AppState]):
    # 状态会自动保存到 SQLite
    pass

2.7 Guardrails（护栏）

是什么： Guardrails 用于验证任务输出，确保 Agent 产生的结果符合预期标准。 为什么重要： - 防止 Agent 输出不符合质量要求的内容 - 自动触发重试机制 - 减少幻觉（Hallucination） 怎么用：

from crewai import Task
def validate_word_count(output):
    """确保博客文章在 500-1000 字之间"""
    word_count = len(output.raw.split())
    if word_count < 500:
        return (False, "内容太短，请扩展到至少 500 字。")
    if word_count > 1000:
        return (False, "内容太长，请精简到 1000 字以内。")
    return (True, output.raw)
task = Task(
    description="撰写技术博客",
    expected_output="500-1000 字的技术博客",
    agent=writer,
    guardrail=validate_word_count  # 添加验证
)

---

三、协作模式对比

CrewAI 支持三种主要的 Agent 协作模式：

3.1 Sequential（顺序执行）

模式说明：任务按预定义的顺序逐一执行，一个任务完成后才开始下一个。

[Task 1] → [Task 2] → [Task 3] → [最终输出]

适用场景： - 流程确定、步骤简单的任务 - 任务之间有明确的依赖关系 - 如：研究 → 撰写 → 编辑 → 发布 代码示例：

from crewai import Crew, Process
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, write_task, edit_task],
    process=Process.sequential,  # 顺序执行
    verbose=True
)

3.2 Hierarchical（层级管理）

模式说明：引入一个 Manager Agent（项目经理），负责分解任务、动态分配、监控进度、整合输出。

        [Manager]
           ↓
    ┌──────┼──────┐
    ↓      ↓      ↓
[Task 1] [Task 2] [Task 3]
    ↓      ↓      ↓
    └──────┼──────┘
           ↓
     [最终输出]

适用场景： - 复杂、开放式、信息不确定的任务 - 需要动态调整流程的情况 - 需要质量把控和审核机制 代码示例：

from crewai import Crew, Process, Agent
from langchain_openai import ChatOpenAI
经理 LLM（建议使用更强的模型）
manager_llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
组建层级团队
crew = Crew(
    agents=[researcher, analyst, writer, reviewer],
    tasks=[final_task],  # 主任务交给 Manager
    process=Process.hierarchical,
    manager_llm=manager_llm,  # 指定经理 LLM
    verbose=True,
    memory=True
)

自定义 Manager Agent：

manager = Agent(
    role="AI项目经理",
    goal="高效协调团队，产出高质量报告",
    backstory="你是一位经验丰富的AI项目经理...",
    allow_delegation=True,  # 必须允许委派
    llm=manager_llm
)
crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.hierarchical,
    manager_agent=manager,  # 使用自定义经理
    verbose=True
)

3.3 Consensual（共识决策）

模式说明： Agent 之间通过民主协商达成共识后执行任务。状态：该模式正在开发中，尚未在正式版本中实现。

3.4 协作模式对比表

特性	Sequential	Hierarchical
执行方式	固定顺序	动态分配
灵活性	低	高
适用场景	流程确定	复杂开放
Manager 角色	无	有（自动或自定义）
令牌消耗	较低	较高（需要 Manager 协调）
质量控制	依赖任务定义	Manager 主动审核
典型用例	内容流水线	技术情报分析

---

四、关键代码示例

4.1 基础多 Agent 系统

from crewai import Agent, Task, Crew, Process
from crewai_tools import DuckDuckGoSearchTool
from langchain_openai import ChatOpenAI
1. 配置 LLM
def openai_llm(**kwargs):
    return ChatOpenAI(
        model="gpt-4o-mini",
        api_key="your-api-key"
    )
2. 定义工具
search_tool = DuckDuckGoSearchTool()
3. 创建 Agent
topic_agent = Agent(
    role="话题研究专家",
    goal="挖掘 {topic} 相关的热点话题",
    backstory="你是内容策略师，擅长识别趋势...",
    tools=[search_tool],
    llm=openai_llm(),
    verbose=True
)
writer_agent = Agent(
    role="科技博客撰稿人",
    goal="基于研究结果撰写博客",
    backstory="你是获奖科技撰稿人...",
    llm=openai_llm(),
    verbose=True
)
4. 创建 Task
research_task = Task(
    description="研究 {topic} 的热点",
    agent=topic_agent,
    expected_output="Markdown 格式的研究报告"
)
write_task = Task(
    description="基于研究撰写博客",
    agent=writer_agent,
    context=[research_task],  # 接收研究结果
    expected_output="1000 字博客草稿"
)
5. 组建 Crew
crew = Crew(
    agents=[topic_agent, writer_agent],
    tasks=[research_task, write_task],
    process=Process.sequential,
    verbose=True
)
6. 启动
result = crew.kickoff(inputs={"topic": "大模型在编程中的应用"})
print(result)

4.2 层级管理多 Agent 系统

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, ScrapeWebsiteTool
from langchain_openai import ChatOpenAI
import os
os.environ["OPENAI_API_KEY"] = "your-key"
os.environ["SERPER_API_KEY"] = "your-key"
经理和员工的 LLM
manager_llm = ChatOpenAI(model="gpt-4o", temperature=0.2)
worker_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.4)
工具
search_tool = SerperDevTool()
scrape_tool = ScrapeWebsiteTool()
定义团队
manager = Agent(
    role="AI项目经理",
    goal="高效协调团队，产出高质量报告",
    backstory="你是经验丰富的AI项目经理...",
    llm=manager_llm,
    allow_delegation=True,
    verbose=True
)
researcher = Agent(
    role="资深研究员",
    goal="获取最新AI技术动态",
    backstory="你精通情报搜集...",
    tools=[search_tool, scrape_tool],
    llm=worker_llm,
    allow_delegation=False,
    verbose=True
)
analyst = Agent(
    role="技术分析师",
    goal="深入分析技术原理",
    backstory="你对LLM架构有深刻理解...",
    tools=[scrape_tool],
    llm=worker_llm,
    allow_delegation=False,
    verbose=True
)
writer = Agent(
    role="高级技术作家",
    goal="撰写专业报告",
    backstory="你的文章曾刊登于AI媒体...",
    llm=worker_llm,
    allow_delegation=False,
    verbose=True
)
最终任务
final_task = Task(
    description="""
        产出一篇深度技术报告：
        - 基于最新真实信息
        - 涵盖技术原理、性能、应用
        - Markdown 格式，约1000字
    """,
    expected_output="完整的技术分析报告",
    agent=manager
)
组建层级团队
crew = Crew(
    agents=[manager, researcher, analyst, writer],
    tasks=[final_task],
    process=Process.hierarchical,
    manager_agent=manager,
    verbose=2
)
执行
result = crew.kickoff()
print(result)

4.3 带记忆的多 Agent 系统

# 启用记忆让 Agent 从历史执行中学习
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,
    memory=True,  # 启用记忆
    verbose=True
)
多次运行，Agent 会记住之前的经验
for i in range(5):
    result = crew.kickoff(inputs={"topic": f"topic_{i}"})
    print(f"Run {i+1} completed")

4.4 带 Guardrails 的质量控制

from crewai import Task
def validate_content(output):
    """验证内容质量"""
    word_count = len(output.raw.split())
    
    # 检查长度
    if word_count < 500:
        return (False, "内容太短，请扩展到至少 500 字。")
    
    # 检查是否包含关键词
    keywords = ["AI", "技术", "分析"]
    if not any(kw in output.raw for kw in keywords):
        return (False, "内容似乎偏离主题，请确保包含核心技术关键词。")
    
    return (True, output.raw)
task = Task(
    description="撰写技术分析报告",
    expected_output="包含核心技术分析的报告",
    agent=writer,
    guardrail=validate_content
)

4.5 Flow 状态管理

from crewai.flow.flow import Flow, start, listen
from pydantic import BaseModel
class ReportState(BaseModel):
    topic: str = ""
    research_results: str = ""
    analysis: str = ""
    final_report: str = ""
class ReportFlow(Flow[ReportState]):
    @start()
    def get_topic(self):
        self.state.topic = "AI Agents 2026"
    
    @listen(get_topic)
    def research(self):
        crew = ResearchCrew()
        result = crew.kickoff(inputs={"topic": self.state.topic})
        self.state.research_results = result.raw
    
    @listen(research)
    def analyze(self):
        crew = AnalysisCrew()
        result = crew.kickoff(inputs={"data": self.state.research_results})
        self.state.analysis = result.raw
    
    @listen(analyze)
    def write(self):
        crew = WritingCrew()
        result = crew.kickoff(inputs={"content": self.state.analysis})
        self.state.final_report = result.raw
运行 Flow
flow = ReportFlow()
result = flow.kickoff()
print(flow.state.final_report)

---

五、行为准则提炼

基于 CrewAI 课程学习和 Agent 实践，我总结了以下行为准则：

准则 1：清晰的角色定义

原则：每个 Agent 必须有明确的 role、goal 和 backstory，避免角色重叠或模糊。实践：

# ✅ 好的例子：具体、互补的角色
researcher = Agent(role="市场研究分析师", ...)
writer = Agent(role="技术内容作家", ...)
❌ 差的例子：重叠或模糊的角色
agent1 = Agent(role="通用助手", ...)
agent2 = Agent(role="帮助者", ...)

应用于看宝AI： - 明确区分"记忆管理Agent"、"对话理解Agent"、"任务执行Agent"等角色 - 每个 Agent 有清晰的职责边界

准则 2：任务描述必须具体可执行

原则：任务描述应包含具体的执行步骤、格式要求和质量标准。实践：

# ✅ 好的任务定义
Task(
    description="""
        研究 AI Agents 领域：
        1. 找到 2024-2026 年的主要发展
        2. 列出关键玩家及其贡献
        3. 识别 3 个技术突破点
        4. 指出主要挑战和限制
        
        格式：Markdown，包含引用来源
    """,
    expected_output="结构化的研究报告"
)
❌ 差的任务定义
Task(description="研究一下 AI Agents", ...)

应用于看宝AI： - 用户请求 → 任务分解 → Agent 执行，每步都要明确 - 避免"帮我处理一下"这类模糊指令

准则 3：善用上下文传递

原则：通过 context 参数让下游 Agent 获得上游任务的输出，避免重复工作。实践：

# 研究员输出作为分析师的输入
analysis_task = Task(
    description="分析研究结果",
    agent=analyst,
    context=[research_task]  # 接收研究员的输出
)
分析师输出作为撰写员的输入
writing_task = Task(
    description="撰写报告",
    agent=writer,
    context=[research_task, analysis_task]  # 接收多个任务的输出
)

应用于看宝AI： - 多步骤任务（如研究→分析→输出）必须传递上下文 - 每个 Agent 应该能看到相关的前置结果

准则 4：层级模式用于复杂任务

原则：对于开放式、复杂的任务，使用 Hierarchical 模式让 Manager Agent 动态协调。实践：

# 复杂情报分析任务 → 使用层级模式
crew = Crew(
    agents=[manager, researcher, analyst, writer, reviewer],
    tasks=[final_task],
    process=Process.hierarchical,
    manager_agent=manager,
    memory=True
)

应用于看宝AI： - 复杂用户请求（如"帮我分析这个市场"）→ 启用层级模式 - Manager Agent 负责任务分解和质量把控

准则 5：记忆系统实现持续学习

原则：为需要跨任务学习的 Agent 启用记忆，让系统能够从历史执行中改进。实践：

crew = Crew(
    agents=[...],
    tasks=[...],
    memory=True  # 启用记忆
)

应用于看宝AI： - 用户偏好、对话历史、项目信息 → 存入记忆 - Agent 下次执行时可以参考之前的经验

准则 6：Guardrails 保证输出质量

原则：为关键任务添加输出验证，防止低质量或错误输出进入下游。实践：

def validate_report(output):
    if len(output.raw.split()) < 500:
        return (False, "报告太短，请扩展内容")
    return (True, output.raw)
task = Task(
    description="撰写报告",
    agent=writer,
    guardrail=validate_report
)

应用于看宝AI： - 关键输出（发送给用户的回复、生成的文件）必须验证 - 防止幻觉、错误信息影响用户体验

准则 7：Flow 用于生产级应用

原则：生产环境中的应用应该使用 Flow 架构，实现状态管理和精确控制。实践：

class ProductionFlow(Flow[AppState]):
    @start()
    def handle_input(self):
        pass
    
    @listen(handle_input)
    def run_crew(self):
        pass
    
    @persist
    class ProductionFlow(...):
        pass  # 持久化状态

应用于看宝AI： - 看宝AI 作为生产系统 → 使用 Flow 管理状态 - 任务执行、用户交互、外部调用 → 通过 Flow 协调

---

六、应用到看宝AI的思考

6.1 看宝AI 的 Agent 架构设计

基于 CrewAI 的设计模式，看宝AI 可以构建如下 Agent 团队：

┌─────────────────────────────────────────────────────────────┐
│                      看宝AI Agent 团队                      │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│                      [用户请求]                              │
│                          ↓                                  │
│    ┌─────────────────────────────────────────────────┐      │
│    │              Manager Agent (协调者)               │      │
│    │  • 解析用户意图                                   │      │
│    │  • 分解复杂任务                                   │      │
│    │  • 分配子任务                                     │      │
│    │  • 整合最终输出                                   │      │
│    └─────────────────────────────────────────────────┘      │
│         ↓                    ↓                    ↓         │
│    ┌─────────┐        ┌─────────────┐        ┌─────────┐    │
│    │ 研究员  │        │   执行者    │        │  审核者  │    │
│    │ Agent   │        │   Agent    │        │  Agent  │    │
│    └─────────┘        └─────────────┘        └─────────┘    │
│         ↓                    ↓                    ↓         │
│    ┌─────────┐        ┌─────────────┐        ┌─────────┐    │
│    │ 记忆管理 │        │   工具调用  │        │  质量把控 │    │
│    │ Agent   │        │   (搜索等)   │        │         │    │
│    └─────────┘        └─────────────┘        └─────────┘    │
│                                                             │
└─────────────────────────────────────────────────────────────┘

6.2 看宝AI 的核心 Agent 定义

# 看宝AI Agent 团队设计
1. 用户意图理解 Agent
user_intent_agent = Agent(
    role="用户意图分析师",
    goal="准确理解用户的真实需求和期望输出格式",
    backstory="""
        你是一位经验丰富的需求分析师，擅长透过用户的表面描述
        看到他们真正的需求。你会考虑用户的背景、使用场景和
        期望结果。
    """,
    allow_delegation=False
)
2. 记忆管理 Agent
memory_agent = Agent(
    role="记忆管理员",
    goal="管理用户偏好、历史对话和项目上下文",
    backstory="""
        你是一位细心的管家，负责记住用户的偏好设置、
        之前的对话内容、项目进展等重要信息。
    """,
    allow_delegation=False,
    memory=True  # 启用记忆
)
3. 任务规划 Agent
planner_agent = Agent(
    role="任务规划师",
    goal="将复杂任务分解为可执行的子任务序列",
    backstory="""
        你是一位出色的项目经理，擅长将大型任务拆解为
        小步骤，合理安排执行顺序，预估所需资源。
    """,
    allow_delegation=True  # 可以分配任务
)
4. 研究员 Agent
researcher_agent = Agent(
    role="研究员",
    goal="收集相关信息、分析数据、提供洞见",
    backstory="""
        你是一位严谨的研究者，擅长从多种来源收集信息、
        分析数据、提炼关键发现，并引用可靠来源。
    """,
    tools=[search_tool, scrape_tool],
    allow_delegation=False
)
5. 内容创作 Agent
creator_agent = Agent(
    role="内容创作者",
    goal="基于研究结果创作高质量内容",
    backstory="""
        你是一位获奖的内容创作者，能够将复杂信息转化为
        清晰、吸引人的内容。
    """,
    allow_delegation=False
)
6. 审核 Agent
reviewer_agent = Agent(
    role="质量审核员",
    goal="确保输出内容符合质量标准",
    backstory="""
        你是一位严谨的编辑，对内容质量有极高的要求。
        你会检查事实准确性、逻辑清晰度、格式规范性。
    """,
    allow_delegation=False
)

6.3 看宝AI 的任务流设计

# 看宝AI 任务流设计
class KanbaoFlow(Flow[KanbaoState]):
    @start()
    def receive_request(self):
        """接收用户请求"""
        # 从用户输入获取信息
        pass
    
    @listen(receive_request)
    def understand_intent(self):
        """理解用户意图"""
        intent_crew = IntentCrew()
        intent = intent_crew.kickoff(inputs={"user_input": self.state.user_input})
        self.state.intent = intent
        return intent
    
    @listen(understand_intent)
    def load_memory(self):
        """加载用户记忆"""
        memory_crew = MemoryCrew()
        memory_data = memory_crew.kickoff(inputs={"user_id": self.state.user_id})
        self.state.memory_context = memory_data
        return memory_data
    
    @listen(load_memory)
    def plan_tasks(self):
        """规划任务"""
        planner_crew = PlannerCrew()
        plan = planner_crew.kickoff(inputs={
            "intent": self.state.intent,
            "memory": self.state.memory_context
        })
        self.state.task_plan = plan
        return plan
    
    @listen(plan_tasks)
    def execute_tasks(self):
        """执行任务"""
        executor_crew = ExecutorCrew()
        result = executor_crew.kickoff(inputs={"plan": self.state.task_plan})
        self.state.execution_result = result
        return result
    
    @listen(execute_tasks)
    def review_output(self):
        """审核输出"""
        reviewer_crew = ReviewerCrew()
        final_output = reviewer_crew.kickoff(inputs={"result": self.state.execution_result})
        self.state.final_output = final_output
        return final_output
    
    @listen(review_output)
    def save_memory(self):
        """保存记忆"""
        # 将本次交互的重要信息存入记忆
        pass
    
    @listen(save_memory)
    def send_response(self):
        """发送响应给用户"""
        pass
@persist
class KanbaoFlow(Flow[KanbaoState]):
    """持久化看宝AI Flow"""
    pass

6.4 看宝AI 的协作模式选择策略

任务类型	推荐模式	原因
简单问答	Sequential	单一 Agent 快速响应
信息收集+整理	Sequential	研究员 → 整理者
复杂分析报告	Hierarchical	Manager 协调多 Agent
多角度评估	Hierarchical	需要专家审核
用户交互对话	Flow	状态管理+上下文传递
长期项目	Flow + Memory	跨会话记忆和状态

6.5 看宝AI 的质量保障机制

# 看宝AI 质量保障机制
1. 任务输出验证
def validate_response_quality(output):
    """验证回复质量"""
    word_count = len(output.raw.split())
    
    # 检查是否为空
    if not output.raw.strip():
        return (False, "回复为空，请重新生成。")
    
    # 检查是否太短
    if word_count < 20:
        return (False, "回复太短，请提供更详细的回答。")
    
    # 检查是否包含免责声明（对于不确定的内容）
    uncertain_keywords = ["不确定", "可能", "建议"]
    if any(kw in output.raw for kw in uncertain_keywords):
        return (True, output.raw)  # 这些是合理的
    
    return (True, output.raw)
2. 事实核查回调
def fact_check_callback(step):
    """在每步执行后检查事实准确性"""
    # 实现事实核查逻辑
    pass
3. 用户满意度追踪
def track_satisfaction(output):
    """追踪用户满意度"""
    # 记录用户反馈
    pass

---

七、企业应用案例

课程中提到了 CrewAI 在企业中的实际应用：

7.1 AB InBev（百威英博）

应用场景：物流和库存管理优化效果： - 优化供应链流程 - 实时库存监控 - 智能补货建议

7.2 Snyk

应用场景：代码安全扫描效果： - 自动化代码审查 - 漏洞检测和修复建议 - 安全报告生成

7.3 Weaviate

应用场景：向量数据库智能检索效果： - 语义搜索优化 - 智能问答系统 - 知识库管理

7.4 Exa AI

应用场景：深度研究代理效果： - 网络信息搜集 - 多源信息整合 - 研究报告生成

---

八、学习资源推荐

8.1 官方资源

资源	链接
CrewAI 官网	https://www.crewai.com/
GitHub 仓库	https://github.com/crewAIInc/crewAI
官方文档	https://docs.crewai.com/
学习平台	https://learn.crewai.com/

8.2 进阶课程

课程	来源	特点
Design, Develop, and Deploy Multi-Agent Systems	DeepLearning.AI	4 模块，20 小时，生产级部署
Advanced Agent Patterns	CrewAI	Flow、MCP、Guardrails 深入

8.3 社区资源

- CrewAI Discord 社区 - GitHub Discussions - Reddit r/CrewAI

---

九、总结

9.1 核心要点

CrewAI 提供了一套完整的多 Agent 系统框架，包括 Agent、Task、Crew、Flow、Memory、Tools、Guardrails 等核心组件。

三种协作模式（Sequential、Hierarchical、Consensual）满足不同场景需求：

- Sequential：简单流水线 - Hierarchical：复杂任务协调

Memory 和 Flow 是生产级应用的关键：

- Memory 实现跨任务学习 - Flow 实现状态管理和精确控制

看宝AI 可以借鉴 CrewAI 的设计模式，构建模块化、可扩展的 Agent 团队架构。

9.2 下一步行动

- [ ] 在看宝AI 项目中实现基础的 Agent 团队 - [ ] 设计看宝AI 的核心 Agent 角色定义 - [ ] 实现任务流的状态管理 - [ ] 添加输出质量验证机制 - [ ] 测试不同协作模式的效果

---

学习笔记由 AI Agent 基于 DeepLearning.AI 课程内容生成 学习日期：2026-04-12 --- ## 📚 学习资源 ### GitHub项目 - [CrewAI](https://github.com/CrewAIInc/crewAI) ⭐ 48k+ - 多Agent协作框架，无需LangChain依赖 - [LangGraph](https://github.com/LangChain-AI/langgraph) ⭐ 25k+ - 构建有状态Agent的图形框架 - [vLLM](https://github.com/vllm-project/vllm) ⭐ 42k+ - 高性能LLM推理服务 - [LlamaIndex](https://github.com/run-llama/llamaindex) ⭐ 35k+ - RAG框架，数据与LLM之间的桥梁 ### 官方文档 - [CrewAI Docs](https://docs.crewai.com) - 官方文档和快速入门 - [LangGraph Docs](https://docs.langchain.com/oss/python/langgraph) - 官方文档 - [vLLM Docs](https://docs.vllm.ai) - 推理服务文档 ### 教程与课程 - [CrewAI Examples](https://github.com/CrewAIInc/crewAI-examples) - 官方示例项目 - [LangChain Academy](https://academy.langchain.com/) - 免费学习课程 - [deeplearning.ai Courses](https://www.deeplearning.ai/) - LLM专项课程

💬 AI学习讨论

0 条评论

暂无评论，成为第一个评论者吧！