CrewAI多代理编排技能Skill crewai-multi-agent

---

name: crewai-multi-agent description: 用于自主AI协作的多代理编排框架。在构建专门代理团队协作解决复杂任务、需要基于角色的代理协作与记忆、或需要顺序/分层执行的生产工作流程时使用。构建无LangChain依赖，以实现精简、快速的执行。 version: 1.0.0 author: Orchestra Research license: MIT tags: [代理, CrewAI, 多代理, 编排, 协作, 基于角色, 自主, 工作流程, 记忆, 生产] dependencies: [crewai>=1.2.0, crewai-tools>=1.2.0]

CrewAI - 多代理编排框架

构建协作解决复杂任务的自主AI代理团队。

何时使用CrewAI

使用CrewAI当：

• 构建具有专门角色的多代理系统
• 需要代理之间的自主协作
• 想要基于角色的任务委托（研究者、作者、分析师）
• 需要顺序或分层过程执行
• 构建具有记忆和可观察性的生产工作流程
• 需要比LangChain/LangGraph更简单的设置

关键特性：

• 独立：无LangChain依赖，精简足迹
• 基于角色：代理有角色、目标和背景故事
• 双范式：团队（自主）+ 流程（事件驱动）
• 50+工具：网络爬虫、搜索、数据库、AI服务
• 记忆：短期、长期和实体记忆
• 生产就绪：跟踪、企业特性

使用替代方案：

• LangChain：通用LLM应用，RAG管道
• LangGraph：复杂有状态工作流程带循环
• AutoGen：微软生态系统，多代理对话
• LlamaIndex：文档问答、知识检索

快速开始

安装

# 核心框架
pip install crewai

# 带50+内置工具
pip install 'crewai[tools]'

使用CLI创建项目

# 创建新团队项目
crewai create crew my_project
cd my_project

# 安装依赖
crewai install

# 运行团队
crewai run

简单团队（仅代码）

from crewai import Agent, Task, Crew, Process

# 1. 定义代理
researcher = Agent(
    role="高级研究分析师",
    goal="发现AI领域的尖端发展",
    backstory="您是新兴趋势识别专家，具有敏锐的分析眼光。",
    verbose=True
)

writer = Agent(
    role="技术作者",
    goal="创建关于技术主题的清晰、吸引人的内容",
    backstory="您擅长向普通受众解释复杂概念。",
    verbose=True
)

# 2. 定义任务
research_task = Task(
    description="研究{topic}的最新发展。找到5个关键趋势。",
    expected_output="带5个关键趋势要点的详细报告。",
    agent=researcher
)

write_task = Task(
    description="基于研究结果撰写博客文章。",
    expected_output="500字Markdown格式的博客文章。",
    agent=writer,
    context=[research_task]  # 使用研究输出
)

# 3. 创建和运行团队
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,  # 任务顺序执行
    verbose=True
)

# 4. 执行
result = crew.kickoff(inputs={"topic": "AI代理"})
print(result.raw)

核心概念

代理 - 自主工作者

from crewai import Agent

agent = Agent(
    role="数据科学家",                    # 职位/角色
    goal="分析数据以发现见解",           # 目标
    backstory="统计学博士...",           # 背景上下文
    llm="gpt-4o",                         # 使用的LLM
    tools=[],                             # 可用工具
    memory=True,                          # 启用记忆
    verbose=True,                         # 显示推理
    allow_delegation=True,                # 可委托给其他人
    max_iter=15,                          # 最大推理迭代
    max_rpm=10                            # 速率限制
)

任务 - 工作单元

from crewai import Task

task = Task(
    description="分析2024年第四季度的销售数据。{context}",
    expected_output="带关键指标和趋势的摘要报告。",
    agent=analyst,                        # 分配代理
    context=[previous_task],              # 来自其他任务的输入
    output_file="report.md",              # 保存到文件
    async_execution=False,                # 同步运行
    human_input=False                     # 无需人工批准
)

团队 - 代理团队

from crewai import Crew, Process

crew = Crew(
    agents=[researcher, writer, editor],  # 团队成员
    tasks=[research, write, edit],        # 完成的任务
    process=Process.sequential,           # 或Process.hierarchical
    verbose=True,
    memory=True,                          # 启用团队记忆
    cache=True,                           # 缓存工具结果
    max_rpm=10,                           # 速率限制
    share_crew=False                      # 选择加入遥测
)

# 带输入执行
result = crew.kickoff(inputs={"topic": "AI趋势"})

# 访问结果
print(result.raw)                         # 最终输出
print(result.tasks_output)                # 所有任务输出
print(result.token_usage)                 # 令牌消耗

流程类型

顺序（默认）

任务按顺序执行，每个代理完成其任务后才进行下一个：

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential  # 任务1 → 任务2 → 任务3
)

分层

自动创建管理代理，负责委托和协调：

crew = Crew(
    agents=[researcher, writer, analyst],
    tasks=[research_task, write_task, analyze_task],
    process=Process.hierarchical,  # 管理代理委托任务
    manager_llm="gpt-4o"           # 管理代理的LLM
)

使用工具

内置工具（50+）

pip install 'crewai[tools]'

from crewai_tools import (
    SerperDevTool,           # 网络搜索
    ScrapeWebsiteTool,       # 网络爬虫
    FileReadTool,            # 读取文件
    PDFSearchTool,           # 搜索PDF
    WebsiteSearchTool,       # 搜索网站
    CodeDocsSearchTool,      # 搜索代码文档
    YoutubeVideoSearchTool,  # 搜索YouTube
)

# 分配工具给代理
researcher = Agent(
    role="研究者",
    goal="查找准确信息",
    backstory="在线查找数据专家。",
    tools=[SerperDevTool(), ScrapeWebsiteTool()]
)

自定义工具

from crewai.tools import BaseTool
from pydantic import Field

class CalculatorTool(BaseTool):
    name: str = "计算器"
    description: str = "执行数学计算。输入：表达式"

    def _run(self, expression: str) -> str:
        try:
            result = eval(expression)
            return f"结果: {result}"
        except Exception as e:
            return f"错误: {str(e)}"

# 使用自定义工具
agent = Agent(
    role="分析师",
    goal="执行计算",
    tools=[CalculatorTool()]
)

YAML配置（推荐）

项目结构

my_project/
├── src/my_project/
│   ├── config/
│   │   ├── agents.yaml    # 代理定义
│   │   └── tasks.yaml     # 任务定义
│   ├── crew.py            # 团队组装
│   └── main.py            # 入口点
└── pyproject.toml

agents.yaml

researcher:
  role: "{topic}高级数据研究员"
  goal: "揭示{topic}领域的尖端发展"
  backstory: >
    您是经验丰富的研究员，善于揭示{topic}的最新发展。
    以查找相关信息并清晰呈现而闻名。

reporting_analyst:
  role: "报告分析师"
  goal: "基于研究数据创建详细报告"
  backstory: >
    您是细心的分析师，将原始数据转化为
    通过结构良好的报告提供可操作的见解。

tasks.yaml

research_task:
  description: >
    进行关于{topic}的彻底研究。
    为{year}年找到最相关信息。
  expected_output: >
    带10个{topic}最相关
    信息要点的列表。
  agent: researcher

reporting_task:
  description: >
    审查研究并创建综合报告。
    专注于关键发现和建议。
  expected_output: >
    带执行摘要、发现
    和建议的Markdown格式详细报告。
  agent: reporting_analyst
  output_file: report.md

crew.py

from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
from crewai_tools import SerperDevTool

@CrewBase
class MyProjectCrew:
    """我的项目团队"""

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

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

    @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
        )

main.py

from my_project.crew import MyProjectCrew

def run():
    inputs = {
        'topic': 'AI代理',
        'year': 2025
    }
    MyProjectCrew().crew().kickoff(inputs=inputs)

if __name__ == "__main__":
    run()

流程 - 事件驱动编排

对于带条件逻辑的复杂工作流程，使用流程：

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

class MyState(BaseModel):
    confidence: float = 0.0

class MyFlow(Flow[MyState]):
    @start()
    def gather_data(self):
        return {"data": "收集的"}

    @listen(gather_data)
    def analyze(self, data):
        self.state.confidence = 0.85
        return analysis_crew.kickoff(inputs=data)

    @router(analyze)
    def decide(self):
        return "高" if self.state.confidence > 0.8 else "低"

    @listen("高")
    def generate_report(self):
        return report_crew.kickoff()

# 运行流程
flow = MyFlow()
result = flow.kickoff()

参见流程指南获取完整文档。

记忆系统

# 启用所有记忆类型
crew = Crew(
    agents=[researcher],
    tasks=[research_task],
    memory=True,           # 启用记忆
    embedder={             # 自定义嵌入
        "provider": "openai",
        "config": {"model": "text-embedding-3-small"}
    }
)

记忆类型： 短期（ChromaDB）、长期（SQLite）、实体（ChromaDB）

LLM提供商

from crewai import LLM

llm = LLM(model="gpt-4o")                              # OpenAI（默认）
llm = LLM(model="claude-sonnet-4-5-20250929")                       # Anthropic
llm = LLM(model="ollama/llama3.1", base_url="http://localhost:11434")  # 本地
llm = LLM(model="azure/gpt-4o", base_url="https://...")              # Azure

agent = Agent(role="分析师", goal="分析数据", llm=llm)

CrewAI vs 替代方案

特性	CrewAI	LangChain	LangGraph
最适合	多代理团队	通用LLM应用	有状态工作流程
学习曲线	低	中	高
代理范式	基于角色	基于工具	基于图
记忆	内置	插件式	自定义

最佳实践

1. 清晰角色 - 每个代理应有明确专长
2. YAML配置 - 大型项目更好的组织
3. 启用记忆 - 改进跨任务上下文
4. 设置max_iter - 防止无限循环（默认15）
5. 限制工具 - 每个代理最多3-5个工具
6. 速率限制 - 设置max_rpm避免API限制

常见问题

代理陷入循环：

agent = Agent(
    role="...",
    max_iter=10,           # 限制迭代
    max_rpm=5              # 速率限制
)

任务未使用上下文：

task2 = Task(
    description="...",
    context=[task1],       # 显式传递上下文
    agent=writer
)

记忆错误：

# 使用环境变量存储
import os
os.environ["CREWAI_STORAGE_DIR"] = "./my_storage"

参考文献

• 流程指南 - 事件驱动工作流程，状态管理
• 工具指南 - 内置工具，自定义工具，MCP
• 故障排除 - 常见问题，调试

资源

• GitHub: https://github.com/crewAIInc/crewAI (25k+ stars)
• 文档: https://docs.crewai.com
• 工具: https://github.com/crewAIInc/crewAI-tools
• 示例: https://github.com/crewAIInc/crewAI-examples
• 版本: 1.2.0+
• 许可证: MIT