什么是 crewAI？

crewAI 是一个先进的多代理自动化框架，旨在释放多代理自动化的真正潜力，提供最佳的速度与智能组合。该框架允许开发者创建能够协同工作的 AI 代理（Agent），使这些代理像团队一样处理复杂任务。

资料来源：README.md:1-15

核心概念

crewAI 的架构围绕以下几个核心概念构建：

资料来源：lib/cli/src/crewai_cli/templates/AGENTS.md:30-50

crewAI 架构图

graph TD
    A[用户] --> B[Crew 团队]
    B --> C[Agent 1: Researcher]
    B --> D[Agent 2: Writer]
    B --> E[Agent N: ...]
    C --> F[Task 1: 研究任务]
    D --> G[Task 2: 写作任务]
    F --> H[结果输出]
    G --> H

安装与配置

环境要求

crewAI 项目对环境有以下要求：

资料来源：lib/cli/src/crewai_cli/templates/tool/README.md:8-10

安装步骤

1. 安装 UV 包管理器（如果尚未安装）：

pip install uv

1. 创建 crewAI 项目：

crewai create crew <project_name> --skip_provider

1. 安装项目依赖：

crewai install

资料来源：lib/cli/src/crewai_cli/templates/tool/README.md:15-22

创建第一个 Crew

项目结构

crewAI 项目通常包含以下结构：

定义 Agent

crewAI 使用 @agent 装饰器来定义代理，代理可以从 YAML 配置文件中读取配置：

from crewai import Agent
from crewai.project import CrewBase, agent
from crewai_tools import SerperDevTool

@CrewBase
class ResearchCrew():
    """研究团队"""

    agents_config = "config/agents.yaml"
    tasks_config = "config/tasks.yaml"

    @agent
    def researcher(self) -> Agent:
        return Agent(
            config=self.agents_config["researcher"],
            tools=[SerperDevTool()],
            verbose=True,
        )

    @agent
    def writer(self) -> Agent:
        return Agent(
            config=self.agents_config["writer"],
            verbose=True,
        )

资料来源：lib/cli/src/crewai_cli/templates/AGENTS.md:35-60

定义 Task

任务定义使用 @task 装饰器：

from crewai import Task
from crewai.project import CrewBase, task

@CrewBase
class ResearchCrew():
    # ... agent 定义 ...

    @task
    def research_task(self) -> Task:
        return Task(
            config=self.tasks_config["research_task"],
        )

    @task
    def writing_task(self) -> Task:
        return Task(
            config=self.tasks_config["writing_task"],
            output_file='output/article.md'
        )

资料来源：lib/cli/src/crewai_cli/templates/AGENTS.md:65-80

YAML 配置示例

agents.yaml 示例：

researcher:
  role: Senior Research Analyst
  goal: Uncover groundbreaking technologies in {topic}
  backstory: |
    You are a Senior Research Analyst at a leading tech think tank.
    Your expertise lies in identifying emerging trends and technologies.
  verbose: true

writer:
  role: Content Writer
  goal: Write engaging article about {topic}
  backstory: |
    You are a skilled content writer known for your ability to explain complex topics.
  verbose: true

tasks.yaml 示例：

research_task:
  description: >
    Conduct a thorough research on {topic}.
    Identify key trends, notable companies, and potential impact.
  expected_output: >
    A comprehensive research report with main topics, each with full section of information.
  agent: researcher

writing_task:
  description: >
    Write an article based on the research findings about {topic}.
  expected_output: >
    A polished 4-paragraph article formatted in markdown.
  agent: writer
  output_file: output/article.md

资料来源：lib/cli/src/crewai_cli/templates/AGENTS.md:5-30

集成工具生态

crewAI 支持丰富的工具生态，以下是常用工具的集成方式：

RagTool - 知识库检索工具

RagTool 支持从多种来源加载内容构建知识库：

from crewai_tools.tools.rag_tool import RagTool

# 从文件加载
rag_tool = RagTool().from_file('path/to/your/file.txt')

# 从目录加载
rag_tool = RagTool().from_directory('path/to/your/directory')

# 从网页加载
rag_tool = RagTool().from_web_page('https://example.com')

资料来源：lib/crewai-tools/src/crewai_tools/tools/rag/README.md:25-35

FileWriterTool - 文件写入工具

FileWriterTool 简化了向文件写入内容的过程，支持创建目录：

from crewai_tools import FileWriterTool

# 初始化工具
file_writer_tool = FileWriterTool()

# 写入内容到指定目录
result = file_writer_tool._run(
    'example.txt', 
    'This is a test content.', 
    'test_directory'
)

资料来源：lib/crewai-tools/src/crewai_tools/tools/file_writer_tool/README.md:20-35

TavilyExtractorTool - 网页内容提取

TavilyExtractorTool 允许从网页提取结构化内容：

from crewai import Agent, Task, Crew
from crewai_tools import TavilyExtractorTool

# 初始化工具
tavily_tool = TavilyExtractorTool()

# 创建使用该工具的代理
extractor_agent = Agent(
    role='Web Content Extractor',
    goal='Extract key information from specified web pages',
    backstory='You are an expert at extracting content from websites.',
    tools=[tavily_tool],
    verbose=True
)

资料来源：lib/crewai-tools/src/crewai_tools/tools/tavily_extractor_tool/README.md:15-40

工具安装方式

# 使用 UV 安装 crewai 及其工具
uv add 'crewai[tools]' tavily-python

# 或使用 pip
pip install 'crewai[tools]' contextual-client

常用命令参考

资料来源：lib/cli/src/crewai_cli/templates/tool/README.md:25-40

Crew 工作流程

graph LR
    A[定义 Agents] --> B[定义 Tasks]
    B --> C[配置 Crew]
    C --> D[kickoff]
    D --> E{Process 执行}
    E -->|顺序| F[Sequential]
    E -->|层级| G[Hierarchical]
    E -->|协作| H[Collaborative]
    F --> I[结果聚合]
    G --> I
    H --> I
    I --> J[输出]

下一步

• 阅读 Agent 设计指南 学习如何配置代理
• 阅读 Task 设计指南 学习如何编写任务描述
• 探索 官方文档 了解更多高级功能

概述

crewAI 是一个多智能体协作框架，旨在通过编排多个 AI Agent 来完成复杂任务。该框架采用模块化架构设计，将核心功能分离到独立模块中，实现关注点分离（Separation of Concerns）原则。

核心设计理念：

• 模块化：各功能组件独立封装，便于维护和扩展
• 可组合性：通过 Crew（团队）概念组合多个 Agent 协同工作
• 流程控制：提供 Flow 机制实现任务编排和状态管理
• 工具扩展：支持自定义工具扩展 Agent 能力

资料来源：lib/crewai/src/crewai/__init__.py

整体架构

graph TD
    subgraph "表现层"
        CLI[CLI 界面]
        API[API 接口]
    end

    subgraph "核心层"
        Crew[Crew 编排器]
        Agent[Agent 智能体]
        Task[Task 任务]
        Flow[Flow 流程]
    end

    subgraph "工具层"
        Tools[Tools 工具集]
        CustomTools[自定义工具]
    end

    subgraph "底层服务"
        LLM[大语言模型]
        Memory[记忆系统]
    end

    CLI --> Crew
    API --> Crew
    Crew --> Agent
    Crew --> Task
    Flow --> Crew
    Agent --> Tools
    Agent --> LLM
    Agent --> Memory

项目结构

代码库组织

核心模块依赖关系

graph LR
    A[crewai core] --> B[Agent]
    A --> C[Crew]
    A --> D[Task]
    A --> E[Flow]
    A --> F[Tools]
    B --> D
    C --> B
    C --> D
    F --> B

核心模块详解

Crew 编排器

Crew 是 crewAI 的核心编排组件，负责管理多个 Agent 和 Task 的协作执行。

主要职责：

• 初始化和管理 Agent 列表
• 协调任务执行顺序
• 管理任务结果汇总
• 处理异常和重试逻辑

资料来源：lib/crewai/src/crewai/crew.py

class Crew:
    def __init__(
        self,
        agents: List[Agent],
        tasks: List[Task],
        process: Process = Process.hierarchical,
        verbose: bool = False,
    ):
        self.agents = agents
        self.tasks = tasks
        self.process = process
        self.verbose = verbose

Agent 智能体

Agent 代表一个具有特定角色的 AI 实体，每个 Agent 拥有：

Task 任务

Task 定义了具体的工作单元，与特定 Agent 关联执行。

Flow 流程控制

Flow 模块提供了高级的流程编排能力，支持状态管理和工作流控制。

资料来源：lib/crewai/src/crewai/flow/flow.py

class Flow:
    def __init__(self):
        self.state: Dict[str, Any] = {}
    
    def kickoff(self):
        """启动流程执行"""
        pass
    
    def crew(self):
        """注册 Crew 到 Flow"""
        pass

工具系统架构

工具层设计

graph TD
    Tools[Tools 基类] --> FileTools[文件工具]
    Tools --> SearchTools[搜索工具]
    Tools --> APITools[API 工具]
    Tools --> BrowserTools[浏览器工具]
    
    FileTools --> ReadTool[读取工具]
    FileTools --> WriteTool[写入工具]

工具注册机制

crewAI 采用基于装饰器的工具注册模式，工具继承 BaseTool 基类并实现 _run 方法。

资料来源：lib/crewai-tools/pyproject.toml

CLI 模块设计

CLI 模块提供命令行交互入口，支持项目管理、任务创建、执行监控等功能。

资料来源：lib/cli/pyproject.toml

包依赖配置

核心包依赖

# lib/crewai/pyproject.toml
[project]
name = "crewai"
version = "0.1.0"
dependencies = [
    "pydantic>=2.0.0",
    "langchain-core>=0.1.0",
    "langchain-openai>=0.0.5",
]

工具包依赖

# lib/crewai-tools/pyproject.toml
[project]
name = "crewai-tools"
dependencies = [
    "crewai>=0.1.0",
    "langchain-core>=0.1.0",
]

执行流程

Crew 执行流程

graph TD
    A[初始化 Crew] --> B[加载 Agents]
    B --> C[加载 Tasks]
    C --> D{执行模式}
    D -->|顺序| E[按序执行 Task]
    D -->|层级| F[Manager Agent 分配]
    E --> G[汇总结果]
    F --> G
    G --> H[返回最终输出]

Flow 与 Crew 集成

graph LR
    A[Flow.kickoff] --> B[初始化 State]
    B --> C[执行 crew]
    C --> D[Agent 执行任务]
    D --> E[更新 State]
    E --> F{是否完成}
    F -->|否| D
    F -->|是| G[返回结果]

配置选项

Crew 配置参数

Agent 配置参数

模块间通信

sequenceDiagram
    participant User
    participant CLI
    participant Crew
    participant Agent
    participant Tools
    participant LLM
    
    User->>CLI: 发起请求
    CLI->>Crew: 创建 Crew 实例
    Crew->>Agent: 分配任务
    Agent->>Tools: 调用工具
    Tools-->>Agent: 返回结果
    Agent->>LLM: 请求推理
    LLM-->>Agent: 返回响应
    Agent-->>Crew: 汇报结果
    Crew-->>CLI: 汇总输出
    CLI-->>User: 显示结果

扩展机制

自定义工具开发

开发者可通过继承 BaseTool 类创建自定义工具：

from crewai.tools import BaseTool

class MyCustomTool(BaseTool):
    name: str = "my_custom_tool"
    description: str = "工具描述"
    
    def _run(self, *args, **kwargs):
        # 工具实现逻辑
        return result

Flow 自定义

通过继承 Flow 类并重写钩子方法实现自定义流程：

版本与依赖管理

crewAI 采用 PEP 440 版本规范，核心包与扩展包版本需保持兼容：

资料来源：lib/crewai/pyproject.toml, lib/crewai-tools/pyproject.toml, lib/cli/pyproject.toml

概述

Agent（智能体）是 CrewAI 框架的核心构建单元，代表能够自主执行任务的人工智能角色。每个 Agent 具有明确的角色（role）、目标（goal）和背景故事（backstory），并可配备各种工具（Tools）来扩展其能力。

Agent 系统的主要职责包括：

• 理解并执行分配的任务
• 与其他 Agent 协作完成复杂目标
• 访问和利用各种工具获取信息或执行操作
• 通过大语言模型（LLM）进行推理和决策

Agent 核心组件

架构概览

graph TD
    A[Agent] --> B[BaseAgent]
    A --> C[Agent Config]
    B --> D[CrewAgentExecutor]
    B --> E[StepExecutor]
    D --> F[LLM]
    D --> G[Tools]
    E --> H[Memory]
    E --> I[Guardrails]

核心类结构

Agent 配置

YAML 配置格式

Agent 通过 agents.yaml 配置文件定义，支持以下核心属性：

# lib/cli/src/crewai_cli/templates/crew/config/agents.yaml
researcher:
  role: >
    {topic} Senior Data Researcher
  goal: >
    Uncover cutting-edge developments in {topic}
  backstory: >
    You're a seasoned researcher with a knack for uncovering the latest
    developments in {topic}. Known for your ability to find the most relevant
    information and present it in a clear and concise manner.

配置参数说明

Agent 创建与使用

使用装饰器创建 Agent

CrewAI 推荐使用 @agent 装饰器在 Crew 类中定义 Agent：

# lib/cli/src/crewai_cli/templates/crew/crew.py
from crewai import Agent
from crewai.project import CrewBase, agent

@CrewBase
class LatestAiDevelopmentCrew():
    agents: List[BaseAgent]

    @agent
    def researcher(self) -> Agent:
        return Agent(
            config=self.agents_config['researcher'],
            verbose=True,
            tools=[SerperDevTool()]
        )

直接实例化 Agent

也可以直接实例化 Agent 类：

from crewai import Agent

researcher = Agent(
    role="Information Researcher",
    goal="Fetch relevant results from search engines.",
    backstory="An expert in online information retrieval...",
    tools=[search_tool],
    verbose=True
)

Agent 工具系统

内置工具

Agent 可使用多种内置工具来扩展能力：

工具配置示例

# 自定义 LLM 和 embeddings 配置
tool = CodeDocsSearchTool(
    config=dict(
        llm=dict(
            provider="ollama",
            config=dict(
                model="llama2",
            ),
        ),
        embedder=dict(
            provider="google",
            config=dict(
                model="models/embedding-001",
                task_type="retrieval_document",
            ),
        ),
    )
)

Agent 执行流程

执行器工作流程

graph TD
    A[Task 任务] --> B[CrewAgentExecutor]
    B --> C{是否需要工具?}
    C -->|是| D[调用 Tool]
    C -->|否| E[直接 LLM 响应]
    D --> F[处理 Tool 结果]
    E --> G[StepExecutor 单步执行]
    F --> G
    G --> H[更新 Memory]
    H --> I[返回结果]

单步执行机制

StepExecutor 负责处理 Agent 的单个执行步骤，包括：

1. 解析当前任务上下文
2. 生成 LLM 调用
3. 处理工具调用请求
4. 解析执行结果
5. 更新 Agent 记忆

Agent 与 Crew 协作

Crew 组合架构

graph LR
    A[Crew] --> B[Agent 1]
    A --> C[Agent 2]
    A --> D[Agent N]
    B --> E[Task 1]
    C --> F[Task 2]
    D --> G[Task N]
    E --> H[Process 流程]
    F --> H
    G --> H

任务委派

Agent 可以将任务委派给其他 Agent：

@agent
def manager(self) -> Agent:
    return Agent(
        config=self.agents_config['manager'],
        verbose=True,
        allow_delegation=True  # 允许委派任务
    )

高级配置

内存系统

Agent 配备内存系统以维护对话上下文：

• 短期记忆：当前会话内的上下文信息
• 长期记忆：跨会话积累的知识
• 实体记忆：关于特定实体的详细信息

Guardrails 安全机制

Agent 支持配置安全护栏（Guardrails），确保输出符合预期：

from crewai import Agent
from crewai.agents import Guardrail

agent = Agent(
    role="Data Analyst",
    goal="Analyze data accurately",
    backstory="Expert in data analysis",
    guardrails=[custom_guardrail]
)

环境配置

必要环境变量

运行 Agent 前需要配置以下环境变量：

依赖安装

# 使用 uv 管理依赖
uv sync

# 或使用 crewai CLI
crewai install

最佳实践

1. 角色设计

• 角色名称应清晰表达 Agent 的专业领域
• 目标描述应具体、可衡量
• 背景故事应与任务需求匹配

2. 工具选择

• 只分配与任务相关的工具
• 避免过多工具导致决策混乱
• 正确配置工具 API 密钥

3. 流程编排

• sequential（顺序）流程：任务有明确依赖关系
• hierarchical（层级）流程：需要管理角色协调

测试与调试

运行测试

# 默认测试（2 次迭代，使用 gpt-4o-mini）
crewai test

# 自定义测试
crewai test -n 5 -m gpt-4o

查看输出

# 查看最新任务输出
crewai log-tasks-outputs

# 重放特定任务
crewai replay -t <task_id>

总结

Agent 智能体系统是 CrewAI 框架的核心，通过结合角色定义、目标设定、工具扩展和记忆机制，实现了灵活且强大的多智能体协作能力。开发者可以通过 YAML 配置或 Python 代码定义 Agent，并将其组合成 Crew 来完成复杂任务。

相关资源

• 官方文档
• GitHub 仓库
• Discord 社区

概述

Crew 是 CrewAI 框架中的核心编排单元，用于组织和管理多个 AI 智能体（Agent）协同完成复杂任务。Crew 本质上是一个任务编排容器，它将多个具有特定角色的智能体组合在一起，通过预定义的流程模式协调它们之间的工作。

Crew 的设计理念是将自治性（Autonomy）和精确性（Precision）相结合，使得开发者能够灵活地构建从简单任务到企业级自动化的各类应用场景。

核心架构

Crew 与 Agent 的关系

graph TD
    A[Crew 智能体团队] --> B[Agent 1: 研究员]
    A --> C[Agent 2: 分析员]
    A --> D[Agent 3: 报告员]
    A --> E[Task 任务列表]
    A --> F[Process 进程模式]
    B -.-> E
    C -.-> E
    D -.-> E
    E -.-> F

一个 Crew 由以下核心元素组成：

Crew 的创建方式

使用装饰器模式（CrewBase）

CrewAI 推荐使用 @CrewBase 装饰器来定义 Crew 类，这种方式将配置与代码分离，便于维护和修改。

from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
from typing import List

@CrewBase
class MyCrew():
    """我的智能体团队"""
    agents: List[BaseAgent]
    tasks: List[Task]

    @agent
    def researcher(self) -> Agent:
        return Agent(
            config=self.agents_config['researcher'],
            verbose=True
        )

    @task
    def research_task(self) -> Task:
        return Task(
            config=self.tasks_config['research_task'],
        )

    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,
            tasks=self.tasks,
            process=Process.sequential,
            verbose=True,
        )

资料来源：lib/cli/src/crewai_cli/templates/crew/crew.py:1-45

直接实例化

也可以不使用装饰器，直接通过代码创建 Crew：

from crewai import Agent, Crew, Process, Task

researcher = Agent(role="研究员", goal="研究最新AI发展", backstory="...")
task1 = Task(description="研究任务", agent=researcher)

crew = Crew(
    agents=[researcher],
    tasks=[task1],
    process=Process.sequential,
    verbose=True
)

进程模式（Process）

Crew 支持两种主要的进程模式，用于控制智能体执行任务的顺序和方式。

顺序执行（Sequential）

顺序执行模式按照任务列表的定义顺序依次执行任务，每个任务完成后才启动下一个任务。

crew = Crew(
    agents=self.agents,
    tasks=self.tasks,
    process=Process.sequential,
    verbose=True,
)

分层执行（Hierarchical）

分层执行模式会自动分配一个管理员角色，负责协调和验证任务的规划与执行结果。

crew = Crew(
    agents=self.agents,
    tasks=self.tasks,
    process=Process.hierarchical,
    verbose=True,
    manager_agent=manager  # 可选：自定义管理员
)

进程模式对比

资料来源：lib/crewai/src/crewai/process.py

Crew 配置参数

主要参数说明

完整配置示例

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[task1, task2, task3],
    process=Process.sequential,
    verbose=True,
    memory=True,
    max_rpm=60,
    respect_context_window=True,
    config={
        "workflow": "research_report",
        "output_format": "markdown"
    }
)

资料来源：lib/crewai/src/crewai/crew.py

Crew 的生命周期

graph TD
    A[创建 Crew 实例] --> B[初始化 Agents 和 Tasks]
    B --> C[调用 kickoff 方法]
    C --> D{进程模式}
    D -->|Sequential| E[按顺序执行任务]
    D -->|Hierarchical| F[管理员协调任务]
    E --> G[收集任务输出]
    F --> G
    G --> H[生成 CrewOutput]
    H --> I[返回最终结果]

执行入口

Crew 的主入口方法是 kickoff()，它接收输入参数并启动整个团队的工作：

inputs = {
    'topic': 'AI Agents',
    'format': 'markdown'
}

result = my_crew.kickoff(inputs=inputs)

输出处理

Crew 执行完成后会返回 CrewOutput 对象，包含所有任务的结果和元数据：

class CrewOutput(BaseOutput):
    tasks_output: List[TaskOutput]
    token_usage: Dict[str, Any]
    raw: Optional[str]

资料来源：lib/crewai/src/crewai/crews/crew_output.py

记忆系统（Memory）

Crew 支持可选的记忆系统，用于实现跨会话学习和上下文保持。

记忆类型

使用记忆

crew = Crew(
    agents=self.agents,
    tasks=self.tasks,
    memory=True,  # 启用记忆系统
    verbose=True
)

工具集成

工具处理器

每个 Agent 可以配备多个工具，工具通过 ToolsHandler 进行管理和调用：

graph LR
    A[Agent] --> B[ToolsHandler]
    B --> C[Tool 1]
    B --> D[Tool 2]
    B --> E[Tool N]

内置工具示例

CrewAI 提供了丰富的内置工具：

自定义工具

开发者可以创建自定义工具继承 BaseTool 类：

from crewai.tools import BaseTool

class MyCustomTool(BaseTool):
    name: str = "my_tool"
    description: str = "执行自定义任务"

    def _run(self, **kwargs):
        # 工具逻辑
        return result

资料来源：lib/crewai/src/crewai/agents/tools_handler.py

最佳实践

1. YAML 配置优先

推荐使用 YAML 文件定义 agents 和 tasks，保持代码简洁：

src/my_project/
├── config/
│   ├── agents.yaml
│   └── tasks.yaml
├── crew.py
└── main.py

2. 结构化输出

对于需要跨任务传递的数据，使用 output_pydantic 定义数据结构：

Task(
    config=self.tasks_config['task_name'],
    output_pydantic=ReportModel  # 定义输出结构
)

3. 速率限制

生产环境中设置合理的 max_rpm 避免 API 限流：

crew = Crew(
    agents=self.agents,
    tasks=self.tasks,
    max_rpm=60  # 每分钟最多60次请求
)

4. 上下文窗口管理

启用 respect_context_window=True 自动处理 token 限制：

crew = Crew(
    agents=self.agents,
    tasks=self.tasks,
    respect_context_window=True
)

错误处理与调试

日志输出

使用 verbose=True 查看详细执行日志：

crewai run  # 运行并显示详细输出
crewai log-tasks-outputs  # 查看任务输出

重放功能

使用 replay 命令从特定任务重新执行：

crewai replay -t <task_id>

常见问题排查

总结

Crew 是 CrewAI 框架中用于组织多智能体协作的核心组件。通过合理配置 agents、tasks 和 process 模式，开发者可以构建从简单到复杂的各类自动化工作流。Crew 提供了灵活的架构，支持顺序执行、分层协调、记忆系统和丰富的工具集成，使其能够适应各种企业级应用场景。

关键要点：

• 使用 @CrewBase 装饰器实现声明式配置
• 根据工作流特性选择合适的进程模式
• 合理使用记忆系统和速率限制提升性能
• 利用结构化输出确保任务间数据传递的准确性

概述

Flow 是 CrewAI 框架中的事件驱动工作流编排模块，专为复杂的多 Crew 协调场景设计。与 Crew 的自主性（Autonomy）定位不同，Flow 提供精确的（Precise）工作流控制能力，允许开发者通过事件机制实现复杂的任务管道编排。

资料来源：lib/cli/src/crewai_cli/templates/AGENTS.md

Flow 的核心设计理念包括：

• YAML 优先配置：通过 YAML 文件定义 agents 和 tasks，保持 Crew 类的简洁性
• 结构化流状态：使用 Pydantic 模型替代非结构化字典，确保类型安全
• 事件驱动架构：通过 @start、@listen、@router 装饰器实现复杂的事件路由逻辑

资料来源：lib/cli/src/crewai_cli/templates/AGENTS.md

核心组件

Flow 模块主要由以下组件构成，它们共同实现了事件驱动的工作流编排能力：

Flow 主类

Flow 类是整个事件驱动工作流的核心，负责：

• 管理事件监听器注册
• 分发事件到对应的处理器
• 协调多个 Crew 的执行顺序

class Flow:
    def __init__(self):
        self._event_bus = EventBus()
        self._listeners = []
    
    def on(self, event_type):
        """注册事件监听器"""
        def decorator(func):
            self._listeners.append((event_type, func))
            return func
        return decorator
    
    def emit(self, event):
        """触发事件"""
        self._event_bus.emit(event)

资料来源：lib/crewai/src/crewai/flow/flow.py

FlowContext 上下文管理

FlowContext 提供了流程内的状态共享机制，允许不同的事件处理器之间传递数据：

class FlowContext:
    def __init__(self, initial_state=None):
        self._state = initial_state or {}
    
    def set(self, key, value):
        """设置状态值"""
        self._state[key] = value
    
    def get(self, key, default=None):
        """获取状态值"""
        return self._state.get(key, default)
    
    def to_dict(self):
        """导出为字典"""
        return self._state.copy()

资料来源：lib/crewai/src/crewai/flow/flow_context.py

FlowConfig 配置管理

FlowConfig 负责管理工作流的配置信息，包括事件路由规则、超时设置等：

class FlowConfig:
    def __init__(
        self,
        max_retries: int = 3,
        timeout: int = 3600,
        enable_persistence: bool = True
    ):
        self.max_retries = max_retries
        self.timeout = timeout
        self.enable_persistence = enable_persistence

资料来源：lib/crewai/src/crewai/flow/flow_config.py

事件驱动架构

事件总线模式

Flow 采用事件总线（Event Bus）模式进行事件分发，所有事件通过中央总线进行路由：

graph TD
    A[事件源] --> B[EventBus 事件总线]
    B --> C[监听器 A]
    B --> D[监听器 B]
    B --> E[监听器 C]
    C --> F[事件处理]
    D --> G[事件处理]
    E --> H[事件处理]

事件总线的核心职责包括：

• 维护事件类型与监听器的映射关系
• 按顺序触发匹配的监听器
• 处理事件传播中的异常

资料来源：lib/crewai/src/crewai/flow/flow.py

事件类型分类

Flow 支持多种事件类型，覆盖完整的工作流生命周期：

资料来源：lib/cli/src/crewai_cli/templates/AGENTS.md

事件监听装饰器

Flow 提供了三个核心装饰器用于定义事件处理器：

from crewai.flow.flow import Flow, start, listen, router

class MyFlow(Flow):
    @start()
    def begin(self):
        """流程启动事件"""
        return {"action": "start_research"}
    
    @listen("research_complete")
    def process_results(self, event):
        """监听研究完成事件"""
        return {"action": "analyze"}
    
    @router()
    def route_by_quality(self, event):
        """条件路由"""
        if event.get("quality_score", 0) > 0.8:
            return "high_quality_path"
        return "standard_path"

资料来源：lib/cli/src/crewai_cli/templates/AGENTS.md

持久化层

SQLite 持久化

Flow 内置了基于 SQLite 的持久化方案，用于保存工作流执行状态：

class SQLitePersistence:
    def __init__(self, db_path: str = "crewai_flow.db"):
        self.db_path = db_path
    
    def save_state(self, flow_id: str, state: dict):
        """保存流程状态"""
        pass
    
    def load_state(self, flow_id: str) -> dict:
        """加载流程状态"""
        pass
    
    def save_event(self, event: dict):
        """保存事件历史"""
        pass

资料来源：lib/crewai/src/crewai/flow/persistence/sqlite.py

持久化装饰器

提供了用于标记持久化行为的装饰器：

@persistence
class MyFlow(Flow):
    """启用自动持久化的 Flow"""
    pass

资料来源：lib/crewai/src/crewai/flow/persistence/decorators.py

持久化配置

持久化支持以下配置选项：

使用模式

顺序执行模式

适用于线性工作流，下一个任务依赖上一个任务的结果：

graph LR
    A[开始] --> B[任务 1]
    B --> C[任务 2]
    C --> D[任务 3]
    D --> E[结束]

并行执行模式

多个任务可以同时执行，提高效率：

graph TD
    A[开始] --> B[任务 A]
    A --> C[任务 B]
    A --> D[任务 C]
    B --> E[汇合]
    C --> E
    D --> E
    E --> F[完成]

条件路由模式

根据事件内容动态选择处理路径：

graph TD
    A[事件] --> B{条件判断}
    B -->|条件 A| C[处理 A]
    B -->|条件 B| D[处理 B]
    B -->|条件 C| E[处理 C]
    C --> F[结束]
    D --> F
    E --> F

完整示例

项目结构

my_flow_project/
├── src/
│   └── my_flow/
│       ├── __init__.py
│       ├── crew.py           # Crew 定义
│       ├── flow.py          # Flow 定义
│       └── main.py          # 入口文件
└── config/
    ├── agents.yaml
    └── tasks.yaml

Flow 定义示例

# src/my_flow/flow.py
from crewai.flow.flow import Flow
from crewai import Crew, Agent, Task
from crewai.flow.flow_context import FlowContext
from pydantic import BaseModel

class FlowState(BaseModel):
    research_results: str = ""
    analysis_results: str = ""
    final_report: str = ""

class ResearchFlow(Flow):
    def __init__(self):
        super().__init__()
        self.state = FlowContext()
    
    def initialize_crew(self):
        """初始化 Crew"""
        researcher = Agent(
            role="Researcher",
            goal="Research the topic thoroughly",
            backstory="Expert researcher"
        )
        analyzer = Agent(
            role="Analyzer",
            goal="Analyze research findings",
            backstory="Expert analyst"
        )
        
        research_task = Task(
            description="Research {topic}",
            expected_output="Comprehensive research report",
            agent=researcher
        )
        
        analysis_task = Task(
            description="Analyze research results",
            expected_output="Analysis summary",
            agent=analyzer
        )
        
        return Crew(
            agents=[researcher, analyzer],
            tasks=[research_task, analysis_task],
            process="sequential"
        )
    
    def crew_callback(self, crew, output):
        """Crew 执行回调"""
        self.state.set("last_output", output.raw)
        return output

# 导出 kickoff 方法供 CLI 调用
def kickoff():
    flow = ResearchFlow()
    result = flow.kickoff()
    return result

资料来源：lib/cli/src/crewai_cli/templates/flow/main.py

入口文件

# src/my_flow/main.py
from crewai import Crew
from crewai.flow.flow import Flow
from crewai.project import CrewBase, agent, crew, task

@CrewBase
class MyFlowCrew():
    """Flow 内部使用的 Crew"""
    agents: list
    tasks: list
    
    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,
            tasks=self.tasks,
            verbose=True
        )

def run_flow(topic: str):
    """运行 Flow"""
    flow = MyFlowCrew()
    result = flow.kickoff(inputs={"topic": topic})
    return result

if __name__ == "__main__":
    result = run_flow("AI Agents")
    print(result)

CLI 命令

Flow 项目支持以下命令行操作：

# 创建 Flow 项目
crewai create flow my_flow --skip_provider

# 安装依赖
crewai install

# 运行 Flow
crewai run

资料来源：lib/cli/src/crewai_cli/templates/AGENTS.md

最佳实践

结构化状态管理

始终使用 Pydantic 模型定义 Flow 状态，而非普通的字典：

from pydantic import BaseModel

class FlowState(BaseModel):
    topic: str
    research_data: dict = {}
    analysis_result: str = ""
    report_path: str = ""

这样可以获得更好的类型检查和 IDE 支持。

事件命名规范

使用清晰的事件命名约定，便于追踪和调试：

• crew_kickoff_started: Crew 开始执行
• crew_kickoff_completed: Crew 执行完成
• task_completed: 单个任务完成
• error_occurred: 错误发生

异常处理

在事件处理器中添加适当的异常处理：

@listen("external_event")
def handle_external(self, event):
    try:
        result = process_event(event)
        return {"status": "success", "data": result}
    except Exception as e:
        return {"status": "error", "message": str(e)}

超时配置

对于可能长时间运行的操作，设置合理的超时时间：

flow_config = FlowConfig(
    timeout=3600,  # 1 小时
    max_retries=3
)

与 Crew 的对比

总结

Flow 事件驱动工作流为 CrewAI 提供了强大的流程编排能力，特别适合以下场景：

• 需要精确控制执行顺序的多步骤流程
• 基于条件进行动态分支的业务逻辑
• 需要保存和恢复执行状态的长时间运行任务
• 需要事件日志和审计追踪的应用

通过结合 Crew 的自主 Agent 能力和 Flow 的精确控制能力，开发者可以构建出既智能又可控的复杂 AI 工作流系统。

概述

Task 任务是 CrewAI 多智能体框架中的核心执行单元。每个 Task 代表一个具体的工作目标，由特定的 Agent 负责执行。Task 系统支持丰富的配置选项，包括输出结构化数据、任务依赖关系、条件执行、输出验证等功能。

Task 系统的主要职责：

• 定义具体的工作任务和预期输出
• 管理 Agent 与 Task 之间的绑定关系
• 处理任务之间的依赖关系和执行顺序
• 支持结构化输出（Pydantic、JSON）
• 提供输出验证和 Guardrails 机制

核心组件

Task 类

Task 是任务系统的核心类，位于 lib/crewai/src/crewai/task.py。

classDiagram
    class Task {
        +str description
        +str expected_output
        +Agent agent
        +List~Task~ dependencies
        +str output_file
        +bool verbose
        +int max_iterations
        +float max_rpm
        +callbacks: List[Callback]
        +async_execute()
        +execute()
        +output: TaskOutput
    }

TaskOutput 类

TaskOutput 类封装任务的执行结果。

资料来源：lib/crewai/src/crewai/tasks/task_output.py

ConditionalTask 类

条件任务允许根据运行时条件决定是否执行任务。

@task
def conditional_research_task(self) -> Task:
    return Task(
        config=self.tasks_config['research_task'],
        condition=lambda context: len(context.get('topics', [])) > 0
    )

资料来源：lib/crewai/src/crewai/tasks/conditional_task.py

Task 属性详解

基础属性

执行控制属性

输出配置属性

资料来源：lib/crewai/src/crewai/task.py

Guardrails 机制

Guardrails 用于验证和约束任务输出，确保输出符合预期标准。

LLM Guardrail

基于 LLM 的输出验证器。

from crewai.tasks.llm_guardrail import LLMGuardrail

guardrail = LLMGuardrail(
    llm=agent.llm,
    threshold=0.7
)

Hallucination Guardrail

检测输出中的幻觉内容（事实性错误）。

from crewai.tasks.hallucination_guardrail import HallucinationGuardrail

guardrail = HallucinationGuardrail(
    llm=agent.llm,
    context="参考文档内容...",
    threshold=8.0
)

资料来源：lib/crewai/src/crewai/tasks/llm_guardrail.py

任务依赖与执行流程

依赖关系管理

graph TD
    A[Task A: 研究任务] --> B[Task B: 分析任务]
    A --> C[Task C: 报告撰写]
    B --> C
    D[Task D: 条件任务] -->|仅当条件满足| E[Task E: 后续任务]

顺序执行流程

在 Process.sequential 模式下，任务按依赖顺序执行：

1. 执行没有依赖的任务
2. 等待所有依赖任务完成
3. 收集依赖任务的输出作为上下文
4. 执行当前任务
5. 重复直到所有任务完成

from crewai import Crew, Process, Task, Agent

researcher = Agent(role="研究员", goal="收集信息", backstory="...")
analyst = Agent(role="分析师", goal="分析数据", backstory="...")

research_task = Task(
    description="研究最新 AI 发展",
    expected_output="AI 发展摘要",
    agent=researcher
)

analysis_task = Task(
    description="分析研究结果",
    expected_output="深度分析报告",
    agent=analyst,
    dependencies=[research_task]  # 依赖于研究任务
)

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

资料来源：lib/crewai/src/crewai/task.py:1-200

YAML 配置

tasks.yaml 配置示例

# lib/cli/src/crewai_cli/templates/crew/config/tasks.yaml

research_task:
  description: >
    深入研究 {topic} 相关信息，收集最新发展和趋势。
    这包括新闻文章、学术论文和行业报告。
  expected_output: >
    一份结构化的研究报告，包含关键发现、
    数据来源和趋势分析。
  agent: researcher

reporting_task:
  description: >
    基于研究报告，创建一份全面的报告
  expected_output: >
    一份完整的报告，包含主要主题，
    每个主题都有详细信息。格式为 markdown，不包含代码块。
  agent: reporting_analyst
  output_file: report.md

条件任务配置

conditional_research_task:
  description: >
    仅当存在特定主题时执行深入研究
  expected_output: >
    针对指定主题的详细研究报告
  agent: researcher
  condition: "topics_exist"  # 条件标识符

资料来源：lib/cli/src/crewai_cli/templates/crew/config/tasks.yaml

编程式创建 Task

基本用法

from crewai import Agent, Task

agent = Agent(
    role="高级数据研究员",
    goal="发现 {topic} 的最新发展",
    backstory="你是一名资深研究员..."
)

task = Task(
    description="深入研究 {topic}",
    expected_output="一份详细的研究报告",
    agent=agent,
    verbose=True
)

结构化输出

from pydantic import BaseModel, Field
from crewai import Task

class ResearchOutput(BaseModel):
    summary: str = Field(description="研究摘要")
    key_findings: list[str] = Field(description="关键发现")
    资料来源：list[str] = Field(description="数据来源")

task = Task(
    description="研究 AI Agents",
    expected_output="结构化的研究结果",
    agent=agent,
    output_pydantic=ResearchOutput
)

带依赖的任务

analysis_task = Task(
    description="分析研究报告",
    expected_output="分析结论",
    agent=analyst,
    dependencies=[research_task]  # 等待研究任务完成
)

输出到文件

report_task = Task(
    description="撰写报告",
    expected_output="完整的报告文档",
    agent=writer,
    output_file="report.md"  # 结果将写入此文件
)

任务执行上下文

上下文传递机制

graph LR
    A[Task A 输出] -->|context| B[Task B 输入]
    B -->|context| C[Task C 输入]
    C -->|最终结果| D[Result]

当一个任务完成后，其输出（TaskOutput）会自动传递给依赖它的任务作为上下文。上下文包含：

• task_output.text: 文本输出
• task_output.pydantic: Pydantic 模型实例
• task_output.json_dict: JSON 字典格式
• task_output.summary: 执行摘要

访问上下文

def analysis_callback(context: dict):
    previous_results = context.get('previous_tasks', [])
    for result in previous_results:
        print(result.text)

task = Task(
    description="基于之前结果继续分析",
    expected_output="深入分析",
    agent=analyst,
    dependencies=[research_task],
    callback=analysis_callback
)

最佳实践

1. 清晰的任务描述

# ✅ 推荐：清晰、具体的描述
task = Task(
    description="从 2024 年 1 月至 6 月的财经新闻中提取关于 AI 投资的信息",
    expected_output="JSON 格式的投资信息列表，包含公司名、投资金额、投资轮次"
)

# ❌ 避免：模糊的描述
task = Task(
    description="查一下新闻"
)

2. 明确的预期输出

task = Task(
    description="分析销售数据",
    expected_output="""
    格式：Markdown 表格
    列：产品名称 | 销量 | 增长率 | 市场份额
    包含趋势分析和改进建议
    """
)

3. 合理的任务粒度

• 单个任务应该完成一个明确的目标
• 避免将多个不相关的任务合并为一个
• 使用依赖关系组织复杂工作流

4. 任务依赖规划

# 正确的依赖设计
task_1 = Task(description="收集原始数据", ...)
task_2 = Task(description="清洗数据", dependencies=[task_1], ...)
task_3 = Task(description="数据分析", dependencies=[task_2], ...)
task_4 = Task(description="生成可视化", dependencies=[task_3], ...)

与 CrewBase 装饰器集成

使用装饰器定义任务

from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, task
from typing import List

@CrewBase
class ResearchCrew():
    agents: List[Agent]
    tasks: List[Task]
    
    @task
    def research_task(self) -> Task:
        return Task(
            config=self.tasks_config['research_task'],
        )

    @task
    def reporting_task(self) -> Task:
        return Task(
            config=self.tasks_config['reporting_task'],
            output_file='report.md'
        )

    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,
            tasks=self.tasks,
            process=Process.sequential,
            verbose=True,
        )

资料来源：lib/crewai/src/crewai/tasks/__init__.py

执行状态与监控

任务状态

stateDiagram-v2
    [*] --> Pending: 创建任务
    Pending --> InProgress: 开始执行
    InProgress --> Completed: 正常完成
    InProgress --> Failed: 执行错误
    InProgress --> Retrying: 重试中
    Failed --> [*]
    Completed --> [*]

迭代控制

task = Task(
    description="复杂分析任务",
    expected_output="分析结果",
    max_iterations=10,  # 最多迭代 10 次
    max_rpm=5           # 每分钟最多 5 次请求
)

Verbose 模式

# 启用详细输出
task = Task(
    description="搜索并分析",
    expected_output="结果摘要",
    verbose=True  # 显示详细执行日志
)

总结

Task 任务系统是 CrewAI 框架的核心组成部分，提供了：

• 灵活的配置：支持 YAML 和编程式两种定义方式
• 结构化输出：通过 Pydantic 模型实现类型安全的输出
• 依赖管理：支持任务的串行和条件执行
• 输出验证：Guardrails 机制确保输出质量
• 上下文传递：任务之间自动共享执行结果

通过合理使用 Task 系统，可以构建复杂的多智能体工作流，实现自动化的问题解决流程。

概述

CrewAI 框架的 LLM（大型语言模型）集成系统为 AI Agent 提供了灵活的多提供者支持。该系统采用模块化架构，支持 OpenAI、Anthropic、Google Gemini、AWS Bedrock、Azure 等主流 LLM 提供者，同时允许通过统一接口自定义配置。

架构设计

系统组件

graph TD
    A[Agent] --> B[LLM 接口层]
    B --> C[BaseLLM 抽象基类]
    C --> D[OpenAI Provider]
    C --> E[Anthropic Provider]
    C --> F[Gemini Provider]
    C --> G[Bedrock Provider]
    C --> H[Azure Provider]
    C --> I[自定义 Provider]

核心文件结构

LLM 配置方式

通过 Agent 配置

在创建 Agent 时直接指定 LLM：

from crewai import Agent
from crewai.llm import LLM

agent = Agent(
    role="研究助手",
    goal="深入研究指定主题",
    backstory="经验丰富的领域专家",
    llm=LLM(model="gpt-4", temperature=0.7)
)

通过 config 字典配置

支持使用配置字典自定义模型和嵌入层：

tool = CodeDocsSearchTool(
    config=dict(
        llm=dict(
            provider="ollama",
            config=dict(
                model="llama2",
                temperature=0.5,
                top_p=1,
                stream=True,
            ),
        ),
        embedder=dict(
            provider="google",
            config=dict(
                model="models/embedding-001",
                task_type="retrieval_document",
            ),
        ),
    )
)

资料来源：lib/crewai-tools/src/crewai_tools/tools/code_docs_search_tool/README.md

支持的 LLM 提供者

提供者配置对照表

OpenAI Provider 实现

# lib/crewai/src/crewai/llms/providers/openai/completion.py
class OpenAICompletion:
    """处理 OpenAI API 调用的提供者实现"""
    
    def __init__(self, model: str, **kwargs):
        self.model = model
        self.api_key = kwargs.get("api_key")
        self.temperature = kwargs.get("temperature", 0.7)

Anthropic Provider 实现

# lib/crewai/src/crewai/llms/providers/anthropic/completion.py
class AnthropicCompletion:
    """Anthropic Claude 系列模型的提供者实现"""
    
    def __init__(self, model: str, **kwargs):
        self.model = model
        self.api_key = kwargs.get("api_key")
        self.max_tokens = kwargs.get("max_tokens", 4096)

Bedrock Provider 实现

支持 AWS Bedrock 上的多种模型，包括 Anthropic Claude 和 Amazon Titan 系列：

# lib/crewai/src/crewai/llms/providers/bedrock/completion.py
class BedrockCompletion:
    """AWS Bedrock 提供者实现"""
    
    def __init__(self, model: str, region: str = "us-east-1", **kwargs):
        self.model = model
        self.region = region
        self.aws_access_key = kwargs.get("aws_access_key_id")
        self.aws_secret_key = kwargs.get("aws_secret_access_key")

Gemini Provider 实现

# lib/crewai/src/crewai/llms/providers/gemini/completion.py
class GeminiCompletion:
    """Google Gemini 模型的提供者实现"""
    
    def __init__(self, model: str, **kwargs):
        self.model = model
        self.api_key = kwargs.get("api_key")

Azure Provider 实现

# lib/crewai/src/crewai/llms/providers/azure/completion.py
class AzureCompletion:
    """Azure OpenAI Service 提供者实现"""
    
    def __init__(self, 
                 deployment_name: str,
                 api_base: str,
                 api_version: str = "2024-02-01",
                 **kwargs):
        self.deployment_name = deployment_name
        self.api_base = api_base
        self.api_version = api_version

BaseLLM 抽象基类

所有 LLM 提供者必须继承自 BaseLLM，实现标准接口：

# lib/crewai/src/crewai/llms/base_llm.py
from abc import ABC, abstractmethod

class BaseLLM(ABC):
    """LLM 提供者抽象基类"""
    
    @abstractmethod
    def call(self, prompt: str, **kwargs) -> str:
        """执行 LLM 调用"""
        pass
    
    @abstractmethod
    def get_model_name(self) -> str:
        """获取当前模型名称"""
        pass
    
    @property
    @abstractmethod
    def supports_system_prompts(self) -> bool:
        """检查提供者是否支持系统提示"""
        pass
    
    @property
    @abstractmethod
    def supports_function_calling(self) -> bool:
        """检查提供者是否支持函数调用"""
        pass

LLM 钩子系统

llm_hooks.py 生命周期钩子

graph LR
    A[LLM 调用开始] --> B[on_llm_new_token]
    B --> C[token 处理中]
    C --> D[on_llm_new_token]
    D --> E[调用完成]
    E --> F[on_llm_end]
    A --> F2[on_llm_start]

钩子类型

自定义钩子实现

# lib/crewai/src/crewai/hooks/llm_hooks.py
class LLMCallbackHandler:
    """LLM 回调处理器基类"""
    
    def on_llm_start(self, serialized, prompts, **kwargs):
        """LLM 调用开始时的钩子"""
        pass
    
    def on_llm_new_token(self, token: str, **kwargs):
        """新 token 生成时的钩子"""
        pass
    
    def on_llm_end(self, response, **kwargs):
        """LLM 调用结束时的钩子"""
        pass
    
    def on_llm_error(self, error, **kwargs):
        """LLM 调用出错时的钩子"""
        pass

LLM 主类接口

llm.py 核心功能

# lib/crewai/src/crewai/llm.py
class LLM:
    """统一的 LLM 接口类"""
    
    def __init__(
        self,
        model: str,
        provider: str = "openai",
        temperature: float = 0.7,
        max_tokens: int | None = None,
        **kwargs
    ):
        """初始化 LLM 实例
        
        Args:
            model: 模型标识符 (如 "gpt-4", "claude-3-opus")
            provider: 提供者名称 (默认 "openai")
            temperature: 生成温度 (0.0-1.0)
            max_tokens: 最大 token 数量限制
            **kwargs: 提供者特定参数
        """
        self.model = model
        self.provider = provider
        self.temperature = temperature
        self.max_tokens = max_tokens
        self._setup_provider(**kwargs)

主要方法

环境变量配置

OpenAI 配置

OPENAI_API_KEY=sk-...
OPENAI_API_BASE=https://api.openai.com/v1  # 可选，自定义端点

Anthropic 配置

ANTHROPIC_API_KEY=sk-ant-...

Google Gemini 配置

GOOGLE_API_KEY=AIza...

AWS Bedrock 配置

AWS_ACCESS_KEY_ID=AKIA...
AWS_SECRET_ACCESS_KEY=...
AWS_REGION=us-east-1

Azure 配置

AZURE_OPENAI_API_KEY=...
AZURE_OPENAI_ENDPOINT=https://xxx.openai.azure.com
AZURE_OPENAI_API_VERSION=2024-02-01

最佳实践

模型选择指南

graph TD
    A[任务类型] --> B{需要函数调用?}
    B -->|是| C{预算有限?|
    C -->|是| D[GPT-4o-mini + function calling]
    C -->|否| E[GPT-4 + function calling]
    B -->|否| F{需要长上下文?|
    F -->|是| G[Claude 3.5 Sonnet]
    F -->|否| H[GPT-4o]

温度参数建议

错误处理模式

from crewai.llm import LLM
from crewai.llms.base_llm import LLMGenerationError

llm = LLM(model="gpt-4", temperature=0.7)

try:
    response = llm.call(
        prompt="解释量子计算原理",
        system_prompt="你是一位物理学家"
    )
except LLMGenerationError as e:
    print(f"LLM 生成失败: {e}")
    # 实现降级逻辑或重试机制

版本检查功能

CrewAI 提供内置版本检查机制：

# lib/crewai/src/crewai/version.py
from crewai.version import (
    check_version,
    get_crewai_version,
    is_newer_version_available,
    get_latest_version_from_pypi,
)

# 检查当前版本
current = get_crewai_version()

# 检查是否有新版本
if is_newer_version_available():
    latest = get_latest_version_from_pypi()
    print(f"有新版本可用: {latest}")

总结

CrewAI 的 LLM 集成系统提供了：

• 多提供者支持：OpenAI、Anthropic、Gemini、Bedrock、Azure 等
• 统一接口：BaseLLM 抽象基类确保一致性
• 灵活配置：支持环境变量和代码配置
• 生命周期钩子：完整的调用监控和扩展能力
• 版本管理：自动检查更新，保持兼容性

此架构使开发者能够轻松切换不同的 LLM 提供者，同时保持应用代码的稳定性和可移植性。

概述

CrewAI 的工具系统是一个模块化的插件架构，允许 AI Agent 通过标准化的接口调用外部工具和资源。该系统支持两种主要的集成方式：内置工具和 MCP（Model Context Protocol）协议集成。通过统一的工具调用机制，Agent 可以执行文件操作、网络搜索、数据库查询、RAG 检索等复杂任务。

工具系统的核心设计理念是将工具定义与工具执行解耦，每个工具通过 Pydantic Schema 定义参数和返回值，确保类型安全和参数验证。

架构设计

工具系统架构图

graph TD
    subgraph "工具层 Tool Layer"
        A[BaseTool] --> B[StructuredTool]
        B --> C[内置工具]
        B --> D[MCP 工具]
    end
    
    subgraph "工具注册层"
        E[ToolRegistry] --> A
        F[MCPToolResolver] --> D
    end
    
    subgraph "Agent 执行层"
        G[Agent] --> H[ToolCalling]
        H --> E
        H --> F
    end
    
    subgraph "外部系统"
        I[MCP Server] --> F
        J[文件系统] --> C
        K[网络服务] --> C
    end

核心组件说明

内置工具详解

BaseTool 基类

BaseTool 是所有工具的基类，提供了工具的基本属性和抽象方法。每个工具必须定义 name（名称）、description（描述）和 _run（执行方法）。

# 资料来源：lib/crewai/src/crewai/tools/base_tool.py
name: str = "tool_name"
description: str = "A tool that..."

FileReadTool 文件读取工具

FileReadTool 用于读取文件内容，支持按行范围读取和指定起始行。

# 初始化工具
tool = FileReadTool(file_path="/path/to/file.txt")

# 读取整个文件
content = tool.run()

# 按行范围读取
content = tool.run(start_line=100, line_count=50)

FileWriterTool 文件写入工具

FileWriterTool 用于将内容写入文件，支持自动创建目录。

# 资料来源：lib/crewai-tools/src/crewai_tools/tools/file_writer_tool/README.md
from crewai_tools import FileWriterTool

file_writer = FileWriterTool()
result = file_writer._run('example.txt', '测试内容', 'test_directory')

RagTool 检索增强生成工具

RagTool 用于从文件、目录或网页加载内容构建知识库，支持 RAG 检索场景。

# 资料来源：lib/crewai-tools/src/crewai_tools/tools/rag/README.md
from crewai_tools.tools.rag_tool import RagTool

# 从文件加载
rag_tool = RagTool().from_file('path/to/file.txt')

# 从目录加载
rag_tool = RagTool().from_directory('path/to/directory')

# 从网页加载
rag_tool = RagTool().from_web_page('https://example.com')

搜索工具集

crewai_tools 提供了多种搜索工具，用于 Agent 执行网络搜索任务。

#### SerplySearchTool

# 资料来源：lib/crewai-tools/src/crewai_tools/tools/serply_api_tool/README.md
from crewai_tools import SerplyWebSearchTool

search_tool = SerplyWebSearchTool()

#### LinkupSearchTool

# 资料来源：lib/crewai-tools/src/crewai_tools/tools/linkup/README.md
from crewai_tools import LinkupSearchTool

linkup_tool = LinkupSearchTool()
response = linkup_tool._run(
    query="Women Nobel Prize Physics",
    depth="standard",
    output_type="searchResults"
)

#### SerplyScholarSearchTool

学术搜索工具，用于检索学术论文和文献。

from crewai_tools import SerplyScholarSearchTool

scholar_tool = SerplyScholarSearchTool()
scholar_tool = SerplyScholarSearchTool(proxy_location="GB")

MCP 协议集成

MCP 概述

MCP（Model Context Protocol）是一种标准化的协议，允许 AI 模型与外部工具和服务进行交互。CrewAI 通过 MCP 集成，可以调用远程 MCP 服务器提供的工具。

MCP 架构图

graph LR
    subgraph "CrewAI 运行时"
        A[Agent] --> B[ToolCalling]
        B --> C[MCPToolResolver]
        C --> D[MCPClient]
    end
    
    subgraph "MCP 协议层"
        D <-->|JSON-RPC| E[MCP Server]
    end
    
    subgraph "外部服务"
        E --> F[文件系统工具]
        E --> G[数据库工具]
        E --> H[API 工具]
    end

MCP 客户端

MCPClient 负责建立和管理与 MCP 服务器的连接，处理协议握手、消息序列化和工具调用路由。

# 资料来源：lib/crewai/src/crewai/mcp/client.py
class MCPClient:
    def __init__(
        self,
        server_url: str,
        api_key: Optional[str] = None,
        timeout: int = 60
    ):
        """初始化 MCP 客户端"""

MCP 工具解析器

MCPToolResolver 负责将 MCP 服务器提供的工具 schema 转换为 CrewAI 可识别的工具格式，并处理工具调用请求。

# 资料来源：lib/crewai/src/crewai/mcp/tool_resolver.py
class MCPToolResolver:
    def resolve_tools(self) -> List[StructuredTool]:
        """解析 MCP 服务器提供的工具列表"""
    
    def call_tool(self, tool_name: str, arguments: dict) -> Any:
        """调用指定工具并返回结果"""

MCP 工具调用流程

sequenceDiagram
    participant Agent
    participant ToolCalling
    participant MCPToolResolver
    participant MCPClient
    participant MCPServer
    
    Agent->>ToolCalling: 请求执行工具
    ToolCalling->>MCPToolResolver: 查找工具定义
    MCPToolResolver->>MCPClient: 发起工具调用
    MCPClient->>MCPServer: 发送 JSON-RPC 请求
    MCPServer-->>MCPClient: 返回执行结果
    MCPClient-->>MCPToolResolver: 解析响应
    MCPToolResolver-->>ToolCalling: 返回结构化结果
    ToolCalling-->>Agent: 返回工具执行结果

工具注册与调用

工具调用机制

ToolCalling 模块负责管理工具的注册、查找和执行流程。

# 资料来源：lib/crewai/src/crewai/tools/tool_calling.py
class ToolCalling:
    def register_tool(self, tool: StructuredTool) -> None:
        """注册新工具"""
    
    def execute_tool(self, tool_name: str, **kwargs) -> Any:
        """执行指定工具"""

StructuredTool 结构化工具

StructuredTool 扩展了 BaseTool，使用 Pydantic Schema 进行参数验证和类型检查。

# 资料来源：lib/crewai/src/crewai/tools/structured_tool.py
from pydantic import BaseModel

class StructuredTool(BaseTool):
    args_schema: type[BaseModel]
    
    def _run(self, **kwargs) -> Any:
        """执行工具逻辑"""

Agent 中使用工具

# 资料来源：lib/cli/src/crewai_cli/templates/AGENTS.md
from crewai import Agent
from crewai_tools import SerperDevTool

researcher = Agent(
    role="Senior Researcher",
    goal="Uncover cutting-edge developments in {topic}",
    backstory="An expert in online information retrieval...",
    tools=[SerperDevTool()],
    verbose=True
)

自定义工具开发

工具开发模板

crewai 提供了标准化的工具开发模板，位于 lib/cli/src/crewai_cli/templates/tool/。

crewai tool publish {{tool_name}}
crewai tool install {{tool_name}}

工具规范定义

工具的元数据和参数规范定义在 tool.specs.json 中。

// 资料来源：lib/crewai-tools/src/crewai_tools/tool.specs.json
{
  "name": "tool_name",
  "description": "Tool description",
  "parameters": {
    "type": "object",
    "properties": {...}
  }
}

工具开发示例

# 创建自定义工具
from crewai.tools.base_tool import BaseTool
from pydantic import BaseModel

class MyToolInput(BaseModel):
    query: str
    limit: int = 10

class MyTool(BaseTool):
    name: str = "my_custom_tool"
    description: str = "A custom tool for..."
    args_schema: type[BaseModel] = MyToolInput
    
    def _run(self, query: str, limit: int = 10) -> str:
        # 工具执行逻辑
        return f"Result for {query}"

工具配置与模型自定义

LLM 和 Embedder 配置

大多数工具支持自定义 LLM 和 Embedder 模型。

# 资料来源：lib/crewai-tools/src/crewai_tools/tools/code_docs_search_tool/README.md
from crewai_tools import CodeDocsSearchTool

tool = CodeDocsSearchTool(
    config=dict(
        llm=dict(
            provider="ollama",
            config=dict(
                model="llama2",
                temperature=0.5,
            ),
        ),
        embedder=dict(
            provider="google",
            config=dict(
                model="models/embedding-001",
                task_type="retrieval_document",
            ),
        ),
    )
)

支持的 LLM 提供商

工作流集成

Crew 中使用工具

# 资料来源：lib/cli/src/crewai_cli/templates/AGENTS.md
from crewai import Agent, Crew, Task
from crewai.project import CrewBase, agent, crew, task

@CrewBase
class ResearchCrew:
    agents_config = "config/agents.yaml"
    tasks_config = "config/tasks.yaml"
    
    @agent
    def researcher(self) -> Agent:
        return Agent(
            config=self.agents_config["researcher"],
            tools=[SerperDevTool()],
            verbose=True,
        )

Task 中配置工具

# 在 tasks.yaml 中配置任务
research_task:
  description: >
    Research the latest developments in {topic}
  expected_output: >
    A comprehensive report with sources and implications.
  agent: researcher
  tools: [search_tool]
  output_file: output/research.md

最佳实践

工具选择策略

1. 任务匹配：根据任务类型选择最合适的工具
2. 性能考虑：优先使用轻量级工具，减少 API 调用开销
3. 可靠性：选择有稳定维护的官方工具
4. 安全性：敏感操作使用本地工具而非远程 API

错误处理

try:
    result = tool._run(param="value")
except ToolExecutionError as e:
    logger.error(f"Tool execution failed: {e}")
    # 降级处理或重试逻辑

工具链组合

多个工具可以组合使用实现复杂工作流：

# 搜索 + 内容转换
from crewai import Agent
from crewai_tools import SerplyWebSearchTool, SerplyWebpageToMarkdownTool

search_tool = SerplyWebSearchTool()
convert_to_markdown = SerplyWebpageToMarkdownTool()

researcher = Agent(
    role='Senior Researcher',
    goal='Uncover groundbreaking technologies in {topic}',
    tools=[search_tool, convert_to_markdown]
)

总结

CrewAI 的工具系统通过模块化的设计，为 AI Agent 提供了强大的外部能力扩展能力。内置工具覆盖了文件操作、网络搜索、RAG 检索等常见场景，而 MCP 协议集成则允许与任意支持 MCP 的外部服务进行交互。通过统一的工具调用接口和 Pydantic 参数校验机制，开发者可以快速构建功能丰富的多 Agent 应用。

概述

在 CrewAI 框架中，MCP (Model Context Protocol) 和 A2A (Agent-to-Agent) 是两个核心的通信协议，分别用于 Agent 与外部工具服务的连接以及 Agent 之间的互操作。这两个协议共同构成了 CrewAI 多智能体系统的通信基础设施，使 Agent 能够与外部系统进行无缝集成和协作。

协议架构总览

graph TB
    subgraph "CrewAI 通信层"
        A1["Agent 1"] --> A2["Agent 2"]
        A1 --> A3["Agent 3"]
        A2 --> A3
    end
    
    subgraph "A2A 协议层"
        A2A["Agent-to-Agent 通信"]
        DEL["委派工具"]
        TYPES["类型定义"]
    end
    
    subgraph "MCP 协议层"
        MCP["Model Context Protocol"]
        STDIO["stdio 传输"]
        SSE["SSE 传输"]
        HTTP["HTTP 传输"]
    end
    
    subgraph "外部服务"
        TOOL["外部工具服务"]
        API["第三方 API"]
    end
    
    A1 -.-> A2A
    A2 -.-> A2A
    A2A -.-> DEL
    MCP -.-> TOOL
    MCP -.-> STDIO
    MCP -.-> SSE
    MCP -.-> HTTP

MCP 协议 (Model Context Protocol)

MCP 协议是 CrewAI 用于与外部工具和服务进行通信的标准协议。它定义了 Agent 如何连接到外部工具、发送请求并接收响应。CrewAI 的 MCP 实现支持多种传输方式，以适应不同的部署场景。

MCP 模块结构

graph TD
    MCP["crewai.mcp 模块"]
    INIT["__init__.py<br/>MCP 包初始化"]
    
    subgraph "传输层 (transports)"
        STDIO["stdio.py<br/>标准输入输出传输"]
        SSE["sse.py<br/>Server-Sent Events 传输"]
        HTTP["http.py<br/>HTTP 请求传输"]
    end
    
    MCP --> INIT
    INIT --> STDIO
    INIT --> SSE
    INIT --> HTTP

传输方式详解

#### stdio 传输

stdio 传输是 MCP 协议最常用的传输方式之一，特别适用于本地进程通信和命令行工具集成。通过标准输入输出流进行数据交换，具有低延迟和简单可靠的特点。

主要特点：

• 同步通信：使用标准输入输出进行同步消息交换
• 进程管理：适合启动和管理本地子进程
• 错误处理：内置进程错误捕获和日志记录机制
• 超时控制：支持请求超时配置

#### SSE 传输

Server-Sent Events (SSE) 传输适用于需要服务器推送场景的 MCP 通信。它允许服务器主动向客户端发送事件流，特别适合长时间运行的任务和实时状态更新。

使用场景：

• 实时日志流推送
• 长时间任务的进度更新
• 异步操作的完成通知

#### HTTP 传输

HTTP 传输是最通用的 MCP 通信方式，适用于与基于 REST API 的外部服务进行交互。它支持完整的 HTTP 协议特性，包括认证、会话管理和请求/响应模式。

MCP 配置参数

A2A 协议 (Agent-to-Agent)

A2A 协议是 CrewAI 框架中用于 Agent 之间进行通信和协作的协议。它定义了 Agent 如何相互发送消息、委派任务、共享状态和协调工作流程。A2A 协议的核心目标是实现多 Agent 系统中的无缝协作。

A2A 核心组件

graph LR
    subgraph "A2A 协议栈"
        A2A_INIT["__init__.py<br/>A2A 包初始化"]
        TYPES["types.py<br/>类型定义与数据模型"]
        DELEG["delegation.py<br/>任务委派工具"]
    end
    
    subgraph "A2A 类型体系"
        REQ["请求类型"]
        RESP["响应类型"]
        TASK["任务类型"]
        STATE["状态类型"]
    end
    
    subgraph "A2A 工具集"
        DEL_TOOL["委派工具"]
        ROUTING["路由工具"]
    end
    
    A2A_INIT --> TYPES
    A2A_INIT --> DELEG
    TYPES --> REQ
    TYPES --> RESP
    TYPES --> TASK
    TYPES --> STATE
    DELEG --> DEL_TOOL
    DELEG --> ROUTING

A2A 类型系统

A2A 协议定义了一套完整的类型系统，用于描述 Agent 通信中的各种数据结构。这些类型包括请求、响应、任务描述、状态信息等。类型系统的设计遵循以下原则：

设计原则：

• 类型安全：使用 Python 类型注解确保编译时和运行时类型检查
• 可扩展性：支持通过继承和组合扩展新类型
• 序列化兼容：支持 JSON 等常见序列化格式

任务委派机制

任务委派是 A2A 协议的核心功能之一，允许一个 Agent 将任务委托给另一个 Agent 执行。委派过程涉及多个步骤，包括任务描述构建、目标 Agent 选择、执行状态跟踪和结果收集。

sequenceDiagram
    participant Agent_A as Agent A (发起者)
    participant A2A as A2A 协议层
    participant DEL as 委派工具
    participant Agent_B as Agent B (执行者)
    
    Agent_A->>A2A: 创建任务委派请求
    A2A->>DEL: 验证委派权限
    DEL->>DEL: 检查 Agent B 可用性
    DEL->>Agent_B: 发送任务描述
    Agent_B->>Agent_B: 执行任务
    Agent_B-->>DEL: 返回任务结果
    DEL-->>A2A: 格式化响应
    A2A-->>Agent_A: 返回执行结果

委派工具 (delegation.py)

delegation.py 模块提供了 A2A 协议中任务委派的核心实现。该模块包含以下主要功能：

A2A 消息流

graph LR
    subgraph "消息生命周期"
        CREATE["创建消息"] --> SERIALIZE["序列化"]
        SERIALIZE --> SEND["发送"]
        SEND --> RECEIVE["接收"]
        RECEIVE --> DESERIALIZE["反序列化"]
        DESERIALIZE --> PROCESS["处理"]
        PROCESS --> RESPOND["响应"]
        RESPOND --> COMPLETE["完成"]
    end
    
    subgraph "消息类型"
        REQ["请求消息"]
        RESP["响应消息"]
        EVENT["事件消息"]
        ERROR["错误消息"]
    end
    
    CREATE -.-> REQ
    PROCESS -.-> RESP
    PROCESS -.-> EVENT
    PROCESS -.-> ERROR

协议比较与选择

实际应用示例

MCP 使用场景

1. 文件操作：读取本地文件、写入日志
2. 网络请求：调用外部 API、获取网页内容
3. 数据库访问：查询和更新数据存储
4. 搜索功能：文档搜索、网络搜索

A2A 使用场景

1. 复杂任务分解：将复杂任务分解为多个子任务
2. 专家咨询：向特定领域的 Agent 请求专业知识
3. 协作验证：多个 Agent 协作验证结果
4. 负载均衡：根据 Agent 负载分配任务

最佳实践

MCP 最佳实践

• 连接复用：对于频繁调用的服务，保持 MCP 连接复用
• 超时设置：根据任务性质合理设置超时时间
• 错误重试：实现指数退避的重试机制
• 资源清理：确保使用完毕后正确关闭连接

A2A 最佳实践

• 任务粒度：合理划分任务粒度，避免过度委派
• 状态同步：使用适当的状态同步机制保持 Agent 间一致性
• 超时处理：设置合理的任务执行超时时间
• 结果聚合：建立有效的结果聚合和处理机制

安全考虑

总结

MCP 和 A2A 协议共同构成了 CrewAI 多智能体系统的通信基础。MCP 协议侧重于 Agent 与外部世界的连接，提供了多种灵活的传输方式；A2A 协议则专注于 Agent 之间的协作，实现了任务委派、状态共享和协调工作等功能。理解这两个协议的设计和使用方法，对于构建高效、可靠的多智能体系统至关重要。

通过合理地结合使用 MCP 和 A2A 协议，开发者可以构建出功能强大、架构灵活的多智能体应用，实现复杂任务的自动化处理和智能协作。

Pitfall Log / 踩坑日志

项目：crewAIInc/crewAI

摘要：发现 23 个潜在踩坑项，其中 4 个为 high/blocking；最高优先级：安装坑 - 来源证据：[FEATURE] Implement Process.consensual with a pluggable ConsensusEngine。

1. 安装坑 · 来源证据：[FEATURE] Implement Process.consensual with a pluggable ConsensusEngine

• 严重度：high
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安装相关的待验证问题：[FEATURE] Implement Process.consensual with a pluggable ConsensusEngine
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_a7c38215ebb04a4fbc6e7c6d2fdb2469 | https://github.com/crewAIInc/crewAI/issues/5708 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

2. 运行坑 · 来源证据：[BUG] Wrong code in document

• 严重度：high
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个运行相关的待验证问题：[BUG] Wrong code in document
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_390380af45524e959d558b160597b38b | https://github.com/crewAIInc/crewAI/issues/5378 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

3. 运行坑 · 来源证据：[FEATURE] Enhance the document about @persisit

• 严重度：high
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个运行相关的待验证问题：[FEATURE] Enhance the document about @persisit
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_effef000d4cf47a892f17850aa033610 | https://github.com/crewAIInc/crewAI/issues/5372 | 来源类型 github_issue 暴露的待验证使用条件。

4. 安全/权限坑 · 来源证据：[FEATURE] GuardrailProvider interface for pre-tool-call authorization

• 严重度：high
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：[FEATURE] GuardrailProvider interface for pre-tool-call authorization
• 对用户的影响：可能阻塞安装或首次运行。
• 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_858a1a8bead2456289d686ec0d2d802c | https://github.com/crewAIInc/crewAI/issues/4877 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

5. 身份坑 · 仓库名和安装名不一致

• 严重度：medium
• 证据强度：runtime_trace
• 发现：仓库名 crewai 与安装入口 skills 不完全一致。
• 对用户的影响：用户照着仓库名搜索包或照着包名找仓库时容易走错入口。
• 建议检查：在 npm/PyPI/GitHub 上确认包名映射和官方 README 说明。
• 复现命令：npx skills
• 防护动作：页面必须同时展示 repo 名和真实安装入口，避免用户搜索错包。
• 证据：identity.distribution | github_repo:710601088 | https://github.com/crewAIInc/crewAI | repo=crewai; install=skills

6. 安装坑 · 来源证据：1.14.4

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安装相关的待验证问题：1.14.4
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_dcfa2a852812414a82683392c2888795 | https://github.com/crewAIInc/crewAI/releases/tag/1.14.4 | 来源类型 github_release 暴露的待验证使用条件。

7. 安装坑 · 来源证据：1.14.4a1

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安装相关的待验证问题：1.14.4a1
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_0de86b993ffa4d9298726daa189abf49 | https://github.com/crewAIInc/crewAI/releases/tag/1.14.4a1 | 来源类型 github_release 暴露的待验证使用条件。

8. 安装坑 · 来源证据：1.14.5a4

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安装相关的待验证问题：1.14.5a4
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_6db6cd94b6ef4bffa23da215458565b4 | https://github.com/crewAIInc/crewAI/releases/tag/1.14.5a4 | 来源类型 github_release 暴露的待验证使用条件。

9. 配置坑 · 来源证据：Scans the client database to extract existing policy details.

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个配置相关的待验证问题：Scans the client database to extract existing policy details.
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_8201c73155314e67801bd0b81ea9820d | https://github.com/crewAIInc/crewAI/issues/5760 | 来源类型 github_issue 暴露的待验证使用条件。

10. 能力坑 · 能力判断依赖假设

• 严重度：medium
• 证据强度：source_linked
• 发现：README/documentation is current enough for a first validation pass.
• 对用户的影响：假设不成立时，用户拿不到承诺的能力。
• 建议检查：将假设转成下游验证清单。
• 防护动作：假设必须转成验证项；没有验证结果前不能写成事实。
• 证据：capability.assumptions | github_repo:710601088 | https://github.com/crewAIInc/crewAI | README/documentation is current enough for a first validation pass.

11. 维护坑 · 来源证据：1.14.5a1

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个维护/版本相关的待验证问题：1.14.5a1
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_c4094bb55c234f1581b9879efd4994c6 | https://github.com/crewAIInc/crewAI/releases/tag/1.14.5a1 | 来源类型 github_release 暴露的待验证使用条件。

12. 维护坑 · 维护活跃度未知

• 严重度：medium
• 证据强度：source_linked
• 发现：未记录 last_activity_observed。
• 对用户的影响：新项目、停更项目和活跃项目会被混在一起，推荐信任度下降。
• 建议检查：补 GitHub 最近 commit、release、issue/PR 响应信号。
• 防护动作：维护活跃度未知时，推荐强度不能标为高信任。
• 证据：evidence.maintainer_signals | github_repo:710601088 | https://github.com/crewAIInc/crewAI | last_activity_observed missing

13. 安全/权限坑 · 下游验证发现风险项

• 严重度：medium
• 证据强度：source_linked
• 发现：no_demo
• 对用户的影响：下游已经要求复核，不能在页面中弱化。
• 建议检查：进入安全/权限治理复核队列。
• 防护动作：下游风险存在时必须保持 review/recommendation 降级。
• 证据：downstream_validation.risk_items | github_repo:710601088 | https://github.com/crewAIInc/crewAI | no_demo; severity=medium

14. 安全/权限坑 · 存在评分风险

• 严重度：medium
• 证据强度：source_linked
• 发现：no_demo
• 对用户的影响：风险会影响是否适合普通用户安装。
• 建议检查：把风险写入边界卡，并确认是否需要人工复核。
• 防护动作：评分风险必须进入边界卡，不能只作为内部分数。
• 证据：risks.scoring_risks | github_repo:710601088 | https://github.com/crewAIInc/crewAI | no_demo; severity=medium

15. 安全/权限坑 · 来源证据：1.14.3

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：1.14.3
• 对用户的影响：可能影响授权、密钥配置或安全边界。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_ff4d292bd8984a039527502ef51aed98 | https://github.com/crewAIInc/crewAI/releases/tag/1.14.3 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

16. 安全/权限坑 · 来源证据：1.14.5a2

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：1.14.5a2
• 对用户的影响：可能影响授权、密钥配置或安全边界。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_0e5616e4bf1f47f59e10e83c1849c86a | https://github.com/crewAIInc/crewAI/releases/tag/1.14.5a2 | 来源类型 github_release 暴露的待验证使用条件。

17. 安全/权限坑 · 来源证据：1.14.5a3

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：1.14.5a3
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_ba25d089d674407d83a29ec4caff6284 | https://github.com/crewAIInc/crewAI/releases/tag/1.14.5a3 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

18. 安全/权限坑 · 来源证据：Question: integration path for Agent Threat Rules detection in crewai/security

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：Question: integration path for Agent Threat Rules detection in crewai/security
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_05e39d8c109f4ed0a2974d448468229f | https://github.com/crewAIInc/crewAI/issues/5763 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

19. 安全/权限坑 · 来源证据：Security: OWASP Agent Memory Guard – protect CrewAI agents from memory poisoning

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：Security: OWASP Agent Memory Guard – protect CrewAI agents from memory poisoning
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_028fe9ebbcc943f5a7134e08d0ed9450 | https://github.com/crewAIInc/crewAI/issues/5762 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

20. 安全/权限坑 · 来源证据：Security: Request to enable Private Vulnerability Reporting / coordinate channel

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：Security: Request to enable Private Vulnerability Reporting / coordinate channel
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_e6b9a787eb544f098525817b61476c11 | https://github.com/crewAIInc/crewAI/issues/5728 | 来源类型 github_issue 暴露的待验证使用条件。

21. 安全/权限坑 · 来源证据：[FEATURE] Tool to add input_files

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：[FEATURE] Tool to add input_files
• 对用户的影响：可能影响授权、密钥配置或安全边界。
• 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_5ca4885df9ee49459ce8502e94d11f50 | https://github.com/crewAIInc/crewAI/issues/5758 | 来源类型 github_issue 暴露的待验证使用条件。

22. 维护坑 · issue/PR 响应质量未知

• 严重度：low
• 证据强度：source_linked
• 发现：issue_or_pr_quality=unknown。
• 对用户的影响：用户无法判断遇到问题后是否有人维护。
• 建议检查：抽样最近 issue/PR，判断是否长期无人处理。
• 防护动作：issue/PR 响应未知时，必须提示维护风险。
• 证据：evidence.maintainer_signals | github_repo:710601088 | https://github.com/crewAIInc/crewAI | issue_or_pr_quality=unknown

23. 维护坑 · 发布节奏不明确

• 严重度：low
• 证据强度：source_linked
• 发现：release_recency=unknown。
• 对用户的影响：安装命令和文档可能落后于代码，用户踩坑概率升高。
• 建议检查：确认最近 release/tag 和 README 安装命令是否一致。
• 防护动作：发布节奏未知或过期时，安装说明必须标注可能漂移。
• 证据：evidence.maintainer_signals | github_repo:710601088 | https://github.com/crewAIInc/crewAI | release_recency=unknown