CrewAI 工作流编排

CrewAI 的角色 - 任务 - 流程设计模式

🧩 1. CrewAI 中的 Crew、Task、Agent 三者是什么关系？

这三者就像一支项目团队（Crew），拿着任务书（Task），派给不同专长的成员（Agent）去执行。

• Agent：一个拥有特定角色（role）、目标（goal）、背景故事（backstory）的 AI 单元。它知道自己是谁、要做什么、用什么工具。比如“你是一位资深市场分析师，擅长用网络搜索和数据分析挖掘市场趋势”。

• Task：一份写清楚“你要干什么、期望产出什么”的工作说明书。包含描述（description）、期望输出（expected_output）、分配给哪个 Agent（agent），还可以指定是否允许异步（async_execution）、依赖哪个 Task（context）。

• Crew：把 Agent 和 Task 组装起来，并决定它们的执行方式（Process）和全局配置（如 Verbose 模式、最多轮次）。

用一句话串起来：Crew 是团队框架，Task 是分工说明书，Agent 是执行任务的专家。Crew 负责编排，让 Task 按顺序或层级流到对应的 Agent 手上。

---

⚖️ 2. CrewAI 的 Process 有 Sequential 和 Hierarchical 两种，有什么区别？

这两种 Process 就像两个不同的项目管理方式：一条流水线（Sequential） vs 一个项目经理分派任务（Hierarchical）。

Sequential 的特点：

• 像工厂流水线，任务一个接一个执行，后面任务可以拿到前面任务的输出作为上下文。

• 简单、确定性强，适合步骤明确、依赖关系固定的流程。

• 缺点是缺乏弹性：如果中间某步失败，整个链条可能卡住，而且无法并行处理无依赖的任务。

Hierarchical 的特点：

• 自动创建一个 Manager Agent（内部 LLM），它阅读所有任务和 Agent 的描述，然后决定什么时候把哪个任务分配给哪个 Agent。

• Manager 还能审查产出、要求返工、拆分任务，甚至调整顺序。

• 更灵活，适合复杂、无法提前完全固定步骤的场景，但多了一次 LLM 管理开销，行为也相对不可预测。

代码上的体现：

from crewai import Crew, Process

# 串行流程：按列表顺序执行
crew_sequential = Crew(
    agents=[...],
    tasks=[task1, task2, task3],
    process=Process.sequential  # 默认值
)

# 层级流程：让 Manager 自主决策
crew_hierarchical = Crew(
    agents=[...],
    tasks=[task1, task2, task3],
    process=Process.hierarchical,
    manager_llm=ChatOpenAI(model="gpt-4o")  # 需要指定管理者模型
)

选型建议： 如果流程可以画成一条清晰的甘特图，用 Sequential，简单可靠；如果流程像“根据初步分析结果再决定后续分析方向”这类带动态判断的，就用 Hierarchical 给 Agent 更大的自主空间。

---

🏗️ 3. 用 CrewAI 实现一个“市场调研报告生成”流程，如何设计 Task 依赖和上下文传递？

我们要设计一个生成《新能源两轮车市场调研报告》的自动化流程，包含四个步骤：竞品数据收集 → 行业趋势分析 → 撰写报告正文 → 审核与润色。

整体架构：

数据收集 (Agent A) ─┐
                    ├─→ 报告撰写 (Agent C) ─→ 审核 (Agent D)
趋势分析 (Agent B) ─┘

数据收集和趋势分析可以并行，因为它们互不依赖。报告撰写必须等它们都完成，审核则依赖报告初稿。

第一步：定义 Agent

from crewai import Agent, Task, Crew
from crewai.tools import SerperDevTool  # 搜索引擎工具

# 共享工具
search_tool = SerperDevTool()

# 竞品分析师
competitor_agent = Agent(
    role="竞品分析专家",
    goal="收集主要竞品的产品特性、定价、市场策略信息",
    backstory="你擅长通过公开网络信息快速整理竞品情报",
    tools=[search_tool],
    verbose=True
)

# 行业趋势分析师
trend_agent = Agent(
    role="行业趋势分析师",
    goal="分析新能源两轮车行业的技术趋势、政策环境和市场规模",
    backstory="你关注新能源、出行领域的宏观动向，能提炼关键洞察",
    tools=[search_tool],
    verbose=True
)

# 报告撰写员
writer_agent = Agent(
    role="报告撰写专家",
    goal="将数据和洞察整合成一份结构清晰、语言专业的市场调研报告",
    backstory="你擅长商业写作，能用客观数据和犀利观点打动读者",
    verbose=True
)

# 审核编辑
reviewer_agent = Agent(
    role="审核编辑",
    goal="检查报告的事实准确性、逻辑严谨性和语言流畅度",
    backstory="你对文字有洁癖，确保每一份报告零差错",
    verbose=True
)

第二步：设计 Task 及依赖关系

关键点在于 context 参数：它接收一个 Task 列表，当前 Task 会把列表中所有前置 Task 的输出自动注入提示词，从而让 Agent 能“看到”前面做了什么。

# Task 1: 竞品数据收集（无依赖）
task_competitor = Task(
    description="调研当前市场上主要的新能源两轮车品牌（如雅迪、爱玛、小牛、九号等），"
                "收集它们的主销车型、价格区间、续航里程、智能功能、渠道策略。",
    expected_output="一份结构化竞品对比表，包含品牌、车型、价格、续航、核心卖点",
    agent=competitor_agent,
    async_execution=True  # 可与 Task 2 并行执行
)

# Task 2: 行业趋势分析（无依赖）
task_trend = Task(
    description="分析新能源两轮车行业的最新政策（如新国标）、锂电池vs钠电池技术路线、"
                "市场规模增长预测、目标用户画像变化。",
    expected_output="一份行业趋势摘要，包含政策、技术、市场、用户四方面的关键发现",
    agent=trend_agent,
    async_execution=True  # 与 Task 1 并行
)

# Task 3: 撰写报告（依赖 Task 1 和 Task 2 的输出）
task_writing = Task(
    description="根据竞品数据和行业趋势，撰写《新能源两轮车市场调研报告》全文。"
                "报告结构：摘要、市场概况、竞品分析、趋势洞察、机会与风险、建议。"
                "使用专业但易懂的语言，1500字左右。",
    expected_output="完整的市场调研报告 Markdown 文本",
    agent=writer_agent,
    context=[task_competitor, task_trend]  # 关键：注入前两个任务的输出
)

# Task 4: 审核润色（依赖 Task 3）
task_review = Task(
    description="对报告初稿进行审校。检查数据准确性（对照竞品表和趋势摘要）、"
                "逻辑连贯性、语言表达。有疑问处标出修改建议，最后输出润色后的终稿。",
    expected_output="润色后的最终报告，附带修改摘要",
    agent=reviewer_agent,
    context=[task_writing]
)

第三步：组装 Crew 并运行

crew = Crew(
    agents=[competitor_agent, trend_agent, writer_agent, reviewer_agent],
    tasks=[task_competitor, task_trend, task_writing, task_review],
    process=Process.sequential,  # 虽然串行，但内部异步任务会并行执行
    verbose=True
)

result = crew.kickoff(inputs={"topic": "新能源两轮车市场调研"})
print(result)

关键设计解读：

• 并行提速：async_execution=True 让 Task 1 和 Task 2 同时跑，整个流程时长大约等于最慢的那个收集任务 + 撰写 + 审核，而不是四者累加。

• 上下文穿透：Task 3 的 context=[task_competitor, task_trend] 会把前两个任务的 expected_output 文本自动拼入 Task 3 的描述中，所以撰写 Agent 能看到真实的竞品数据和趋势分析，而不是凭空想象。

• 质量闭环：最后一个审核 Task 用 context=[task_writing] 拿到初稿，并让审核 Agent 对照原始数据（在提示词中要求“对照竞品表和趋势摘要”），从而形成事实核查的闭环，减少幻觉。

• 确定性优先：选择了 Sequential Process，因为这里的步骤逻辑清晰，不需要 Manager 动态决策。如果希望将来流程更灵活（比如审核不通过自动打回重写），可以升级为 Hierarchical。

---

这个设计把 CrewAI 的三个核心概念串了起来：Agent 是脑，Task 是手，Crew 是神经中枢。依赖关系用 context 自然表达，并行标记让无依赖的任务互不等待，最终形成一条从数据到洞察、从洞察到报告、从报告到终稿的完整知识生产线。

---

🧠 4. CrewAI 的 Memory 机制（短期记忆、长期记忆、实体记忆）如何工作？对多轮任务执行有什么影响？

CrewAI 的记忆系统模拟了人类的认知过程：短期记忆 让你记住当前对话，长期记忆 让你积累跨任务的经验，实体记忆 帮你结构化地记住关键事实。

短期记忆：Agent 在单次任务执行中，看到的上下文、工具调用结果和推理过程都保存在短期记忆里。它本质上就是 LLM 上下文窗口中的历史消息，受 max_rpm 和上下文窗口限制。当同一 Crew 的多个 Task 串行时，短期记忆通过 context 参数在 Task 之间传递。

长期记忆：CrewAI 会把执行完毕的任务总结、关键经验存入长期记忆存储（可以是内存、SQLite 或用户自定义的向量库）。当同一个 Crew 再次执行类似任务时，长期记忆会自动检索相关经验，注入到 System Prompt 中，让 Agent “回忆”起以前的成功做法。这通过 memory=True 开启，并可配置存储后端。

实体记忆：它像一个关系数据库，记录任务中提到的实体（如用户、公司、产品）及其属性。实体记忆通过 LLM 自动提取，并允许 Agent 后续查询“这个实体我们还知道什么”。这对多轮、跨会话的信息追踪特别有用。

工作流程示意：

from crewai import Crew, Agent, Task, Process

# 开启记忆的 Crew
crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,                    # 开启长期记忆
    memory_config={
        "provider": "mem0",         # 可选 mem0, 本地文件等
        "config": {"user_id": "my-app"},
    },
)
crew.kickoff()
# 任务执行后，关键经验存入长期记忆。下次执行类似任务时自动检索。

对多轮任务执行的影响：

• 连贯性：短期记忆让 Agent 知道“刚才做了什么”，在顺序依赖的任务流（如先查数据、后写报告）中，无需手动复制粘贴所有中间结果，context 参数自动传递。

• 学习进化：长期记忆让 Agent 从过去的执行中学习。例如，安全审查 Agent 记住了上次发现的常见漏洞模式，下次审查时会更快定位。

• 事实一致性：实体记忆使 Agent 在多轮交互中保持对同一事物的认知一致。比如用户在第一轮提到“项目 Alpha”，三天后再问“Alpha 的进展”，Agent 能调出实体的属性。

• 注意事项：记忆不是越多越好，长期记忆需要清理或压缩，避免检索噪音；实体记忆过度提取可能引入幻觉。配置时需权衡深度和存储成本。

---

🛠️ 5. CrewAI 中如何自定义 Tool 并绑定给特定 Agent？Tool 的错误处理机制是怎样的？

CrewAI 的工具体系基于 LangChain 的 BaseTool，可以轻松扩展。自定义工具只需继承 BaseTool，实现 _run 方法，并在 Agent 定义时传入。

自定义工具步骤：

from crewai.tools import BaseTool
from typing import Type, Optional
from pydantic import BaseModel, Field
import requests

# 1. 定义工具的输入 Schema
class CodeAnalyzerInput(BaseModel):
    code_snippet: str = Field(description="需要分析的代码片段")
    language: Optional[str] = Field(default="python", description="编程语言")

# 2. 实现工具类
class CodeAnalyzerTool(BaseTool):
    name: str = "代码静态分析器"
    description: str = "分析给定的代码片段，返回潜在的代码异味、复杂度警告和改进建议。"
    args_schema: Type[BaseModel] = CodeAnalyzerInput

    def _run(self, code_snippet: str, language: str = "python") -> str:
        # 实际可以调用 Pylint、SonarQube 或 LLM 进行分析
        try:
            # 模拟分析过程
            issues = self._analyze(code_snippet, language)
            return f"分析结果：\n{issues}"
        except Exception as e:
            return f"工具执行出错：{str(e)}"

    def _analyze(self, code, lang):
        # 这里放入真实的分析逻辑
        if "eval(" in code:
            return "⚠️ 安全风险：发现 eval() 调用，可能导致代码注入。"
        if len(code) > 200:
            return "ℹ️ 代码较长，建议拆分为更小的函数。"
        return "✅ 未发现明显问题。"

# 3. 绑定给特定 Agent
code_agent = Agent(
    role="代码分析专家",
    goal="审查代码质量并给出改进建议",
    backstory="你是一位资深的代码审查者，能够快速发现代码中的坏味道。",
    tools=[CodeAnalyzerTool()],
    verbose=True
)

工具的错误处理机制：

CrewAI 对工具调用有一套内置的容错逻辑：

• 工具内部抛出异常时，CrewAI 捕获异常并将错误消息（如 "工具执行出错：...") 作为 Observation 返回给 Agent。

• Agent 看到错误后，可以尝试修正参数并再次调用，或者在推理中放弃该工具并改用其他方式。

• 如果启用了 verbose=True，你会在日志中看到工具调用失败的具体原因。

• 对于连续失败，Agent 会在达到最大重试次数（默认通常为 3）后放弃该工具，继续下一步推理。

自定义错误处理的最佳实践：

• 在 _run 中捕获所有异常，返回结构化的错误信息，而不是让异常直接抛出。这样 Agent 能理解错误并调整行为。

• 对于有副作用的工具（如写文件），使用事务或补偿逻辑，避免部分成功导致脏数据。

• 可以结合 CrewAI 的 max_iter 限制全局重试，防止无限循环。

---

🏗️ 6. 用 CrewAI 实现一个多 Agent 代码审查系统，如何设计 Agent 角色、Task 依赖和输出格式？

我们设计一个系统，对提交的代码进行静态分析、安全漏洞扫描、性能评估，最后汇总成审查报告。四个 Agent 分工明确，依赖关系清晰。

架构图：

第一步：定义 Agent（角色、目标、背景、工具）

from crewai import Agent, Task, Crew, Process
from crewai.tools import BaseTool

# 工具占位（可以复用前面的自定义工具）
class StaticAnalysisTool(BaseTool):
    name: str = "静态分析工具"
    description: str = "检查代码风格、复杂度、潜在Bug"
    def _run(self, code: str) -> str:
        # 模拟分析
        return "代码行数: 150, 圈复杂度: 8, 未使用的变量: 2"

class SecurityScanTool(BaseTool):
    name: str = "安全扫描工具"
    description: str = "扫描常见安全漏洞：SQL注入、XSS、硬编码密钥等"
    def _run(self, code: str) -> str:
        return "发现硬编码的API密钥1处，SQL注入风险1处"

class PerformanceProfilerTool(BaseTool):
    name: str = "性能分析工具"
    description: str = "评估时间复杂度、内存占用等"
    def _run(self, code: str) -> str:
        return "最坏时间复杂度 O(n²)，建议优化循环"

# Agent 定义
static_agent = Agent(
    role="代码静态分析专家",
    goal="对代码进行静态分析，识别代码异味、风格问题和逻辑错误",
    backstory="你是一位注重代码质量的工匠，擅长用工具和眼光发现隐藏的问题。",
    tools=[StaticAnalysisTool()],
    verbose=True
)

security_agent = Agent(
    role="安全审查专家",
    goal="发现代码中的安全漏洞并提供修复建议",
    backstory="你是一位白帽黑客，熟悉OWASP Top 10，能够快速定位安全风险。",
    tools=[SecurityScanTool()],
    verbose=True
)

performance_agent = Agent(
    role="性能优化专家",
    goal="评估代码的时间和空间复杂度，找出性能瓶颈",
    backstory="你是一位追求极致性能的工程师，对算法复杂度了如指掌。",
    tools=[PerformanceProfilerTool()],
    verbose=True
)

reporter_agent = Agent(
    role="审查报告汇总专家",
    goal="整合各专项审查结果，生成一份结构清晰、可操作的代码审查报告",
    backstory="你是一位技术文档撰写高手，能将零散发现整理成规范的审查报告。",
    verbose=True
)

第二步：定义 Task 及依赖关系

# 待审查的代码（实际可从文件读取或API传入）
code_sample = """
def fetch_user(user_id):
    query = "SELECT * FROM users WHERE id = " + user_id
    return database.execute(query)
"""

# Task 1-3：三个专项审查，可并行（async_execution）
task_static = Task(
    description=f"对以下代码执行静态分析：\n{code_sample}",
    expected_output="列出代码异味、风格问题和潜在Bug的清单",
    agent=static_agent,
    async_execution=True
)

task_security = Task(
    description=f"对以下代码进行安全漏洞扫描：\n{code_sample}",
    expected_output="列出所有安全漏洞，按严重程度分级，并给出修复建议",
    agent=security_agent,
    async_execution=True
)

task_performance = Task(
    description=f"分析以下代码的性能特征：\n{code_sample}",
    expected_output="评估时间/空间复杂度，指出性能瓶颈和优化方向",
    agent=performance_agent,
    async_execution=True
)

# Task 4：汇总，依赖前三个输出
task_report = Task(
    description="根据代码静态分析、安全扫描、性能分析的结果，生成一份综合代码审查报告。"
                "报告需包含：摘要、静态分析结果、安全漏洞详情、性能评估、总体建议。"
                "使用 Markdown 格式，对每个问题提供代码片段和行号。",
    expected_output="一份完整的 Markdown 格式代码审查报告",
    agent=reporter_agent,
    context=[task_static, task_security, task_performance]  # 关键：自动注入前置结果
)

第三步：组装 Crew 并运行

crew = Crew(
    agents=[static_agent, security_agent, performance_agent, reporter_agent],
    tasks=[task_static, task_security, task_performance, task_report],
    process=Process.sequential,   # 串行，但内部 async_execution 实现并行
    verbose=True
)

result = crew.kickoff()
print(result)

设计要点解析：

• 角色专精：每个 Agent 只负责一个维度，System Prompt 和工具都高度定制，保证了审查的专业深度。

• 并行提速：三个专项审查设置为 async_execution=True，CrewAI 会同时启动它们，总耗时约等于最慢的一个审查时间。

• 上下文粘合：汇总 Agent 的 Task 通过 context 获得了前三个 Task 的输出全文，无需人工拼接。汇总 Agent 在提示词中直接看到“静态分析结果：...”、“安全扫描结果：...”，从而能写出连贯的报告。

• 输出格式约束：在 expected_output 中明确要求 Markdown 格式、包含修复建议等，让最终产物可直接作为 PR 评论发布。

• 扩展性：如果未来需要增加“依赖分析 Agent”，只需新加一个 Agent 和 Task，并将其 context 设为现有任务即可，不影响其他 Agent。

实际运行可能的输出片段：

# 代码审查报告

## 摘要
本次审查发现 **3个安全漏洞**、**2个代码异味**、**1个性能瓶颈**，建议立即修复。

## 安全漏洞
| 严重程度 | 位置 | 描述 | 修复建议 |
|---------|------|-----|---------|
| 高危 | fetch_user() L2 | SQL注入：直接拼接user_id | 使用参数化查询 |

## 静态分析
- 未使用的变量：无
- 圈复杂度：8 (偏高)

## 性能评估
- 数据库查询未加索引提示，可能导致全表扫描

7、CrewAI 的 Memory 机制（短期记忆、长期记忆、实体记忆）如何工作？对多轮任务执行有什么影响？⭐⭐⭐

难度级别：⭐⭐⭐（Memory 机制、多轮任务、上下文传递）

1️⃣ Common Answer

嗯，CrewAI 的 Memory 机制就是让 Agent 记住之前执行过的任务。它有三种记忆，短期记忆就是记住最近几轮对话，长期记忆就是保存到数据库里，实体记忆就是记住人名地名这些实体。对多轮任务执行来说，有了记忆就可以让后面的任务用到前面的结果，不用重复输入。比如前面查了某个数据，后面分析的时候就能直接用。不过记忆太多了可能会影响性能，所以要注意控制。

2️⃣ Impressive Answer

CrewAI 的 Memory 机制是支持多 Agent 协作和多轮任务执行的核心能力，通过三种类型的记忆实现不同层次的信息持久化和复用。

1. 短期记忆
2. 基于 Token 窗口的上下文记忆，在单次 Crew 执行过程中维护
3. 使用 RAGCallbackHandler 捕获每个 Task 的输入输出，存储在内存中
4. 后续 Task 可通过 memory.read() 获取历史上下文，自动注入到 Prompt 中

5. 适用于单次会话内的任务依赖，如 Task A 的输出作为 Task B 的输入

6. 长期记忆

7. 基于 Vector Database 的持久化存储（支持 Chroma、Pinecone、Redis Vector）
8. 使用 Embedding 模型将 Task 结果向量化存储，支持语义检索
9. 通过 LongTermMemory 类实现，跨 Crew 执行会话共享历史经验

10. 典型应用场景：代码审查 Agent 记住之前发现的模式，市场调研 Agent 积累行业知识

11. 实体记忆

12. 基于 NER（命名实体识别）的结构化记忆，提取关键实体（人名、公司名、技术术语）
13. 使用 EntityMemory 类维护实体关系图谱，支持实体消歧和关联查询
14. 通过 EntityExtractor 从对话中提取实体，存储为 JSON 格式
15. 实际案例：客服 Agent 记住用户偏好、产品 Agent 维护产品知识图谱

对多轮任务执行的影响：

from crewai import Agent, Task, Crew, Process
from crewai.memory import LongTermMemory, ShortTermMemory, EntityMemory

# 配置三种记忆
memory_config = {
    "short_term": ShortTermMemory(),
    "long_term": LongTermMemory(
        storage_backend="chroma",
        embedding_model="text-embedding-3-small"
    ),
    "entity": EntityMemory(
        entity_extractor="openai",
        enable_entity_graph=True
    )
}

# 创建带记忆的 Agent
researcher = Agent(
    role="研究员",
    goal="收集并分析市场数据",
    memory=memory_config  # 绑定记忆配置
)

# Task 依赖记忆
task1 = Task(
    description="收集 2024 年 AI 市场数据",
    expected_output="JSON 格式的市场数据"
)

task2 = Task(
    description="基于 task1 的数据，分析增长趋势",
    context=[task1],  # 明确依赖关系
    expected_output="趋势分析报告"
)

crew = Crew(
    agents=[researcher],
    tasks=[task1, task2],
    process=Process.sequential,
    memory=True  # 启用记忆
)

关键影响：

• 上下文传递：Task 间自动传递历史信息，减少重复输入

• 知识积累：长期记忆让 Agent 越用越聪明，积累领域知识

• 一致性保证：实体记忆确保多轮对话中实体指代一致

• 性能权衡：记忆占用 Token 空间，需设置 max_token_limit 控制成本

3️⃣ Key Differences

查看内嵌表格

---

CrewAI 的核心概念与角色驱动的任务编排

🧱 1. CrewAI 的四层结构分别是什么？

CrewAI 的设计可以分为从上到下的四个层级：Orchestration Layer、Agent Layer、Task Layer、Tool Layer。每一层都有明确的职责和对应的核心概念。

各层职责与关系：

• 编排层 决定了任务如何流转：是固定顺序还是交由管理 Agent 动态分派。它还负责开启记忆、设置最大 RPM 等全局参数。

• Agent 层 是执行智能体，每个 Agent 有自己的“人设”和推理方式。它接受编排层分配的任务，使用工具完成任务并产出结果。

• 任务层 是工作的原子单位，它定义了要做什么、期望什么产出、谁来做、依赖哪些前置任务。

• 工具层 是 Agent 的手和脚，让 Agent 能够搜索、计算、读写文件等。

用一个代码片段感受四层是如何被实例化的：

from crewai import Crew, Agent, Task, Process
from crewai_tools import SerperDevTool

# 工具层
search_tool = SerperDevTool()

# Agent 层
analyst = Agent(
    role="市场分析师",
    goal="收集并整理市场数据",
    backstory="你擅长数据搜集与分析",
    tools=[search_tool],
    verbose=True
)

# 任务层
task_collect = Task(
    description="收集新能源两轮车市场前三大品牌的信息",
    expected_output="包含品牌、市场份额、价格区间的结构化 JSON 列表",
    agent=analyst
)

# 编排层
crew = Crew(
    agents=[analyst],
    tasks=[task_collect],
    process=Process.sequential,
    memory=True,
    verbose=True
)

result = crew.kickoff()

这个四层模型的关键是 关注点分离：你可以单独改进某个 Agent 的 prompt，或者新增一个 Tool，或者把 Task 的依赖关系重新编排，而不会牵一发动全身。

---

⚖️ 2. CrewAI 的 Sequential 和 Hierarchical 两种 Process 模式有什么区别，与 LangGraph 相比设计哲学有何不同？

2.1 Sequential vs Hierarchical 的区别

这是两种完全不同的工作流控制模式，就像 自动化流水线 和 项目负责人分配任务 的区别。

核心区别表：

查看内嵌表格

代码示例：

# Sequential：按列表顺序执行
crew_seq = Crew(
    agents=[agent1, agent2, agent3],
    tasks=[task1, task2, task3],
    process=Process.sequential
)

# Hierarchical：Manager Agent 自主分配
crew_hier = Crew(
    agents=[agent1, agent2, agent3],
    tasks=[task1, task2, task3],
    process=Process.hierarchical,
    manager_llm=ChatOpenAI(model="gpt-4o")  # 指定管理 LLM
)

2.2 与 LangGraph 的设计哲学对比

这是一个更深刻的问题：两个框架都解决“多步骤 LLM 应用编排”，但出发点截然不同。

LangGraph：把流程建模为有向图（StateGraph），节点是函数（Agent、Tool、LLM 调用），边是控制流。它追求的是精确可控——你显式定义节点和条件边，每一步的输入输出都有 Schema 约束，可以中断、重放、时间旅行。它的哲学是 “把 Agent 拆成可验证的状态机”。

CrewAI：把流程建模为团队协作，Agent 是角色，Task 是工作说明书。它追求的是自然直观——用角色、目标、背景故事这些拟人概念来定义 Agent，让开发者像管理团队一样管理 AI。它的哲学是 “把 AI 当成团队成员，而不是状态机节点”。

哲学差异一览：

查看内嵌表格

选择建议：

• 如果你希望快速搭建一个由多个角色协作完成的工作流，且角色分工自然，用 CrewAI，它的概念门槛低，业务人员也能看懂。

• 如果你需要极致控制，比如在某个步骤后插入人工审批、根据中间结果动态分支、持久化每一步状态，用 LangGraph，它给你手术刀级别的控制力。

• 两者不是互斥的：LangGraph 可以作为底层引擎实现更复杂的 CrewAI 编排，或者在一个 LangGraph 图中用一个节点调用一个 CrewAI Crew 来完成特定子任务。

---

🎛️ 3. 在 CrewAI Hierarchical 模式下，Manager Agent 行为不可预测怎么处理？

Hierarchical 模式的灵活换来的是不确定性：Manager 可能分配错任务、跳过关键步骤、过早结束或无限循环。我们需要一套约束 + 监控 + 兜底的策略来驯服它。

策略一：约束 Manager 的 System Prompt

Manager Agent 是一个内部创建的 LLM，我们可以通过 manager_agent_config 参数注入自定义指令，明确它的行为边界。

crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.hierarchical,
    manager_llm=ChatOpenAI(model="gpt-4o"),
    manager_agent_config={
        "role": "严格的项目经理",
        "goal": "确保所有任务按顺序高质量完成，不跳过任何步骤",
        "backstory": "你是一个一丝不苟的管理者，必须让每个 Agent 完成指定的全部 Task 后才结束。",
        "allow_delegation": False,  # 禁止 Manager 把任务分配给不在列表中的 Agent
        "max_iter": 10,            # 限制最大推理轮次
    }
)

策略二：用 Task 的 expected_output 和 context 锁定预期

Manager 的行为很大程度上取决于 Task 定义的清晰度。模糊的任务描述会导致 Manager 随意解读。

# ❌ 模糊：Manager 可能自己脑补
task = Task(description="分析市场", expected_output="市场分析报告")

# ✅ 具体：Manager 清楚必须产出什么，如何验证
task = Task(
    description="收集中国市场前三大新能源两轮车品牌的2025年销量、均价、用户评分。"
                "然后分析它们的竞争优劣势。",
    expected_output="一个 JSON 数组，每个元素包含 brand, sales_2025, avg_price, rating, strengths, weaknesses",
    agent=analyst
)

策略三：设置硬性约束和早停机制

在 Crew 级别设置 max_rpm（每分钟最大请求数）防止刷爆 API，设置 verbose=True 实时观察 Manager 的决策。还可以通过回调函数监控异常行为。

def monitor_callback(step_output):
    # 如果 Manager 连续三次分配同一个 Agent，发出警告或强制终止
    if detect_loop(step_output):
        raise RuntimeError("检测到 Manager 陷入循环，终止执行")

crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.hierarchical,
    verbose=True,
    step_callback=monitor_callback,
    max_rpm=10
)

策略四：混合模式降级 —— 用 Sequential 作为兜底

如果 Hierarchical 模式在生产环境的表现不够稳定，可以将流程拆分为外层 Sequential + 内层 Hierarchical。外层保证关键步骤的顺序，内层在某个步骤内部使用 Hierarchical 获得灵活性。

# 外层：固定流程
task_collect = Task(...)  # Sequential
task_analyze = Task(...)  # 内部使用 Hierarchical Crew

# 内层 Hierarchical 只在分析阶段使用
inner_crew = Crew(
    agents=[analyst1, analyst2],
    tasks=[task_analyze],
    process=Process.hierarchical
)

def analyze_step():
    return inner_crew.kickoff()

# 外层 Sequential 调用内层函数节点

策略五：人工审批节点

在 Task 定义中可以通过 human_input=True 要求 Manager 在完成某些关键步骤后等待人工审批，确保不跑偏。

task_report = Task(
    description="生成最终分析报告",
    expected_output="Markdown 格式报告",
    agent=reporter,
    human_input=True  # Manager 执行完此任务后会暂停，等待人工确认
)

策略六：事后评估与自动重试

每次运行后，用一个评估 Agent 审查 Manager 的决策日志，如果发现不合理的分配，自动重新运行并附带纠正指令。这个评估可以作为 CI/CD 的一部分，在每次部署新 prompt 时验证。

# 评估 Manager 决策质量
eval_agent = Agent(
    role="质量评估员",
    goal="检查 Manager 的决策日志，判断是否合理",
    backstory="你负责监督 AI 团队的工作质量"
)

eval_task = Task(
    description=f"审查以下决策日志：{manager_log}，判断 Manager 是否合理分配了任务，"
                "如果没有，指出具体问题。",
    expected_output="通过/不通过，以及原因",
    agent=eval_agent
)

总结一句话：

Hierarchical Manager 就像一个聪明但偶尔随性的实习生——你给它的指令越具体、边界越清晰、监控越实时，它的表现就越可靠。永远不要完全放任一个 LLM 自主决策，而是用约束、任务模板、回调、混合模式和人工兜底，把它框在一个安全又高效的范围内。

---