CrewAI与AutoGen

概述

CrewAI和AutoGen是两个专注于多智能体协作的框架。CrewAI采用角色扮演范式，而AutoGen采用可对话智能体范式。两者各有优势，适用于不同场景。

CrewAI

核心理念

CrewAI的设计灵感来自现实世界的团队协作——每个智能体扮演特定角色，共同完成任务。

graph TD
    subgraph CrewAI
        A[Crew 团队] --> B[Agent 1<br/>Researcher 研究员]
        A --> C[Agent 2<br/>Writer 撰稿人]
        A --> D[Agent 3<br/>Reviewer 审核员]

        B --> E[Task 1<br/>研究主题]
        C --> F[Task 2<br/>撰写文章]
        D --> G[Task 3<br/>审核质量]

        E --> F
        F --> G
    end

    B --> B1[Tools: 搜索/RAG]
    C --> C1[Tools: 写作辅助]
    D --> D1[Tools: 质量检查]

Agent定义

from crewai import Agent, Task, Crew, Process

# 定义智能体角色
researcher = Agent(
    role="Senior Research Analyst",
    goal="Discover and analyze the latest trends in AI",
    backstory="""You are a senior research analyst at a leading 
    tech think tank. You have a knack for identifying emerging 
    trends and extracting actionable insights from data.""",
    tools=[search_tool, scrape_tool],
    llm="gpt-4o",
    verbose=True,
    allow_delegation=True,  # 允许委派任务给其他Agent
)

writer = Agent(
    role="Technical Content Writer",
    goal="Create compelling and accurate technical content",
    backstory="""You are a skilled technical writer known for 
    making complex topics accessible. You excel at transforming 
    research into engaging articles.""",
    tools=[],
    llm="gpt-4o",
    verbose=True,
)

reviewer = Agent(
    role="Quality Assurance Editor",
    goal="Ensure content accuracy and quality",
    backstory="""You are a meticulous editor with expertise in 
    technical content. You ensure factual accuracy, readability, 
    and adherence to style guidelines.""",
    tools=[fact_check_tool],
    llm="gpt-4o",
    verbose=True,
)

Task定义

# 定义任务
research_task = Task(
    description="""Research the latest developments in AI agents.
    Focus on:
    1. New frameworks and tools
    2. Research breakthroughs
    3. Industry adoption trends

    Provide a comprehensive research report with sources.""",
    expected_output="A detailed research report with key findings and sources",
    agent=researcher,
)

writing_task = Task(
    description="""Based on the research report, write a 1500-word 
    article about AI agents. The article should be:
    - Engaging and accessible
    - Technically accurate
    - Well-structured with clear sections""",
    expected_output="A polished 1500-word article in markdown format",
    agent=writer,
    context=[research_task],  # 依赖research_task的输出
)

review_task = Task(
    description="""Review the article for:
    1. Factual accuracy
    2. Readability and flow
    3. Technical depth
    4. Grammar and style

    Provide specific feedback and a final revised version.""",
    expected_output="Reviewed and revised article with feedback notes",
    agent=reviewer,
    context=[writing_task],
)

Process类型

# 顺序执行
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, writing_task, review_task],
    process=Process.sequential,  # 按顺序执行
    verbose=True,
)

# 层级执行（需要Manager Agent）
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, writing_task, review_task],
    process=Process.hierarchical,  # Manager分配任务
    manager_llm="gpt-4o",
    verbose=True,
)

# 运行
result = crew.kickoff()
print(result)

CrewAI工具

from crewai_tools import (
    SerperDevTool,      # 搜索引擎
    ScrapeWebsiteTool,  # 网页抓取
    FileReadTool,       # 文件读取
    PDFSearchTool,      # PDF搜索
    CodeInterpreterTool, # 代码执行
)

# 自定义工具
from crewai.tools import tool

@tool("Database Query")
def query_database(sql: str) -> str:
    """Execute a SQL query against the analytics database.
    Use this to get data for analysis."""
    return db.execute(sql)

高级特性

记忆系统

crew = Crew(
    agents=[...],
    tasks=[...],
    memory=True,           # 启用记忆
    embedder={
        "provider": "openai",
        "config": {"model": "text-embedding-3-small"}
    }
)
# 支持短期记忆（当前任务）、长期记忆（跨任务）、实体记忆（实体知识）

委派机制

researcher = Agent(
    role="Lead Researcher",
    allow_delegation=True,  # 可以将子任务委派给其他Agent
    # 例如：researcher可以要求writer帮忙整理格式
)

AutoGen

核心理念

AutoGen（微软）基于可对话智能体（Conversable Agent）范式，智能体通过对话来协作。

graph TD
    subgraph AutoGen
        A[UserProxy<br/>用户代理] <-->|对话| B[Assistant<br/>AI助手]
        B <-->|对话| C[Coder<br/>代码员]
        A <-->|对话| C

        D[GroupChat Manager<br/>群聊管理器]
        D --> A
        D --> B
        D --> C
    end

ConversableAgent

import autogen

# 配置LLM
config_list = [
    {"model": "gpt-4o", "api_key": "..."}
]

llm_config = {"config_list": config_list, "temperature": 0}

# AI助手
assistant = autogen.AssistantAgent(
    name="assistant",
    system_message="""You are a helpful AI assistant. 
    Solve tasks using your coding and language skills.
    When you need to execute code, write it in python code blocks.""",
    llm_config=llm_config,
)

# 用户代理（可执行代码）
user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",  # NEVER / TERMINATE / ALWAYS
    max_consecutive_auto_reply=10,
    is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
    code_execution_config={
        "work_dir": "coding",
        "use_docker": False,
    },
)

# 启动对话
user_proxy.initiate_chat(
    assistant,
    message="Write a Python script to analyze the top 10 most popular programming languages."
)

GroupChat

# 多智能体群聊
coder = autogen.AssistantAgent(
    name="Coder",
    system_message="You are a Python expert. Write clean, efficient code.",
    llm_config=llm_config,
)

reviewer = autogen.AssistantAgent(
    name="Reviewer",
    system_message="""You review code for bugs, efficiency, and best practices. 
    Be specific in your feedback.""",
    llm_config=llm_config,
)

tester = autogen.AssistantAgent(
    name="Tester",
    system_message="You write comprehensive test cases for the given code.",
    llm_config=llm_config,
)

# 创建群聊
groupchat = autogen.GroupChat(
    agents=[user_proxy, coder, reviewer, tester],
    messages=[],
    max_round=12,
    speaker_selection_method="auto",  # auto / round_robin / random
)

# 群聊管理器
manager = autogen.GroupChatManager(
    groupchat=groupchat,
    llm_config=llm_config,
)

# 启动群聊
user_proxy.initiate_chat(
    manager,
    message="Create a REST API for a todo list application with proper error handling."
)

代码执行

AutoGen的一个核心能力是安全地执行代码：

user_proxy = autogen.UserProxyAgent(
    name="executor",
    code_execution_config={
        "work_dir": "workspace",
        "use_docker": True,          # 使用Docker隔离
        "timeout": 60,               # 超时秒数
        "last_n_messages": 3,        # 从最近N条消息中提取代码
    },
)

# 工作流程：
# 1. Assistant生成代码
# 2. UserProxy在Docker中执行
# 3. 执行结果反馈给Assistant
# 4. Assistant根据结果修改或继续

自定义Speaker选择

def custom_speaker_selection(last_speaker, groupchat):
    """自定义发言顺序"""
    messages = groupchat.messages

    if last_speaker.name == "Coder":
        return reviewer  # Coder之后是Reviewer
    elif last_speaker.name == "Reviewer":
        if "LGTM" in messages[-1]["content"]:
            return tester  # 审核通过后进入测试
        else:
            return coder   # 审核未通过，回到Coder
    elif last_speaker.name == "Tester":
        if "all tests passed" in messages[-1]["content"].lower():
            return user_proxy  # 测试通过，结束
        else:
            return coder  # 测试失败，回到Coder

    return coder  # 默认

groupchat = autogen.GroupChat(
    agents=[user_proxy, coder, reviewer, tester],
    messages=[],
    speaker_selection_method=custom_speaker_selection,
)

AutoGen的高级模式

嵌套对话

# Agent A 在处理过程中启动与Agent B的子对话
assistant.register_nested_chats(
    [
        {
            "recipient": expert_agent,
            "message": "I need expert advice on: {question}",
            "max_turns": 3,
        }
    ],
    trigger=lambda sender: needs_expert_help(sender.last_message()),
)

可教学Agent

from autogen.agentchat.contrib.teachable_agent import TeachableAgent

teachable = TeachableAgent(
    name="teachable_agent",
    llm_config=llm_config,
    teach_config={
        "reset_db": False,       # 保留之前学到的知识
        "path_to_db_dir": "./teach_db",
    },
)
# 用户可以教Agent新知识，Agent会记住并在未来使用

CrewAI vs AutoGen 对比

维度	CrewAI	AutoGen
范式	角色扮演+任务驱动	对话驱动
定义方式	Agent + Task + Crew	Agent + Chat
流程控制	顺序/层级	自由对话/自定义
代码执行	通过工具	原生Docker隔离
记忆	内置多层记忆	通过扩展
学习曲线	平缓	中等
灵活性	中等	高
适合场景	明确角色分工的工作流	探索性/研究性任务
社区	快速增长	稳定
维护方	CrewAI Inc.	Microsoft Research

何时选择CrewAI

• 角色和任务明确定义
• 需要可预测的工作流程
• 团队协作类比直观
• 快速原型开发

# CrewAI适合：角色明确的生产流水线
# 例如：内容创作团队
crew = Crew(
    agents=[researcher, writer, editor, seo_specialist],
    tasks=[research, draft, edit, optimize],
    process=Process.sequential
)

何时选择AutoGen

• 需要灵活的对话交互
• 涉及代码生成和执行
• 探索性/研究性任务
• 需要人在回路中

# AutoGen适合：探索性的编程任务
# 例如：数据分析对话
user_proxy.initiate_chat(
    analyst,
    message="Explore this dataset and find interesting patterns"
)
# analyst可以自由地写代码、执行、观察结果、再调整

实际案例对比

案例：市场分析报告

CrewAI实现:

# 明确的角色和任务链
analyst = Agent(role="Market Analyst", ...)
researcher = Agent(role="Data Researcher", ...)
writer = Agent(role="Report Writer", ...)

crew = Crew(
    agents=[analyst, researcher, writer],
    tasks=[
        Task(description="Identify key market trends", agent=analyst),
        Task(description="Gather supporting data", agent=researcher),
        Task(description="Write the final report", agent=writer),
    ],
    process=Process.sequential
)

AutoGen实现:

# 自由的对话协作
analyst = AssistantAgent(name="analyst", ...)
data_agent = AssistantAgent(name="data_agent", ...)
writer = AssistantAgent(name="writer", ...)

groupchat = GroupChat(
    agents=[user_proxy, analyst, data_agent, writer],
    max_round=20,
    speaker_selection_method="auto"
)
# 智能体自由讨论，动态决定下一步

总结

需要	选择	原因
结构化的团队工作流	CrewAI	角色和任务定义清晰
灵活的多智能体对话	AutoGen	对话范式更灵活
快速原型	CrewAI	API更简洁
代码生成+执行	AutoGen	Docker隔离的代码执行
可教学智能体	AutoGen	内置Teachable Agent
生产部署	CrewAI	更可预测的行为

两个框架都在快速发展中，建议根据具体需求选择，并关注各自社区的最新动态。

---

评论