crewai-multi-agent

CrewAI - 多 Agent 编排框架

构建能够协作解决复杂任务的自主 AI Agent 团队。

何时使用 CrewAI

适用于以下情况：

• 构建由具备专门职责的 Agent 组成的多 Agent 系统
• 需要 Agent 之间自主协作
• 希望按角色分配任务（研究员、写作者、分析师）
• 需要按顺序或分层执行流程
• 构建具备记忆和可观测性的生产工作流
• 需要比 LangChain/LangGraph 更简单的设置

主要功能：

• 独立运行：不依赖 LangChain，轻量精简
• 基于角色：Agent 拥有角色、目标和背景故事
• 双重范式：Crew（自主运行）+ Flow（事件驱动）
• 50+ 种工具：网页抓取、搜索、数据库、AI 服务
• 记忆：短期、长期和实体记忆
• 可用于生产环境：追踪和企业级功能

以下情况请改用其他方案：

• LangChain：通用大模型应用、RAG 流水线
• LangGraph：包含循环的复杂有状态工作流
• AutoGen：Microsoft 生态系统、多 Agent 对话
• LlamaIndex：文档问答、知识检索

快速入门

安装

# Core framework
pip install crewai

# With 50+ built-in tools
pip install 'crewai[tools]'

使用 CLI 创建项目

# Create new crew project
crewai create crew my_project
cd my_project

# Install dependencies
crewai install

# Run the crew
crewai run

简单 Crew（仅代码）

from crewai import Agent, Task, Crew, Process

# 1. Define agents
researcher = Agent(
    role="Senior Research Analyst",
    goal="Discover cutting-edge developments in AI",
    backstory="You are an expert analyst with a keen eye for emerging trends.",
    verbose=True
)

writer = Agent(
    role="Technical Writer",
    goal="Create clear, engaging content about technical topics",
    backstory="You excel at explaining complex concepts to general audiences.",
    verbose=True
)

# 2. Define tasks
research_task = Task(
    description="Research the latest developments in {topic}. Find 5 key trends.",
    expected_output="A detailed report with 5 bullet points on key trends.",
    agent=researcher
)

write_task = Task(
    description="Write a blog post based on the research findings.",
    expected_output="A 500-word blog post in markdown format.",
    agent=writer,
    context=[research_task]  # Uses research output
)

# 3. Create and run crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,  # Tasks run in order
    verbose=True
)

# 4. Execute
result = crew.kickoff(inputs={"topic": "AI Agents"})
print(result.raw)

核心概念

Agent - 自主工作者

from crewai import Agent

agent = Agent(
    role="Data Scientist",                    # Job title/role
    goal="Analyze data to find insights",     # What they aim to achieve
    backstory="PhD in statistics...",         # Background context
    llm="gpt-4o",                             # LLM to use
    tools=[],                                 # Tools available
    memory=True,                              # Enable memory
    verbose=True,                             # Show reasoning
    allow_delegation=True,                    # Can delegate to others
    max_iter=15,                              # Max reasoning iterations
    max_rpm=10                                # Rate limit
)

Task - 工作单元

from crewai import Task

task = Task(
    description="Analyze the sales data for Q4 2024. {context}",
    expected_output="A summary report with key metrics and trends.",
    agent=analyst,                            # Assigned agent
    context=[previous_task],                  # Input from other tasks
    output_file="report.md",                  # Save to file
    async_execution=False,                    # Run synchronously
    human_input=False                         # No human approval needed
)

Crew - Agent 团队

from crewai import Crew, Process

crew = Crew(
    agents=[researcher, writer, editor],      # Team members
    tasks=[research, write, edit],            # Tasks to complete
    process=Process.sequential,               # Or Process.hierarchical
    verbose=True,
    memory=True,                              # Enable crew memory
    cache=True,                               # Cache tool results
    max_rpm=10,                               # Rate limit
    share_crew=False                          # Opt-in telemetry
)

# Execute with inputs
result = crew.kickoff(inputs={"topic": "AI trends"})

# Access results
print(result.raw)                             # Final output
print(result.tasks_output)                    # All task outputs
print(result.token_usage)                     # Token consumption

流程类型

顺序（默认）

按顺序执行任务，每个 Agent 完成自己的任务后，再由下一个 Agent 执行：

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

分层

自动创建负责分配和协调任务的管理 Agent：

crew = Crew(
    agents=[researcher, writer, analyst],
    tasks=[research_task, write_task, analyze_task],
    process=Process.hierarchical,  # Manager delegates tasks
    manager_llm="gpt-4o"           # LLM for manager
)

使用工具

内置工具（50+ 种）

pip install 'crewai[tools]'

from crewai_tools import (
    SerperDevTool,           # Web search
    ScrapeWebsiteTool,       # Web scraping
    FileReadTool,            # Read files
    PDFSearchTool,           # Search PDFs
    WebsiteSearchTool,       # Search websites
    CodeDocsSearchTool,      # Search code docs
    YoutubeVideoSearchTool,  # Search YouTube
)

# Assign tools to agent
researcher = Agent(
    role="Researcher",
    goal="Find accurate information",
    backstory="Expert at finding data online.",
    tools=[SerperDevTool(), ScrapeWebsiteTool()]
)

自定义工具

from crewai.tools import BaseTool
from pydantic import Field
from scripts.safe_arithmetic import calculate

class CalculatorTool(BaseTool):
    name: str = "Calculator"
    description: str = "Performs mathematical calculations. Input: expression"

    def _run(self, expression: str) -> str:
        try:
            result = calculate(expression)
            return f"Result: {result}"
        except Exception as e:
            return f"Error: {str(e)}"

# Use custom tool
agent = Agent(
    role="Analyst",
    goal="Perform calculations",
    tools=[CalculatorTool()]
)

YAML 配置（推荐）

详情、代码示例和参数表：references/yaml-configuration-recommended.md。适用时请阅读。

Flows - 事件驱动编排

对于包含条件逻辑的复杂工作流，请使用 Flows：

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": "collected"}

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

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

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

# Run flow
flow = MyFlow()
result = flow.kickoff()

完整文档请参阅 Flows 指南。

记忆系统

# Enable all memory types
crew = Crew(
    agents=[researcher],
    tasks=[research_task],
    memory=True,           # Enable memory
    embedder={             # Custom embeddings
        "provider": "openai",
        "config": {"model": "text-embedding-3-small"}
    }
)

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

大模型提供商

from crewai import LLM

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

agent = Agent(role="Analyst", goal="Analyze data", llm=llm)

CrewAI 与其他方案对比

功能	CrewAI	LangChain	LangGraph
最适用场景	多 Agent 团队	通用大模型应用	有状态工作流
学习曲线	低	中	较高
Agent 范式	基于角色	基于工具	基于图
记忆	内置	基于插件	自定义

最佳实践

1. 明确角色 - 每个 Agent 都应有不同的专长
2. YAML 配置 - 更便于组织较大的项目
3. 启用记忆 - 改善任务间的上下文延续
4. 设置 max_iter - 防止无限循环（默认值为 15）
5. 限制工具数量 - 每个 Agent 最多使用 3-5 种工具
6. 限制请求速率 - 设置 max_rpm 以避免触及 API 限制

常见问题

Agent 陷入循环：

agent = Agent(
    role="...",
    max_iter=10,           # Limit iterations
    max_rpm=5              # Rate limit
)

任务未使用上下文：

task2 = Task(
    description="...",
    context=[task1],       # Explicitly pass context
    agent=writer
)

记忆错误：

# Use environment variable for storage
import os
os.environ["CREWAI_STORAGE_DIR"] = "./my_storage"

参考资料

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

资源

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

Agent 操作流程

1. 检查环境。 确认框架版本、模型提供方、API 密钥和速率限制。
2. 明确输入。 根据数据或用户提供的信息确认格式、标识符和参数。任何会改变结果的值都应询问，而不是猜测。
3. 先运行小规模版本。 使用已知输入测试单次调用或链，并检查原始输出。
4. 使用上述说明和参考资料执行完整任务。
5. 验证结果。 使用小型标注数据集进行评估；根据架构检查结构化输出；记录提示词和响应。
6. 报告。 说明运行了什么（版本、命令、参数）、检查了什么，以及仍有哪些不确定之处。

如果发生这种情况	请这样做
输出与预期架构不匹配	添加验证和重试、收紧架构，或简化提示词。
这些说明中的函数、标志或端点在已安装版本中不存在	查阅已安装版本自己的文档（help()、--help、官方文档），进行调整并告知用户。切勿虚构 API。
必需的输入、标识符或参数不明确	询问用户，或在运行前明确说明假设。

诚信准则

• 切勿编造结果、参数、标识符、引文或统计数据。如果无法运行或验证某项内容，请如实说明。
• 未经用户同意，切勿将私人或敏感数据发送至外部 API。
• 请将此处特定于版本的细节视为可能过时，并根据已安装版本的官方文档进行确认。
• 对于会产生费用、消耗共享 GPU 或云配额、涉及个人或患者数据，或无法撤销的操作，须事先征求同意。

相关技能

• autogpt-agents：用于构建和部署持续运行 Agent 的自主 AI Agent 平台。
• langchain：用于构建由大模型驱动、包含 Agent、链和 RAG 的应用程序的框架。
• aeon：此技能适用于时间序列机器学习任务，包括分类、回归、聚类、预测、异常检测等。