CrewAI框架深度解析：2025年多智能体协作的革命性解决方案

┌─────────────────────────────────────────────────────────────────┐
│                         CrewAI 架构                            │
├─────────────┬─────────────────────────────┬─────────────────────┤
│ 团队层      │ 团队管理器(CrewManager)     │ 目标设定器(GoalSetter) │
│             │ 任务协调器(TaskCoordinator) │ 成果评估器(ResultEvaluator)│
├─────────────┼─────────────────────────────┼─────────────────────┤
│ 智能体层    │ 智能体管理器(AgentManager)  │ 角色分配器(RoleAssigner) │
│             │ 能力评估器(CapabilityEvaluator)│ 技能库(SkillLibrary)   │
├─────────────┼─────────────────────────────┼─────────────────────┤
│ 协作层      │ 通信模块(CommunicationModule)│ 信息共享中心(InformationHub)│
│             │ 冲突解决器(ConflictResolver)│ 协调协议(CoordinationProtocol)│
├─────────────┼─────────────────────────────┼─────────────────────┤
│ 工具层      │ 工具管理器(ToolManager)     │ 工具库(ToolLibrary)    │
│             │ 执行引擎(ExecutionEngine)   │ 资源管理器(ResourceManager)│
└─────────────┴─────────────────────────────┴─────────────────────┘

from crewai import Crew, Agent
from langchain.chat_models import ChatOpenAI

# 创建智能体
researcher = Agent(
    role="研究员",
    goal="收集和分析信息",
    backstory="你是一名经验丰富的研究员，擅长收集和分析各类信息。",
    model=ChatOpenAI(model="gpt-4")
)

writer = Agent(
    role="作家",
    goal="将复杂信息转化为通俗易懂的文章",
    backstory="你是一名优秀的作家，擅长将复杂的信息转化为通俗易懂的文章。",
    model=ChatOpenAI(model="gpt-3.5-turbo")
)

# 创建智能体团队
content_team = Crew(
    agents=[researcher, writer],
    goal="创建一篇关于2025年AI技术趋势的高质量文章",
    verbose=True
)

# 运行团队任务
result = content_team.run()
print(result)

from crewai import Crew, Agent, Tool
from langchain.chat_models import ChatOpenAI
from langchain.tools import DuckDuckGoSearchRun

# 创建搜索工具
search_tool = Tool(
    name="DuckDuckGo搜索",
    func=DuckDuckGoSearchRun().run,
    description="用于搜索互联网上的信息"
)

# 创建带有工具的智能体
researcher = Agent(
    role="高级研究员",
    goal="深入研究特定主题并提供详细分析",
    backstory="你是一名高级研究员，擅长使用各种工具深入研究复杂主题。",
    model=ChatOpenAI(model="gpt-4"),
    tools=[search_tool]
)

analyst = Agent(
    role="数据分析专家",
    goal="分析数据并提取有价值的洞察",
    backstory="你是一名数据分析专家，擅长从复杂数据中提取有价值的信息。",
    model=ChatOpenAI(model="gpt-4")
)

# 创建智能体团队
research_team = Crew(
    agents=[researcher, analyst],
    goal="研究2025年全球气候变化趋势并提供数据分析报告",
    verbose=True
)

# 运行团队任务
result = research_team.run()
print(result)

from crewai import Crew, Agent, Task
from langchain.chat_models import ChatOpenAI

# 创建智能体
planner = Agent(
    role="项目经理",
    goal="制定详细的项目计划",
    backstory="你是一名经验丰富的项目经理，擅长制定详细的项目计划。",
    model=ChatOpenAI(model="gpt-4")
)

developer = Agent(
    role="软件开发者",
    goal="根据计划开发高质量软件",
    backstory="你是一名优秀的软件开发者，擅长根据需求开发高质量的软件。",
    model=ChatOpenAI(model="gpt-4")
)

tester = Agent(
    role="质量测试工程师",
    goal="确保软件质量和性能",
    backstory="你是一名专业的测试工程师，擅长发现和解决软件中的问题。",
    model=ChatOpenAI(model="gpt-3.5-turbo")
)

# 创建任务
planning_task = Task(
    description="为AI聊天机器人项目制定详细的开发计划",
    agent=planner
)

development_task = Task(
    description="根据项目计划开发AI聊天机器人",
    agent=developer,
    dependencies=[planning_task]
)

testing_task = Task(
    description="测试AI聊天机器人，确保其功能正常和性能良好",
    agent=tester,
    dependencies=[development_task]
)

# 创建智能体团队
project_team = Crew(
    agents=[planner, developer, tester],
    tasks=[planning_task, development_task, testing_task],
    verbose=True
)

# 运行团队任务
result = project_team.run()
print(result)

from crewai import Crew, Agent
from langchain.chat_models import ChatOpenAI

# 定义不同角色的智能体
product_manager = Agent(
    role="产品经理",
    goal="定义产品需求和用户体验",
    backstory="你是一名经验丰富的产品经理，擅长理解用户需求并定义产品功能。",
    model=ChatOpenAI(model="gpt-4")
)

ux_designer = Agent(
    role="UI/UX设计师",
    goal="创建直观美观的用户界面",
    backstory="你是一名优秀的UI/UX设计师，擅长创建直观美观且用户友好的界面。",
    model=ChatOpenAI(model="gpt-4")
)

frontend_developer = Agent(
    role="前端开发者",
    goal="实现高质量的前端界面",
    backstory="你是一名专业的前端开发者，擅长将设计稿转化为高质量的前端界面。",
    model=ChatOpenAI(model="gpt-4")
)

backend_developer = Agent(
    role="后端开发者",
    goal="构建稳定高效的后端系统",
    backstory="你是一名资深的后端开发者，擅长构建稳定、高效、可扩展的后端系统。",
    model=ChatOpenAI(model="gpt-4")
)

# 创建跨职能团队
product_team = Crew(
    agents=[product_manager, ux_designer, frontend_developer, backend_developer],
    goal="设计和开发一款创新的移动应用程序",
    verbose=True
)

# 运行团队任务
result = product_team.run()
print(result)

from crewai import Crew, Agent, Task
from langchain.chat_models import ChatOpenAI

# 创建具有特定专长的智能体
data_scientist = Agent(
    role="数据科学家",
    goal="分析数据并提取洞察",
    backstory="你是一名数据科学家，擅长分析复杂数据并提取有价值的洞察。",
    model=ChatOpenAI(model="gpt-4")
)

business_analyst = Agent(
    role="商业分析师",
    goal="将数据洞察转化为商业策略",
    backstory="你是一名商业分析师，擅长理解业务需求并将数据洞察转化为可行的商业策略。",
    model=ChatOpenAI(model="gpt-4")
)

marketing_strategist = Agent(
    role="市场营销策略师",
    goal="制定有效的市场营销策略",
    backstory="你是一名市场营销策略师，擅长制定和执行有效的市场营销策略。",
    model=ChatOpenAI(model="gpt-3.5-turbo")
)

# 创建协作任务
data_analysis_task = Task(
    description="分析客户行为数据，识别购买模式和趋势",
    agent=data_scientist
)

business_strategy_task = Task(
    description="基于数据分析结果，制定提升客户留存率的商业策略",
    agent=business_analyst,
    dependencies=[data_analysis_task]
)

marketing_plan_task = Task(
    description="根据商业策略，制定详细的市场营销计划",
    agent=marketing_strategist,
    dependencies=[business_strategy_task]
)

# 创建协作团队
strategy_team = Crew(
    agents=[data_scientist, business_analyst, marketing_strategist],
    tasks=[data_analysis_task, business_strategy_task, marketing_plan_task],
    verbose=True
)

# 运行团队任务
result = strategy_team.run()
print(result)

from crewai import Crew, Agent, Task
from langchain.chat_models import ChatOpenAI

# 创建基础智能体
team_lead = Agent(
    role="团队领导",
    goal="领导团队完成复杂项目",
    backstory="你是一名经验丰富的团队领导，擅长带领团队解决复杂问题。",
    model=ChatOpenAI(model="gpt-4")
)

# 创建可调整角色的智能体
versatile_agent = Agent(
    role="多面手",
    goal="根据团队需求承担各种角色",
    backstory="你是一名多面手，能够根据团队需求灵活承担各种角色。",
    model=ChatOpenAI(model="gpt-4")
)

# 创建动态任务
def create_dynamic_tasks(project_type):
    tasks = []
    
    # 项目规划任务
    planning_task = Task(
        description=f"规划{project_type}项目的实施路径",
        agent=team_lead
    )
    tasks.append(planning_task)
    
    # 根据项目类型动态调整角色和任务
    if project_type == "市场调研":
        versatile_agent.role = "市场研究员"
        research_task = Task(
            description="进行市场调研，收集和分析市场数据",
            agent=versatile_agent,
            dependencies=[planning_task]
        )
        tasks.append(research_task)
    elif project_type == "产品开发":
        versatile_agent.role = "产品开发者"
        dev_task = Task(
            description="开发和测试新产品功能",
            agent=versatile_agent,
            dependencies=[planning_task]
        )
        tasks.append(dev_task)
    elif project_type == "营销活动":
        versatile_agent.role = "营销专员"
        marketing_task = Task(
            description="策划和执行市场营销活动",
            agent=versatile_agent,
            dependencies=[planning_task]
        )
        tasks.append(marketing_task)
    
    # 总结任务
    summary_task = Task(
        description="总结项目成果并提出改进建议",
        agent=team_lead,
        dependencies=[tasks[-1]]
    )
    tasks.append(summary_task)
    
    return tasks

# 创建自适应团队
adaptive_team = Crew(
    agents=[team_lead, versatile_agent],
    tasks=create_dynamic_tasks("市场调研"),
    verbose=True
)

# 运行团队任务
result = adaptive_team.run()
print(result)

# 调整项目类型，重新配置团队
adaptive_team.tasks = create_dynamic_tasks("产品开发")
result2 = adaptive_team.run()
print(result2)

from crewai import Task

# 创建基本任务
basic_task = Task(
    description="分析2025年AI技术发展趋势",
    agent=researcher
)

# 创建带详细配置的任务
detailed_task = Task(
    description="创建一份关于2025年AI技术趋势的详细报告",
    agent=writer,
    dependencies=[basic_task],  # 依赖其他任务
    context="基于研究员提供的分析结果",  # 任务上下文
    expected_output="一份3000字以上的详细报告，包含图表和案例分析",  # 期望输出
    tools=[search_tool, data_analysis_tool]  # 任务可用工具
)

from crewai import Crew, Agent, Task
from langchain.chat_models import ChatOpenAI

# 创建智能体
project_manager = Agent(
    role="项目经理",
    goal="管理和协调复杂项目",
    backstory="你是一名经验丰富的项目经理，擅长分解和管理复杂项目。",
    model=ChatOpenAI(model="gpt-4")
)

# 创建任务分解函数
def decompose_project(complex_project):
    # 分解复杂项目为子任务
    tasks = []
    
    # 1. 项目启动阶段
    initiation_task = Task(
        description=f"启动{complex_project}项目，制定项目章程和初步计划",
        agent=project_manager
    )
    tasks.append(initiation_task)
    
    # 2. 规划阶段
    planning_task = Task(
        description="制定详细的项目计划，包括时间表、资源分配和风险评估",
        agent=project_manager,
        dependencies=[initiation_task]
    )
    tasks.append(planning_task)
    
    # 3. 执行阶段
    execution_task = Task(
        description="执行项目计划，完成各项项目活动",
        agent=project_manager,
        dependencies=[planning_task]
    )
    tasks.append(execution_task)
    
    # 4. 监控与控制阶段
    monitoring_task = Task(
        description="监控项目进度，控制项目风险，确保项目按计划进行",
        agent=project_manager,
        dependencies=[execution_task]
    )
    tasks.append(monitoring_task)
    
    # 5. 收尾阶段
    closure_task = Task(
        description="完成项目收尾工作，包括验收、文档归档和经验总结",
        agent=project_manager,
        dependencies=[monitoring_task]
    )
    tasks.append(closure_task)
    
    return tasks

# 创建项目管理团队
project_team = Crew(
    agents=[project_manager],
    tasks=decompose_project("企业数字化转型项目"),
    verbose=True
)

# 运行团队任务
result = project_team.run()
print(result)

from crewai import Crew, Agent, Task
from langchain.chat_models import ChatOpenAI
import time

# 创建带监控功能的智能体
monitor_agent = Agent(
    role="项目监控专员",
    goal="监控项目进度并提供实时反馈",
    backstory="你是一名项目监控专员，擅长跟踪项目进度并提供及时反馈。",
    model=ChatOpenAI(model="gpt-3.5-turbo")
)

# 创建任务
execution_task = Task(
    description="执行复杂数据处理任务",
    agent=data_scientist
)

monitor_task = Task(
    description="持续监控数据处理任务的进度，每10分钟提供一次状态更新",
    agent=monitor_agent
)

# 创建监控团队
monitor_team = Crew(
    agents=[data_scientist, monitor_agent],
    tasks=[execution_task, monitor_task],
    verbose=True
)

# 自定义执行逻辑，支持实时监控
def run_with_realtime_monitoring(crew):
    # 启动执行任务
    execution_thread = threading.Thread(target=crew.run_task, args=(execution_task,))
    execution_thread.start()
    
    # 启动监控任务
    for i in range(6):  # 监控1小时
        time.sleep(600)  # 每10分钟监控一次
        status = crew.get_task_status(execution_task)
        print(f"任务状态更新 ({time.strftime('%H:%M:%S')}): {status}")
        
        # 如果任务已完成，停止监控
        if status == "completed":
            break
    
    # 等待执行任务完成
    if execution_thread.is_alive():
        execution_thread.join()
    
    # 获取最终结果
    final_result = crew.get_task_result(execution_task)
    return final_result

# 运行带监控的任务
result = run_with_realtime_monitoring(monitor_team)
print("最终结果:", result)

from crewai import Crew, Agent, Task
from langchain.chat_models import ChatOpenAI
from langchain.tools import Tool
import database

# 创建工具
def query_customer_data(customer_id):
    """查询客户数据"""
    return database.get_customer_info(customer_id)

customer_data_tool = Tool(
    name="客户数据查询",
    func=query_customer_data,
    description="用于查询客户的基本信息、购买历史和服务记录"
)

def process_refund(order_id):
    """处理退款申请"""
    return database.process_refund_request(order_id)

refund_tool = Tool(
    name="退款处理",
    func=process_refund,
    description="用于处理客户的退款申请"
)

# 创建客服团队智能体
frontline_agent = Agent(
    role="一线客服代表",
    goal="快速响应客户咨询，提供基本帮助",
    backstory="你是一名友好的一线客服代表，擅长与客户沟通并解决基本问题。",
    model=ChatOpenAI(model="gpt-3.5-turbo"),
    tools=[customer_data_tool]
)

technical_support_agent = Agent(
    role="技术支持专家",
    goal="解决复杂的技术问题",
    backstory="你是一名技术支持专家，擅长解决各种复杂的技术问题。",
    model=ChatOpenAI(model="gpt-4")
)

billing_agent = Agent(
    role="账单专员",
    goal="处理账单和支付相关问题",
    backstory="你是一名账单专员，擅长处理各种账单查询和支付问题。",
    model=ChatOpenAI(model="gpt-3.5-turbo"),
    tools=[customer_data_tool, refund_tool]
)

# 创建客服流程任务
initial_query_task = Task(
    description="初步了解客户问题并进行分类",
    agent=frontline_agent
)

# 根据问题类型分配任务
def assign_to_specialist(data):
    problem_type = data["problem_type"]
    if problem_type == "technical":
        return technical_support_task
    elif problem_type == "billing":
        return billing_task
    else:
        return resolution_task

technical_support_task = Task(
    description="提供详细的技术支持和解决方案",
    agent=technical_support_agent,
    dependencies=[initial_query_task]
)

billing_task = Task(
    description="处理账单问题或退款申请",
    agent=billing_agent,
    dependencies=[initial_query_task]
)

resolution_task = Task(
    description="提供最终解决方案并确保客户满意",
    agent=frontline_agent,
    dependencies=[technical_support_task, billing_task]
)

# 创建客服团队
customer_service_crew = Crew(
    agents=[frontline_agent, technical_support_agent, billing_agent],
    tasks=[initial_query_task, technical_support_task, billing_task, resolution_task],
    verbose=True
)

# 处理客户咨询
def handle_customer_inquiry(customer_input):
    # 运行客服团队处理咨询
    result = customer_service_crew.run({
        "customer_message": customer_input
    })
    return result

# 示例：处理客户咨询
customer_message = "我的订单#12345一直没有收到，我想申请退款"
response = handle_customer_inquiry(customer_message)
print("客服回复:", response)

from crewai import Crew, Agent, Task
from langchain.chat_models import ChatOpenAI
from langchain.tools import Tool
import market_research
import design_tools
import development_tools

# 创建市场分析工具
market_research_tool = Tool(
    name="市场研究",
    func=market_research.analyze_market_trends,
    description="用于分析市场趋势、客户需求和竞争对手情况"
)

# 创建产品设计工具
design_tool = Tool(
    name="产品设计",
    func=design_tools.generate_product_design,
    description="用于生成产品设计方案和原型"
)

# 创建产品开发工具
development_tool = Tool(
    name="产品开发",
    func=development_tools.generate_product_code,
    description="用于生成产品代码和实现方案"
)

# 创建产品研发团队智能体
product_strategist = Agent(
    role="产品战略专家",
    goal="识别市场机会，定义产品战略",
    backstory="你是一名产品战略专家，擅长识别市场机会并制定产品战略。",
    model=ChatOpenAI(model="gpt-4"),
    tools=[market_research_tool]
)

ux_designer = Agent(
    role="UX设计师",
    goal="创建用户友好的产品设计",
    backstory="你是一名资深UX设计师，擅长创建用户友好且具有吸引力的产品设计。",
    model=ChatOpenAI(model="gpt-4"),
    tools=[design_tool]
)

product_developer = Agent(
    role="产品开发者",
    goal="实现高质量的产品功能",
    backstory="你是一名经验丰富的产品开发者，擅长将设计转化为高质量的产品。",
    model=ChatOpenAI(model="gpt-4"),
    tools=[development_tool]
)

quality_assurance = Agent(
    role="质量保证专员",
    goal="确保产品质量和性能",
    backstory="你是一名质量保证专员，擅长发现和解决产品中的问题。",
    model=ChatOpenAI(model="gpt-3.5-turbo")
)

# 创建产品研发流程任务
market_research_task = Task(
    description="分析智能家居市场趋势和客户需求",
    agent=product_strategist
)

product_definition_task = Task(
    description="定义智能灯泡产品的核心功能和价值主张",
    agent=product_strategist,
    dependencies=[market_research_task]
)

design_task = Task(
    description="设计智能灯泡的用户界面和交互体验",
    agent=ux_designer,
    dependencies=[product_definition_task]
)

development_task = Task(
    description="实现智能灯泡的控制应用程序",
    agent=product_developer,
    dependencies=[design_task]
)

qa_task = Task(
    description="测试智能灯泡应用程序，确保其功能正常和性能良好",
    agent=quality_assurance,
    dependencies=[development_task]
)

# 创建产品研发团队
product_team = Crew(
    agents=[product_strategist, ux_designer, product_developer, quality_assurance],
    tasks=[market_research_task, product_definition_task, design_task, development_task, qa_task],
    verbose=True
)

# 启动产品研发流程
result = product_team.run()
print("产品研发结果:", result)

from crewai import Crew, Agent, Task
from langchain.chat_models import ChatOpenAI
from langchain.tools import Tool
import financial_data
import market_analysis
import risk_assessment

# 创建金融数据工具
financial_data_tool = Tool(
    name="金融数据查询",
    func=financial_data.get_financial_data,
    description="用于查询股票、债券、基金等金融产品的历史数据和基本面信息"
)

# 创建市场分析工具
market_analysis_tool = Tool(
    name="市场分析",
    func=market_analysis.analyze_market_trends,
    description="用于分析市场趋势、行业动态和宏观经济数据"
)

# 创建风险评估工具
risk_assessment_tool = Tool(
    name="风险评估",
    func=risk_assessment.assess_investment_risk,
    description="用于评估投资产品的风险水平和潜在回报"
)

# 创建投资分析团队智能体
economic_analyst = Agent(
    role="经济分析师",
    goal="分析宏观经济趋势和政策影响",
    backstory="你是一名资深经济分析师，擅长分析宏观经济趋势和政策对投资市场的影响。",
    model=ChatOpenAI(model="gpt-4"),
    tools=[market_analysis_tool]
)

fundamental_analyst = Agent(
    role="基本面分析师",
    goal="分析公司基本面和财务状况",
    backstory="你是一名基本面分析师，擅长分析公司的财务状况、业务模式和竞争优势。",
    model=ChatOpenAI(model="gpt-4"),
    tools=[financial_data_tool]
)

technical_analyst = Agent(
    role="技术分析师",
    goal="分析市场价格走势和技术指标",
    backstory="你是一名技术分析师，擅长分析市场价格走势、交易量和各种技术指标。",
    model=ChatOpenAI(model="gpt-4"),
    tools=[financial_data_tool]
)

risk_manager = Agent(
    role="风险经理",
    goal="评估投资风险和构建投资组合",
    backstory="你是一名风险经理，擅长评估投资风险并构建多元化的投资组合。",
    model=ChatOpenAI(model="gpt-4"),
    tools=[risk_assessment_tool]
)

# 创建投资分析流程任务
market_overview_task = Task(
    description="分析当前宏观经济形势和市场整体趋势",
    agent=economic_analyst
)

sector_analysis_task = Task(
    description="分析科技行业的发展趋势和投资机会",
    agent=economic_analyst,
    dependencies=[market_overview_task]
)

stock_selection_task = Task(
    description="选择3-5家具有投资价值的科技公司",
    agent=fundamental_analyst,
    dependencies=[sector_analysis_task]
)

technical_validation_task = Task(
    description="对选定股票进行技术分析，确定买入时机",
    agent=technical_analyst,
    dependencies=[stock_selection_task]
)

portfolio_construction_task = Task(
    description="构建一个平衡的科技股票投资组合，包括权重分配和风险控制策略",
    agent=risk_manager,
    dependencies=[technical_validation_task]
)

# 创建投资分析团队
investment_team = Crew(
    agents=[economic_analyst, fundamental_analyst, technical_analyst, risk_manager],
    tasks=[market_overview_task, sector_analysis_task, stock_selection_task, technical_validation_task, portfolio_construction_task],
    verbose=True
)

# 启动投资分析流程
result = investment_team.run({
    "investment_amount": "100万元",
    "investment_horizon": "3年",
    "risk_tolerance": "中等"
})
print("投资分析结果:", result)