CrewAI 深度评测：多 Agent 协作框架的优与劣

from crewai import Agent, Task, Crew, Process

# 定义角色
researcher = Agent(
    role="市场研究员",
    goal="收集目标市场的详细数据，包括规模、竞品、趋势",
    backstory="你是一位有10年经验的市场研究员，擅长数据分析和趋势预测",
    tools=[search_tool, web_scraper],
    llm="gpt-4o"
)

writer = Agent(
    role="技术撰稿人",
    goal="将研究结果转化为清晰、有深度的分析报告",
    backstory="你曾是科技媒体主编，擅长把复杂技术概念转化为易懂文字",
    llm="gpt-4o"
)

# 定义任务
research_task = Task(
    description="研究AI Agent框架市场，包括LangChain、CrewAI、AutoGen的市场份额、用户增长、社区活跃度",
    expected_output="包含数据图表的市场分析报告，至少2000字",
    agent=researcher
)

write_task = Task(
    description="基于研究结果撰写深度分析文章",
    expected_output="结构完整的分析文章，含执行摘要、市场概况、竞品分析、趋势预测",
    expected_output_length="3000字以上",
    agent=writer,
    context=[research_task]  # 依赖研究任务的输出
)

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

result = crew.kickoff()

analyst = Agent(
    role="财务分析师",
    goal="分析财务数据，识别风险和机会",
    backstory="CPA持证，曾在四大审计，擅长从数字中发现问题",
    tools=[python_repl, excel_reader],  # 工具绑定到角色
    allow_delegation=False,             # 不允许委派任务
    max_iter=15,                        # 最大迭代次数
    memory=True,                        # 启用记忆
    verbose=True
)

Research → Analyze → Write → Review → Publish

crew = Crew(
    agents=[researcher, analyst, writer, reviewer],
    tasks=[research_task, analysis_task, write_task, review_task],
    process=Process.hierarchical,
    manager_llm="gpt-4o",  # 管理者使用的LLM
    manager_agent=None      # 自动创建管理者
)

from langchain.tools import Tool
from crewai.tools import BaseTool

# 方式1：使用 LangChain 工具
search_tool = Tool(
    name="搜索",
    func=tavily_search,
    description="搜索互联网获取最新信息"
)

# 方式2：自定义 CrewAI 工具
class DataAnalysisTool(BaseTool):
    name: str = "数据分析"
    description: str = "分析CSV数据并生成统计摘要"
    
    def _run(self, file_path: str) -> str:
        import pandas as pd
        df = pd.read_csv(file_path)
        return df.describe().to_string()

# 方式3：装饰器方式
from crewai.tools import tool

@tool("计算器")
def calculate(expression: str) -> str:
    """计算数学表达式"""
    return str(eval(expression))

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,
    max_rpm=10,          # 限制每分钟请求数
    memory=True,
    embedder={           # 使用便宜的嵌入模型
        "provider": "openai",
        "config": {"model": "text-embedding-3-small"}
    }
)