评估 CrewAI 使用场景

理解决策框架

在使用 CrewAI 构建 AI 应用时，你要做的最重要决定之一，就是为具体用例选择合适的方案。应该使用 Crew？Flow？还是两者结合？本指南会帮助你评估需求，并做出有依据的架构决策。

这个决策的核心，是理解应用中的 复杂度 与精度之间的关系：

Complexity vs. Precision Matrix — CrewAI 应用中的复杂度与精度矩阵

这个矩阵帮助你直观理解不同方案如何匹配不同的复杂度和精度需求。我们来看看每个象限的含义，以及它如何指导你的架构选择。

复杂度-精度矩阵说明

什么是复杂度？

在 CrewAI 应用中，复杂度 指的是：

所需的不同步骤或操作数量
需要执行的任务种类
不同组件之间的相互依赖
对条件逻辑和分支的需求
整体工作流的复杂程度

什么是精度？

这里的精度指的是：

最终输出所需的准确性
对结构化、可预测结果的需求
可复现性的重要性
对每一步所需控制程度
对输出变化的容忍度

四个象限

1. 低复杂度，低精度

特征：

任务简单、直接
可以容忍一定的输出变化
步骤数量有限
偏创意或探索型应用

推荐方案： 使用少量 agent 的简单 Crew

示例场景：

基础内容生成
想法头脑风暴
简单摘要任务
创意写作辅助

2. 低复杂度，高精度

特征：

需要精确、结构化输出的简单工作流
需要结果可复现
步骤少，但对准确性要求高
常涉及数据处理或转换

推荐方案： 使用直接 LLM 调用的 Flows，或带结构化输出的简单 Crews

示例场景：

数据提取与转换
表单填写与校验
结构化内容生成（JSON、XML）
简单分类任务

3. 高复杂度，低精度

特征：

多阶段流程，包含很多步骤
输出偏创意或探索性
组件之间交互复杂
最终结果允许一定变化

推荐方案： 使用多个专门化 agent 的复杂 Crews

示例场景：

调研与分析
内容生产流水线
探索性数据分析
创造性问题解决

4. 高复杂度，高精度

特征：

需要结构化输出的复杂工作流
多个相互依赖的步骤，对准确性要求严格
同时需要复杂处理和精确结果
通常是关键任务型应用

推荐方案： 使用 Flows 编排多个 Crews，并加入校验步骤

示例场景：

企业决策支持系统
复杂数据处理流水线
多阶段文档处理
受监管行业应用

在 Crews 和 Flows 之间选择

何时选择 Crews

当满足以下情况时，Crews 很适合：

你需要协作式智能 - 多个具有不同专长的 agent 需要协同工作
问题需要涌现式思维 - 解决方案能从不同视角和方法中受益
任务主要是创意或分析型 - 工作涉及调研、内容创作或分析
你更看重适应性而不是严格结构 - 工作流可以从 agent 自主性中获益
输出格式可以比较灵活 - 对输出结构的变化有一定容忍度

# 示例：用于市场分析的 Research Crew
from crewai import Agent, Crew, Process, Task

# 创建专门化 agent
researcher = Agent(
    role="市场研究专家",
    goal="查找关于新兴技术的全面市场数据",
    backstory="你擅长发现市场趋势并收集数据。"
)

analyst = Agent(
    role="市场分析师",
    goal="分析市场数据并识别关键机会",
    backstory="你擅长解读市场数据并发现有价值的洞察。"
)

# 定义任务
research_task = Task(
    description="研究 AI 驱动医疗解决方案的当前市场格局",
    expected_output="包括主要参与者、市场规模和增长趋势在内的完整市场数据",
    agent=researcher
)

analysis_task = Task(
    description="分析市场数据并识别前三个投资机会",
    expected_output="包含 3 个推荐投资机会及其理由的分析报告",
    agent=analyst,
    context=[research_task]
)

# 创建 crew
market_analysis_crew = Crew(
    agents=[researcher, analyst],
    tasks=[research_task, analysis_task],
    process=Process.sequential,
    verbose=True
)

# 运行 crew
result = market_analysis_crew.kickoff()

何时选择 Flows

当满足以下情况时，Flows 很合适：

你需要对执行进行精确控制 - 工作流要求严格的顺序和状态管理
应用有复杂的状态需求 - 你需要在多个步骤之间维护和转换状态
你需要结构化、可预测的输出 - 应用要求一致、格式化的结果
工作流涉及条件逻辑 - 需要根据中间结果选择不同路径
你需要把 AI 与过程式代码结合 - 解决方案同时需要 AI 能力和传统编程

# 示例：带结构化处理的客户支持 Flow
from crewai.flow.flow import Flow, listen, or_, router, start
from pydantic import BaseModel
from typing import List, Dict

# 定义结构化状态
class SupportTicketState(BaseModel):
    ticket_id: str = ""
    customer_name: str = ""
    issue_description: str = ""
    category: str = ""
    priority: str = "medium"
    resolution: str = ""
    satisfaction_score: int = 0

class CustomerSupportFlow(Flow[SupportTicketState]):
    @start()
    def receive_ticket(self):
        # 在真实应用中，这里可能来自 API
        self.state.ticket_id = "TKT-12345"
        self.state.customer_name = "Alex Johnson"
        self.state.issue_description = "付款后无法访问高级功能"
        return "Ticket received"

    @listen(receive_ticket)
    def categorize_ticket(self, _):
        # 使用直接 LLM 调用来分类
        from crewai import LLM
        llm = LLM(model="openai/gpt-4o-mini")

        prompt = f"""
        将以下客户支持问题分类到下面类别之一：
        - Billing
        - Account Access
        - Technical Issue
        - Feature Request
        - Other

        问题：{self.state.issue_description}

        只返回类别名称。
        """

        self.state.category = llm.call(prompt).strip()
        return self.state.category

    @router(categorize_ticket)
    def route_by_category(self, category):
        # 根据类别路由到不同处理器
        return category.lower().replace(" ", "_")

    @listen("billing")
    def handle_billing_issue(self):
        # 处理账单相关逻辑
        self.state.priority = "high"
        # 更多账单处理...
        return "Billing issue handled"

    @listen("account_access")
    def handle_access_issue(self):
        # 处理访问相关逻辑
        self.state.priority = "high"
        # 更多访问处理...
        return "Access issue handled"

    # 其他类别处理器...

    @listen(or_("billing", "account_access", "technical_issue", "feature_request", "other"))
    def resolve_ticket(self, resolution_info):
        # 最终解决步骤
        self.state.resolution = f"Issue resolved: {resolution_info}"
        return self.state.resolution

# 运行 flow
support_flow = CustomerSupportFlow()
result = support_flow.kickoff()

何时结合使用 Crews 和 Flows

最复杂的应用通常会同时受益于 Crews 和 Flows 的结合：

复杂的多阶段流程 - 用 Flows 编排整体过程，用 Crews 处理复杂子任务
既需要创造力又需要结构的应用 - 用 Crews 处理创意任务，用 Flows 处理结构化流程
企业级 AI 应用 - 用 Flows 管理状态和流程，同时利用 Crews 处理专门工作

# 示例：结合 Crews 和 Flows 的内容生产流水线
from crewai.flow.flow import Flow, listen, start
from crewai import Agent, Crew, Process, Task
from pydantic import BaseModel
from typing import List, Dict

class ContentState(BaseModel):
    topic: str = ""
    target_audience: str = ""
    content_type: str = ""
    outline: Dict = {}
    draft_content: str = ""
    final_content: str = ""
    seo_score: int = 0

class ContentProductionFlow(Flow[ContentState]):
    @start()
    def initialize_project(self):
        # 设置初始参数
        self.state.topic = "可持续投资"
        self.state.target_audience = "千禧一代投资者"
        self.state.content_type = "博客文章"
        return "Project initialized"

    @listen(initialize_project)
    def create_outline(self, _):
        # 使用 research crew 创建大纲
        researcher = Agent(
            role="内容研究员",
            goal=f"为 {self.state.topic} 针对 {self.state.target_audience} 进行调研",
            backstory="你是一名擅长内容创作的资深研究员。"
        )

        outliner = Agent(
            role="内容策略师",
            goal=f"为 {self.state.content_type} 创建引人入胜的大纲",
            backstory="你擅长组织内容以实现最大化参与度。"
        )

        research_task = Task(
            description=f"围绕 {self.state.topic} 进行调研，重点关注 {self.state.target_audience} 会感兴趣的内容",
            expected_output="包含关键点和统计数据的完整研究笔记",
            agent=researcher
        )

        outline_task = Task(
            description=f"为关于 {self.state.topic} 的 {self.state.content_type} 创建大纲",
            expected_output="包含章节和关键点的详细内容大纲",
            agent=outliner,
            context=[research_task]
        )

        outline_crew = Crew(
            agents=[researcher, outliner],
            tasks=[research_task, outline_task],
            process=Process.sequential,
            verbose=True
        )

        # 运行 crew 并存储结果
        result = outline_crew.kickoff()

        # 解析大纲（在真实应用中，你可能会使用更稳健的解析方式）
        import json
        try:
            self.state.outline = json.loads(result.raw)
        except:
            # 如果不是有效 JSON，则使用兜底方案
            self.state.outline = {"sections": result.raw}

        return "Outline created"

    @listen(create_outline)
    def write_content(self, _):
        # 使用写作 crew 创建内容
        writer = Agent(
            role="内容写作者",
            goal=f"为 {self.state.target_audience} 撰写吸引人的内容",
            backstory="你是一名能够创作引人入胜内容的熟练写作者。"
        )

        editor = Agent(
            role="内容编辑",
            goal="确保内容经过润色、准确且有吸引力",
            backstory="你对细节有敏锐洞察，并擅长改进内容。"
        )

        writing_task = Task(
            description=f"根据这个大纲为 {self.state.topic} 撰写一篇 {self.state.content_type}：{self.state.outline}",
            expected_output="Markdown 格式的完整草稿内容",
            agent=writer
        )

        editing_task = Task(
            description="编辑并改进草稿内容，使其更清晰、更具吸引力且更准确",
            expected_output="Markdown 格式的润色后最终内容",
            agent=editor,
            context=[writing_task]
        )

        writing_crew = Crew(
            agents=[writer, editor],
            tasks=[writing_task, editing_task],
            process=Process.sequential,
            verbose=True
        )

        # 运行 crew 并存储结果
        result = writing_crew.kickoff()
        self.state.final_content = result.raw

        return "Content created"

    @listen(write_content)
    def optimize_for_seo(self, _):
        # 使用直接 LLM 调用进行 SEO 优化
        from crewai import LLM
        llm = LLM(model="openai/gpt-4o-mini")

        prompt = f"""
        分析这篇内容针对关键词 "{self.state.topic}" 的 SEO 效果。
        按 1-100 打分，并提供 3 条具体改进建议。

        内容：{self.state.final_content[:1000]}...（为简洁起见已截断）

        请将响应格式化为 JSON，结构如下：
        {{
            "score": 85,
            "recommendations": [
                "建议 1",
                "建议 2",
                "建议 3"
            ]
        }}
        """

        seo_analysis = llm.call(prompt)

        # 解析 SEO 分析
        import json
        try:
            analysis = json.loads(seo_analysis)
            self.state.seo_score = analysis.get("score", 0)
            return analysis
        except:
            self.state.seo_score = 50
            return {"score": 50, "recommendations": ["无法解析 SEO 分析"]}

# 运行 flow
content_flow = ContentProductionFlow()
result = content_flow.kickoff()

实用评估框架

要判断你的具体用例适合哪种方案，请按下面的步骤框架来评估：

第 1 步：评估复杂度

根据以下因素，把应用复杂度从 1 到 10 打分：

步骤数量：需要多少个不同的操作？
- 1-3 步：低复杂度（1-3）
- 4-7 步：中等复杂度（4-7）
- 8 步以上：高复杂度（8-10）
相互依赖：各部分之间有多强的耦合？
- 依赖很少：低复杂度（1-3）
- 有一些依赖：中等复杂度（4-7）
- 复杂依赖很多：高复杂度（8-10）
条件逻辑：需要多少分支和决策？
- 线性流程：低复杂度（1-3）
- 一些分支：中等复杂度（4-7）
- 复杂决策树：高复杂度（8-10）
领域知识：所需专业知识有多深？
- 通用知识：低复杂度（1-3）
- 一些专业知识：中等复杂度（4-7）
- 多个领域的深度专业知识：高复杂度（8-10）

计算平均分来确定整体复杂度。

第 2 步：评估精度需求

根据以下因素，把精度需求从 1 到 10 打分：

输出结构：输出需要多结构化？
- 自由文本：低精度（1-3）
- 半结构化：中等精度（4-7）
- 严格格式化（JSON、XML）：高精度（8-10）
准确性需求：事实准确有多重要？
- 创意内容：低精度（1-3）
- 信息性内容：中等精度（4-7）
- 关键信息：高精度（8-10）
可复现性：跨运行结果需要多一致？
- 允许变化：低精度（1-3）
- 需要一定一致性：中等精度（4-7）
- 需要完全可复现：高精度（8-10）
错误容忍度：错误带来的影响有多大？
- 影响小：低精度（1-3）
- 影响中等：中等精度（4-7）
- 影响高：高精度（8-10）

计算平均分来确定整体精度需求。

第 3 步：映射到矩阵

将你的复杂度和精度分数映射到矩阵中：

低复杂度（1-4），低精度（1-4）：简单 Crews
低复杂度（1-4），高精度（5-10）：带直接 LLM 调用的 Flows
高复杂度（5-10），低精度（1-4）：复杂 Crews
高复杂度（5-10），高精度（5-10）：编排 Crews 的 Flows

第 4 步：考虑其他因素

除了复杂度和精度，还要考虑：

开发时间：Crews 往往更快原型化
维护需求：Flows 通常更利于长期维护
团队经验：考虑团队对不同方案的熟悉程度
扩展性要求：Flows 往往更适合复杂应用的扩展
集成需求：考虑解决方案如何与现有系统集成

结论

在 Crews 和 Flows 之间做选择 - 或者将它们结合起来 - 是一个会影响 CrewAI 应用有效性、可维护性和可扩展性的关键架构决策。通过沿着复杂度和精度两个维度评估你的用例，你可以做出符合具体需求的明智决策。

请记住，最佳方案通常会随着应用成熟而演进。先从满足需求的最简单方案开始，并准备好在积累经验、需求变得更清晰时调整架构。

下一步

了解更多关于 crafting effective agents
探索 building your first crew
深入了解 mastering flow state management
查看核心概念以获得更深入的理解