评估 CrewAI 的用例

了解如何评估你的 AI 应用需求，并根据复杂度和精确度要求，在 Crews 与 Flows 之间选择合适的方法。

理解决策框架

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

这一决策的核心，在于理解你的应用中 复杂度 与 精确度 之间的关系：

复杂度 vs. 精确度矩阵

这个矩阵有助于可视化不同方法如何对应不同的复杂度与精确度要求。接下来我们将逐一解释每个象限的含义，以及它如何指导你的架构选择。

复杂度 - 精确度矩阵详解

什么是复杂度？

在 CrewAI 应用的语境下，复杂度 指的是：

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

什么是精确度？

这里的 精确度 指的是：

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

四个象限

1. 低复杂度，低精确度

特征：

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

推荐方法： 使用少量 Agent 的简单 Crews

示例用例：

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

2. 低复杂度，高精确度

特征：

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

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

示例用例：

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

3. 高复杂度，低精确度

特征：

  • 多阶段、多步骤流程
  • 偏创意型或探索型输出
  • 组件之间存在复杂交互
  • 对最终结果中的变化有一定容忍度

推荐方法： 使用包含多个专业 Agent 的复杂 Crews

示例用例：

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

4. 高复杂度，高精确度

特征：

  • 需要结构化输出的复杂工作流
  • 多个相互依赖的步骤，且精度要求严格
  • 同时需要复杂处理能力与精确结果
  • 通常属于关键业务应用

推荐方法： 使用带验证步骤的 Flows 来编排多个 Crews

示例用例：

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

在 Crews 与 Flows 之间进行选择

何时选择 Crews

Crews 适合以下情况：

  1. 你需要协作式智能 —— 多个不同专长的 Agent 需要协同工作
  2. 问题需要涌现式思考 —— 解决方案能够从不同视角和方法中受益
  3. 任务主要偏创意或分析型 —— 工作涉及研究、内容创作或分析
  4. 你更看重适应性而非严格结构 —— 工作流可以从 Agent 自主性中获益
  5. 输出格式可以有一定弹性 —— 输出结构允许一定变化

```python theme={null}

Example: Research Crew for market analysis

from crewai import Agent, Crew, Process, Task

Create specialized agents

researcher = Agent( role=”Market Research Specialist”, goal=”Find comprehensive market data on emerging technologies”, backstory=”You are an expert at discovering market trends and gathering data.” )

analyst = Agent( role=”Market Analyst”, goal=”Analyze market data and identify key opportunities”, backstory=”You excel at interpreting market data and spotting valuable insights.” )

Define their tasks

research_task = Task( description=”Research the current market landscape for AI-powered healthcare solutions”, expected_output=”Comprehensive market data including key players, market size, and growth trends”, agent=researcher )

analysis_task = Task( description=”Analyze the market data and identify the top 3 investment opportunities”, expected_output=”Analysis report with 3 recommended investment opportunities and rationale”, agent=analyst, context=[research_task] )

Create the crew

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

Run the crew

result = market_analysis_crew.kickoff()

  2. ### 何时选择 Flows
  4. Flows 适合以下情况：
  6. 1. **你需要对执行过程进行精确控制** —— 工作流需要精确的顺序控制与状态管理
  7. 2. **应用存在复杂状态需求** —— 你需要在多个步骤之间维护并转换状态
  8. 3. **你需要结构化、可预测的输出** —— 应用要求一致且格式固定的结果
  9. 4. **工作流包含条件逻辑** —— 需要根据中间结果走不同路径
  10. 5. **你需要将 AI 与过程式代码结合** —— 解决方案同时需要 AI 能力与传统编程
  12. ```python theme={null}
  13. # Example: Customer Support Flow with structured processing
  14. from crewai.flow.flow import Flow, listen, router, start
  15. from pydantic import BaseModel
  16. from typing import List, Dict
  18. # Define structured state
  19. class SupportTicketState(BaseModel):
  20.     ticket_id: str = ""
  21.     customer_name: str = ""
  22.     issue_description: str = ""
  23.     category: str = ""
  24.     priority: str = "medium"
  25.     resolution: str = ""
  26.     satisfaction_score: int = 0
  28. class CustomerSupportFlow(Flow[SupportTicketState]):
  29.     @start()
  30.     def receive_ticket(self):
  31.         # In a real app, this might come from an API
  32.         self.state.ticket_id = "TKT-12345"
  33.         self.state.customer_name = "Alex Johnson"
  34.         self.state.issue_description = "Unable to access premium features after payment"
  35.         return "Ticket received"
  37.     @listen(receive_ticket)
  38.     def categorize_ticket(self, _):
  39.         # Use a direct LLM call for categorization
  40.         from crewai import LLM
  41.         llm = LLM(model="openai/gpt-4o-mini")
  43.         prompt = f"""
  44.         Categorize the following customer support issue into one of these categories:
  45.         - Billing
  46.         - Account Access
  47.         - Technical Issue
  48.         - Feature Request
  49.         - Other
  51.         Issue: {self.state.issue_description}
  53.         Return only the category name.
  54.         """
  56.         self.state.category = llm.call(prompt).strip()
  57.         return self.state.category
  59.     @router(categorize_ticket)
  60.     def route_by_category(self, category):
  61.         # Route to different handlers based on category
  62.         return category.lower().replace(" ", "_")
  64.     @listen("billing")
  65.     def handle_billing_issue(self):
  66.         # Handle billing-specific logic
  67.         self.state.priority = "high"
  68.         # More billing-specific processing...
  69.         return "Billing issue handled"
  71.     @listen("account_access")
  72.     def handle_access_issue(self):
  73.         # Handle access-specific logic
  74.         self.state.priority = "high"
  75.         # More access-specific processing...
  76.         return "Access issue handled"
  78.     # Additional category handlers...
  80.     @listen("billing", "account_access", "technical_issue", "feature_request", "other")
  81.     def resolve_ticket(self, resolution_info):
  82.         # Final resolution step
  83.         self.state.resolution = f"Issue resolved: {resolution_info}"
  84.         return self.state.resolution
  86. # Run the flow
  87. support_flow = CustomerSupportFlow()
  88. result = support_flow.kickoff()

何时结合使用 Crews 与 Flows

最复杂的应用通常会受益于将 Crews 与 Flows 结合使用：

  1. 复杂的多阶段流程 —— 用 Flows 编排整体流程，用 Crews 处理复杂子任务
  2. 同时要求创造性与结构性的应用 —— 用 Crews 处理创意任务，用 Flows 处理结构化流程
  3. 企业级 AI 应用 —— 用 Flows 管理状态与流程，同时借助 Crews 完成专业化工作

```python theme={null}

Example: Content Production Pipeline combining Crews and 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):

  1.     # Set initial parameters
  2.     self.state.topic = "Sustainable Investing"
  3.     self.state.target_audience = "Millennial Investors"
  4.     self.state.content_type = "Blog Post"
  5.     return "Project initialized"
  7. @listen(initialize_project)
  8. def create_outline(self, _):
  9.     # Use a research crew to create an outline
  10.     researcher = Agent(
  11.         role="Content Researcher",
  12.         goal=f"Research {self.state.topic} for {self.state.target_audience}",
  13.         backstory="You are an expert researcher with deep knowledge of content creation."
  14.     )
  16.     outliner = Agent(
  17.         role="Content Strategist",
  18.         goal=f"Create an engaging outline for a {self.state.content_type}",
  19.         backstory="You excel at structuring content for maximum engagement."
  20.     )
  22.     research_task = Task(
  23.         description=f"Research {self.state.topic} focusing on what would interest {self.state.target_audience}",
  24.         expected_output="Comprehensive research notes with key points and statistics",
  25.         agent=researcher
  26.     )
  28.     outline_task = Task(
  29.         description=f"Create an outline for a {self.state.content_type} about {self.state.topic}",
  30.         expected_output="Detailed content outline with sections and key points",
  31.         agent=outliner,
  32.         context=[research_task]
  33.     )
  35.     outline_crew = Crew(
  36.         agents=[researcher, outliner],
  37.         tasks=[research_task, outline_task],
  38.         process=Process.sequential,
  39.         verbose=True
  40.     )
  42.     # Run the crew and store the result
  43.     result = outline_crew.kickoff()
  45.     # Parse the outline (in a real app, you might use a more robust parsing approach)
  46.     import json
  47.     try:
  48.         self.state.outline = json.loads(result.raw)
  49.     except:
  50.         # Fallback if not valid JSON
  51.         self.state.outline = {"sections": result.raw}
  53.     return "Outline created"
  55. @listen(create_outline)
  56. def write_content(self, _):
  57.     # Use a writing crew to create the content
  58.     writer = Agent(
  59.         role="Content Writer",
  60.         goal=f"Write engaging content for {self.state.target_audience}",
  61.         backstory="You are a skilled writer who creates compelling content."
  62.     )
  64.     editor = Agent(
  65.         role="Content Editor",
  66.         goal="Ensure content is polished, accurate, and engaging",
  67.         backstory="You have a keen eye for detail and a talent for improving content."
  68.     )
  70.     writing_task = Task(
  71.         description=f"Write a {self.state.content_type} about {self.state.topic} following this outline: {self.state.outline}",
  72.         expected_output="Complete draft content in markdown format",
  73.         agent=writer
  74.     )
  76.     editing_task = Task(
  77.         description="Edit and improve the draft content for clarity, engagement, and accuracy",
  78.         expected_output="Polished final content in markdown format",
  79.         agent=editor,
  80.         context=[writing_task]
  81.     )
  83.     writing_crew = Crew(
  84.         agents=[writer, editor],
  85.         tasks=[writing_task, editing_task],
  86.         process=Process.sequential,
  87.         verbose=True
  88.     )
  90.     # Run the crew and store the result
  91.     result = writing_crew.kickoff()
  92.     self.state.final_content = result.raw
  94.     return "Content created"
  96. @listen(write_content)
  97. def optimize_for_seo(self, _):
  98.     # Use a direct LLM call for SEO optimization
  99.     from crewai import LLM
  100.     llm = LLM(model="openai/gpt-4o-mini")
  102.     prompt = f"""
  103.     Analyze this content for SEO effectiveness for the keyword "{self.state.topic}".
  104.     Rate it on a scale of 1-100 and provide 3 specific recommendations for improvement.
  106.     Content: {self.state.final_content[:1000]}... (truncated for brevity)
  108.     Format your response as JSON with the following structure:
  109.     {{
  110.         "score": 85,
  111.         "recommendations": [
  112.             "Recommendation 1",
  113.             "Recommendation 2",
  114.             "Recommendation 3"
  115.         ]
  116.     }}
  117.     """
  119.     seo_analysis = llm.call(prompt)
  121.     # Parse the SEO analysis
  122.     import json
  123.     try:
  124.         analysis = json.loads(seo_analysis)
  125.         self.state.seo_score = analysis.get("score", 0)
  126.         return analysis
  127.     except:
  128.         self.state.seo_score = 50
  129.         return {"score": 50, "recommendations": ["Unable to parse SEO analysis"]}

Run the flow

content_flow = ContentProductionFlow() result = content_flow.kickoff() ```

实用评估框架

要为你的具体用例确定最合适的方法，可以按照以下逐步评估框架进行：

第 1 步：评估复杂度

从 1 到 10 为你的应用复杂度打分，考虑以下因素：

  1. 步骤数量：需要多少个不同操作？

    • 1 - 3 个步骤：低复杂度（ 1 - 3 ）
    • 4 - 7 个步骤：中等复杂度（ 4 - 7 ）
    • 8 个及以上步骤：高复杂度（ 8 - 10 ）

  2. 相互依赖性：不同部分之间有多强的关联？

    • 依赖很少：低复杂度（ 1 - 3 ）
    • 有一些依赖：中等复杂度（ 4 - 7 ）
    • 有许多复杂依赖：高复杂度（ 8 - 10 ）

  3. 条件逻辑：需要多少分支与决策？

    • 线性流程：低复杂度（ 1 - 3 ）
    • 有一些分支：中等复杂度（ 4 - 7 ）
    • 复杂决策树：高复杂度（ 8 - 10 ）

  4. 领域知识：所需知识有多专业？

    • 通用知识：低复杂度（ 1 - 3 ）
    • 有一些专业知识：中等复杂度（ 4 - 7 ）
    • 需要多个领域的深度专业知识：高复杂度（ 8 - 10 ）

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

第 2 步：评估精确度要求

从 1 到 10 为你的精确度要求打分，考虑以下因素：

  1. 输出结构：输出需要多严格的结构？

    • 自由文本：低精确度（ 1 - 3 ）
    • 半结构化：中等精确度（ 4 - 7 ）
    • 严格格式化（ JSON、XML ）：高精确度（ 8 - 10 ）

  2. 准确性需求：事实准确性有多重要？

    • 创意内容：低精确度（ 1 - 3 ）
    • 信息类内容：中等精确度（ 4 - 7 ）
    • 关键信息：高精确度（ 8 - 10 ）

  3. 可复现性：多次运行之间结果需要多一致？

    • 可以接受变化：低精确度（ 1 - 3 ）
    • 需要一定一致性：中等精确度（ 4 - 7 ）
    • 需要完全可复现：高精确度（ 8 - 10 ）

  4. 容错性：错误造成的影响有多大？

    • 影响较小：低精确度（ 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 ）：由 Flows 编排的 Crews

第 4 步：考虑其他因素

除了复杂度与精确度之外，还应考虑：

  1. 开发时间：Crews 通常更容易快速做出原型
  2. 维护需求：Flows 更利于长期维护
  3. 团队专长：考虑团队对不同方法的熟悉程度
  4. 可扩展性要求：对于复杂应用，Flows 通常更容易扩展
  5. 集成需求：考虑解决方案如何与现有系统集成

结论

在 Crews 与 Flows 之间做选择，或将两者结合使用，是一项关键的架构决策，它会影响你的 CrewAI 应用的有效性、可维护性与可扩展性。通过从复杂度与精确度两个维度评估你的用例，你就能够做出更符合实际需求的决策。

请记住，最佳方法往往会随着应用成熟而不断演进。从能够满足当前需求的最简单方案开始，并在你积累经验、需求逐渐明确后，持续优化你的架构。

现在，你已经拥有了一套评估 CrewAI 用例并基于复杂度与精确度要求选择合适方法的框架。这将帮助你构建更高效、更易维护、也更具扩展性的 AI 应用。