CrewAI 教程：构建 AI 团队自动化复杂任务

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool
import os

# 设置环境变量 (或者在代码中直接设置)
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"
# os.environ["SERPER_API_KEY"] = "YOUR_SERPER_API_KEY" # 用于网络搜索工具

# 1. 定义工具
search_tool = SerperDevTool()

# 2. 定义代理
researcher = Agent(
    role='高级市场研究员',
    goal='发现2026年AI市场最新趋势和关键技术突破',
    backstory='一位经验丰富的市场分析师，擅长深入挖掘数据并识别新兴模式。',
    verbose=True,
    allow_delegation=False,
    tools=[search_tool],
    llm=None # 可以指定具体的LLM，例如 Multi AI 中的 gpt-5-chat
)

writer = Agent(
    role='内容策略师',
    goal='根据研究结果撰写引人入胜的博客文章，分析AI趋势对商业的影响',
    backstory='一位屡获殊荣的内容创作者，擅长将复杂的技术概念转化为清晰易懂的叙述。',
    verbose=True,
    allow_delegation=True,
    llm=None # 可以指定具体的LLM，例如 Multi AI 中的 claude-opus-4-6
)

# 3. 定义任务
research_task = Task(
    description='对2026年AI市场进行全面研究，重点关注生成式AI、多模态模型和边缘AI的最新进展。识别至少5个关键趋势。',
    expected_output='一份详细的市场研究报告，包含关键趋势、数据支持和未来预测。',
    agent=researcher
)

write_task = Task(
    description='根据市场研究报告，撰写一篇关于“2026年AI趋势如何重塑商业格局”的博客文章。文章应具有吸引力，信息丰富，并包含实际案例。',
    expected_output='一篇至少1000字的博客文章，结构清晰，语言流畅，适合商业受众。',
    agent=writer,
    context=[research_task] # 确保写作任务在研究任务完成后开始
)

# 4. 创建团队
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential, # 任务按顺序执行
    verbose=2 # 详细输出日志
)

# 5. 运行团队
result = crew.kickoff()
print("\n\n------------------------------------------------------")
print("CrewAI 团队已完成任务，最终输出：")
print(result)

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, FileReadTool
from pydantic import BaseModel, Field
import os

# 确保已设置 API 密钥
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"
# os.environ["SERPER_API_KEY"] = "YOUR_SERPER_API_KEY"

# 定义报告输出的 Pydantic 模型
class MarketAnalysisReport(BaseModel):
    title: str = Field(description="报告标题，总结主要发现")
    executive_summary: str = Field(description="执行摘要，概述报告关键点和建议")
    key_trends: list[str] = Field(description="识别出的主要市场趋势，每项趋势以短句描述")
    data_points: list[str] = Field(description="支持关键趋势的数据点或市场事实")
    recommendations: list[str] = Field(description="基于分析提出的具体商业建议")

# 工具定义
search_tool = SerperDevTool()
file_read_tool = FileReadTool()

# 代理定义
data_analyst = Agent(
    role='数据分析师',
    goal='收集并分析最新的市场数据和行业报告，识别关键指标。',
    backstory='一位精通数据挖掘和统计分析的专家，能够从海量信息中提取有价值的见解。',
    verbose=True,
    allow_delegation=False,
    tools=[search_tool],
    llm=None # 例如使用 Google Gemini 3.1 Pro Preview
)

strategist = Agent(
    role='业务策略师',
    goal='根据数据分析结果，制定可行的商业策略和市场进入建议。',
    backstory='一位经验丰富的商业顾问，擅长将市场趋势转化为战略行动。',
    verbose=True,
    allow_delegation=True,
    llm=None # 例如使用 OpenAI GPT-5 Chat
)

report_writer = Agent(
    role='报告撰写员',
    goal='将分析结果和策略建议整合为一份结构清晰、引人入胜的市场分析报告。',
    backstory='一位专业的报告撰写员，擅长将复杂信息清晰地呈现给高层决策者。',
    verbose=True,
    allow_delegation=False,
    llm=None # 例如使用 Anthropic Claude Opus 4.6
)

# 任务定义
collect_data_task = Task(
    description='使用搜索工具收集关于“2026年全球AI芯片市场”的最新数据和分析报告。寻找市场规模、增长率、主要参与者和技术创新。',
    expected_output='一份包含原始数据链接和关键数据点的简要列表。',
    agent=data_analyst
)

analyze_data_task = Task(
    description='分析收集到的数据，识别AI芯片市场的核心趋势、增长驱动因素和潜在挑战。',
    expected_output='一份详细的数据分析总结，突出显示2-3个关键市场趋势和其影响。',
    agent=data_analyst,
    context=[collect_data_task]
)

develop_strategy_task = Task(
    description='基于AI芯片市场分析，制定针对新进入者的市场策略建议，包括产品定位和潜在合作伙伴。',
    expected_output='一份包含3-5条具体商业策略建议的列表。',
    agent=strategist,
    context=[analyze_data_task]
)

write_report_task = Task(
    description='综合所有研究和策略建议，撰写一份全面的市场分析报告。报告必须严格遵循 MarketAnalysisReport Pydantic 模型的结构。',
    expected_output='一份格式为 MarketAnalysisReport Pydantic 模型的市场分析报告。',
    agent=report_writer,
    context=[analyze_data_task, develop_strategy_task],
    output_pydantic=MarketAnalysisReport
)

# 创建团队
market_analysis_crew = Crew(
    agents=[data_analyst, strategist, report_writer],
    tasks=[collect_data_task, analyze_data_task, develop_strategy_task, write_report_task],
    process=Process.sequential,
    verbose=2
)

# 运行团队
report_result = market_analysis_crew.kickoff()
print("\n\n------------------------------------------------------")
print("AI 市场分析报告已生成：")
print(report_result.model_dump_json(indent=2))