CrewAI vs AutoGen：多Agent框架的2026年对比

from crewai import Agent, Task, Crew, Process

# 定义角色
researcher = Agent(
    role="市场研究员",
    goal="收集并分析目标市场的竞争格局",
    backstory="你是拥有10年经验的市场分析师，擅长数据驱动的竞争分析",
    tools=[search_tool, web_scraper],
    llm="gpt-5",
)

writer = Agent(
    role="技术作家",
    goal="将研究结果转化为清晰的市场分析报告",
    backstory="你曾在McKinsey担任商业分析师，擅长结构化写作",
    llm="claude-4.5-sonnet",
)

# 定义任务
research_task = Task(
    description="分析中国电动车市场2026年Q1竞争格局",
    expected_output="包含市场份额、关键玩家、技术趋势的结构化数据",
    agent=researcher,
)

report_task = Task(
    description="基于研究结果撰写市场分析报告",
    expected_output="2000字的商业分析报告，包含图表建议和战略建议",
    agent=writer,
    context=[research_task],  # 依赖前置任务
)

# 组建团队并执行
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, report_task],
    process=Process.sequential,
)

result = crew.kickoff()

from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager

# 创建Agent
market_analyst = AssistantAgent(
    name="市场分析师",
    system_message="你是市场分析专家。分析市场数据并提供洞察。"
                   "回复格式：TERMINATE 表示任务完成。",
    llm_config={"model": "gpt-5"},
)

data_scientist = AssistantAgent(
    name="数据科学家",
    system_message="你是数据分析专家。负责数据处理、统计分析和可视化建议。"
                   "回复格式：TERMINATE 表示任务完成。",
    llm_config={"model": "gpt-5"},
)

reviewer = AssistantAgent(
    name="审阅者",
    system_message="你是报告审阅者。检查报告质量，确认无误后回复 TERMINATE。",
    llm_config={"model": "claude-4.5-sonnet"},
)

user_proxy = UserProxyAgent(
    name="用户",
    human_input_mode="NEVER",
    code_execution_config={"work_dir": "workspace"},
)

# 创建群聊
group_chat = GroupChat(
    agents=[user_proxy, market_analyst, data_scientist, reviewer],
    messages=[],
    max_round=15,
)

manager = GroupChatManager(groupchat=group_chat)

# 启动对话
user_proxy.initiate_chat(
    manager,
    message="分析中国电动车市场2026年Q1竞争格局并生成报告",
)

# 顺序流程
crew = Crew(agents=agents, tasks=tasks, process=Process.sequential)

# 层级流程（有管理者Agent分配任务）
crew = Crew(
    agents=[manager, researcher, writer, reviewer],
    tasks=tasks,
    process=Process.hierarchical,
    manager_agent=manager,
)

# 共识流程（Agent投票决策）
crew = Crew(
    agents=[analyst_a, analyst_b, analyst_c],
    tasks=[analysis_task],
    process=Process.consensus,
)

# 自定义发言权选择函数
def custom_speaker_selection(last_speaker, groupchat):
    messages = groupchat.messages
    if last_speaker.name == "市场分析师":
        return "数据科学家"  # 分析师发言后交给数据科学家
    elif last_speaker.name == "数据科学家":
        return "审阅者"      # 数据科学家后交给审阅者
    return None  # 结束对话

group_chat = GroupChat(
    agents=agents,
    speaker_selection_method=custom_speaker_selection,
    max_round=10,
)

# CrewAI：内置详细的执行日志
crew = Crew(agents=agents, tasks=tasks, verbose=True)
result = crew.kickoff()

# 输出示例：
# [2026-07-29 10:00:01] Task 1 assigned to 市场研究员
# [2026-07-29 10:00:15] 市场研究员 completed Task 1
# [2026-07-29 10:00:16] Task 2 assigned to 技术作家
# ...

# AutoGen：通过消息历史调试
for msg in group_chat.messages:
    print(f"[{msg['role']}] {msg['name']}: {msg['content'][:100]}...")

# 可接入OpenTelemetry进行分布式追踪

from crewai import Crew
from crewai.telemetry import Telemetry

# 内存管理
crew = Crew(
    agents=agents,
    tasks=tasks,
    memory=True,              # 启用记忆
    memory_config={
        "provider": "chroma",
        "short_term": True,
        "long_term": True,
    },
    cache=True,               # 缓存工具调用结果
    max_rpm=50,               # 速率限制
)

# 可观测性
telemetry = Telemetry()
telemetry.track_crew(crew)

# 流式输出
for chunk in crew.kickoff_stream():
    print(chunk, end="", flush=True)

from autogen import AgentHub, RuntimeContext

# 分布式执行
runtime = RuntimeContext(
    mode="distributed",
    registry_url="redis://localhost:6379",
    max_concurrent_agents=10,
)

# Agent注册中心
hub = AgentHub(runtime=runtime)
hub.register(market_analyst)
hub.register(data_scientist)

# 可观测性
from autogen.observability import setup_telemetry
setup_telemetry(
    exporter="otlp",
    endpoint="http://localhost:4317",
    service_name="multi-agent-system",
)