CrewAI 角色驱动协作

from crewai import Agent
from crewai_tools import SerperDevTool

# 基础研究 Agent
research_agent = Agent(
    role="AI技术研究员",
    goal="发现并总结AI领域的最新发展动态",
    backstory="你是一位经验丰富的技术研究员，擅长发现前沿技术趋势",
    tools=[SerperDevTool()],
    verbose=True
)

# 代码开发 Agent
dev_agent = Agent(
    role="高级Python开发者",
    goal="编写和调试Python代码",
    backstory="拥有10年经验的Python开发专家",
    allow_code_execution=True,
    code_execution_mode="safe",  # 使用Docker安全执行
    max_execution_time=300,      # 5分钟超时
    max_retry_limit=3            # 复杂代码任务多给重试机会
)

# 数据分析 Agent
analysis_agent = Agent(
    role="数据分析师",
    goal="对大型数据集进行深度分析",
    backstory="专精于大数据分析和模式识别",
    memory=True,
    respect_context_window=True,
    max_rpm=10,                        # 限制API调用频率
    function_calling_llm="gpt-4o-mini" # 工具调用用更便宜的模型
)

from crewai import Task
from pydantic import BaseModel

class Blog(BaseModel):
    title: str
    content: str

# 基础任务
research_task = Task(
    description="研究AI Agent领域的最新发展，确保找到有趣且相关的信息",
    expected_output="包含10个要点的AI Agent最新发展列表",
    agent=research_agent
)

# 带结构化输出的任务
blog_task = Task(
    description="根据研究结果撰写一篇关于AI的博客文章，200字以内",
    expected_output="一篇引人入胜的博客，包含标题和正文",
    agent=blog_agent,
    output_pydantic=Blog,  # 强制输出符合Pydantic模型
)

# 异步任务 + 上下文依赖
research_ai_task = Task(
    description="研究AI最新发展",
    expected_output="AI最新发展列表",
    async_execution=True,  # 异步执行，不阻塞后续任务
    agent=research_agent,
)

research_ops_task = Task(
    description="研究AI Ops最新发展",
    expected_output="AI Ops最新发展列表",
    async_execution=True,  # 异步执行
    agent=research_agent,
)

write_blog_task = Task(
    description="撰写关于AI及其最新动态的博客文章",
    expected_output="一篇4段式的AI文章",
    agent=writer_agent,
    context=[research_ai_task, research_ops_task]  # 等待两个异步任务完成
)

# 带回调的任务
def callback_function(output):
    print(f"任务完成！输出: {output.raw}")

task_with_callback = Task(
    description='总结最新的AI新闻',
    expected_output='5条最重要的AI新闻摘要',
    agent=research_agent,
    callback=callback_function
)

from typing import Tuple, Union, Any
from crewai import TaskOutput

def validate_blog_content(result: TaskOutput) -> Tuple[bool, Any]:
    """验证博客内容是否符合要求"""
    word_count = len(result.raw.split())
    if word_count > 200:
        return (False, "博客内容超过200字限制")
    return (True, result.raw.strip())

blog_task = Task(
    description="撰写关于AI的博客文章",
    expected_output="200字以内的博客文章",
    agent=blog_agent,
    guardrail=validate_blog_content,  # 验证不通过会自动重试
    max_retries=3
)

from crewai import Agent, Crew, Task, Process

# 顺序执行模式
crew = Crew(
    agents=[research_agent, writer_agent],
    tasks=[research_task, write_task],
    process=Process.sequential,  # 任务按顺序执行
    verbose=True
)

# 层级管理模式（需要Manager Agent）
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[task1, task2, task3],
    process=Process.hierarchical,  # Manager Agent 协调分配
    manager_llm="gpt-4o",          # 层级模式下必需
    verbose=True
)

# 自定义管理 Agent
manager = Agent(
    role="项目经理",
    goal="高效协调团队成员完成任务",
    backstory="经验丰富的项目管理者",
    allow_delegation=True
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[task1, task2],
    process=Process.hierarchical,
    manager_agent=manager  # 使用自定义管理Agent
)

# 执行
result = crew.kickoff(inputs={"topic": "AI Agents"})
print(result.raw)

# 异步执行
result = await crew.kickoff_async(inputs={"topic": "AI Agents"})

# 批量执行
results = crew.kickoff_for_each(
    inputs=[{"topic": "AI医疗"}, {"topic": "AI金融"}]
)

from crewai.tools import tool

@tool("搜索工具")
def search_tool(query: str) -> str:
    """根据查询字符串搜索互联网信息。"""
    # 实际搜索逻辑
    return f"搜索结果: {query}的相关信息"

@tool("计算器")
def calculator(expression: str) -> str:
    """计算数学表达式的结果。"""
    try:
        result = eval(expression)  # 注意：生产环境应使用安全的计算方式
        return str(result)
    except Exception as e:
        return f"计算错误: {e}"

from typing import Type
from crewai.tools import BaseTool
from pydantic import BaseModel, Field

class MyToolInput(BaseModel):
    """工具输入模型"""
    argument: str = Field(..., description="输入参数描述")

class MyCustomTool(BaseTool):
    name: str = "自定义工具"
    description: str = "此工具的功能描述，对Agent理解何时使用至关重要"
    args_schema: Type[BaseModel] = MyToolInput

    def _run(self, argument: str) -> str:
        """工具核心逻辑"""
        return f"处理结果: {argument}"

@tool("可缓存的搜索工具")
def cached_search(query: str) -> str:
    """搜索并缓存结果。"""
    return f"搜索结果: {query}"

def cache_strategy(arguments: dict, result: str) -> bool:
    """自定义缓存策略：仅缓存成功结果"""
    return "错误" not in result

cached_search.cache_function = cache_strategy

from crewai import Crew, Process

crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, write_task, edit_task],
    process=Process.sequential
)

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[task1, task2, task3],
    process=Process.hierarchical,
    manager_llm="gpt-4o"
)

from crewai import Crew, Agent, Task, Process
from crewai.memory import LongTermMemory, ShortTermMemory, EntityMemory
from crewai.memory.storage.rag_storage import RAGStorage
from crewai.memory.storage.ltm_sqlite_storage import LTMSqliteStorage

# 基础记忆配置
crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,  # 启用记忆
    verbose=True
)

# 自定义记忆配置
crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    long_term_memory=LongTermMemory(
        storage=LTMSqliteStorage(
            db_path="/my_crew/long_term_memory_storage.db"
        )
    ),
    short_term_memory=ShortTermMemory(
        storage=RAGStorage(
            embedder_config={
                "provider": "openai",
                "config": {"model": "text-embedding-3-small"}
            },
            type="short_term",
            path="/my_crew/"
        )
    ),
    entity_memory=EntityMemory(
        storage=RAGStorage(
            embedder_config={
                "provider": "openai",
                "config": {"model": "text-embedding-3-small"}
            },
            type="short_term",
            path="/my_crew/"
        )
    ),
)

# 集成外部记忆（Mem0）
import os
os.environ["MEM0_API_KEY"] = "m0-xx"

crew = Crew(
    agents=[...],
    tasks=[...],
    memory=True,
    memory_config={
        "provider": "mem0",
        "config": {"user_id": "john"},
        "user_memory": {}
    },
)

# 重置记忆
crew.reset_memories(command_type='all')  # 重置所有
# 可选: 'long', 'short', 'entities', 'kickoff_outputs', 'knowledge'

# Agent 步骤回调
def step_callback(output):
    print(f"Agent步骤完成: {output}")

agent = Agent(
    role="研究员",
    goal="研究AI技术",
    backstory="资深研究员",
    step_callback=step_callback
)

# 任务完成回调
def task_callback(output):
    print(f"任务完成: {output.raw}")

task = Task(
    description="研究AI最新动态",
    expected_output="AI动态列表",
    agent=agent,
    callback=task_callback
)

# Crew 级别回调
crew = Crew(
    agents=[agent],
    tasks=[task],
    step_callback=step_callback,    # 所有Agent步骤后触发
    task_callback=task_callback,    # 所有任务完成后触发
)

from crewai.project import CrewBase, agent, task, crew, before_kickoff, after_kickoff

@CrewBase
class MyCrew:
    agents_config = 'config/agents.yaml'
    tasks_config = 'config/tasks.yaml'

    @before_kickoff
    def prepare_inputs(self, inputs):
        inputs['additional_data'] = "额外信息"
        return inputs

    @after_kickoff
    def process_output(self, output):
        output.raw += "\n后处理完成"
        return output

    @agent
    def researcher(self) -> Agent:
        return Agent(config=self.agents_config['researcher'], verbose=True)

    @crew
    def crew(self) -> Crew:
        return Crew(agents=self.agents, tasks=self.tasks, process=Process.sequential)

crew = Crew(
    agents=[researcher, writer],
    tasks=[task1, task2],
    planning=True,          # 启用规划
    planning_llm="gpt-4o",  # 规划使用的模型
    process=Process.sequential
)

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

# 定义 Agent
researcher = Agent(
    role="技术研究员",
    goal="发现{topic}领域的前沿技术和趋势",
    backstory="你是一位专注于技术趋势的资深研究员，擅长信息检索和整理",
    tools=[SerperDevTool()],
    verbose=True
)

analyst = Agent(
    role="技术分析师",
    goal="分析研究数据，提取关键洞察",
    backstory="你是一位敏锐的技术分析师，擅长从数据中发现模式和价值",
    verbose=True
)

writer = Agent(
    role="技术撰稿人",
    goal="将分析结果转化为清晰易懂的文章",
    backstory="你是一位经验丰富的技术作家，擅长将复杂概念简明表达",
    verbose=True
)

# 定义任务
research_task = Task(
    description="研究{topic}领域的最新发展，找出5个最重要的趋势",
    expected_output="包含5个关键趋势的详细列表，每个趋势附有简短说明",
    agent=researcher
)

analysis_task = Task(
    description="基于研究结果，分析各趋势的影响力和潜在价值",
    expected_output="趋势影响力分析报告，包含优先级排序和理由",
    agent=analyst,
    context=[research_task]
)

writing_task = Task(
    description="将分析报告转化为一篇结构清晰的技术文章",
    expected_output="一篇800字左右的技术洞察文章，包含标题、引言、主体和结论",
    agent=writer,
    context=[research_task, analysis_task],
    output_file="output/tech_insight.md"
)

# 组建团队
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.sequential,
    verbose=True
)

# 执行
result = crew.kickoff(inputs={"topic": "大语言模型Agent"})
print(result.raw)

from crewai import Agent, Crew, Task, Process

reviewer = Agent(
    role="代码审查专家",
    goal="识别代码中的问题和改进点",
    backstory="10年代码审查经验，精通代码质量标准和安全最佳实践",
    allow_code_execution=True,
    code_execution_mode="safe",
    verbose=True
)

optimizer = Agent(
    role="性能优化专家",
    goal="优化代码性能和可维护性",
    backstory="专精于性能调优和重构，深入理解算法复杂度",
    allow_code_execution=True,
    verbose=True
)

review_task = Task(
    description="审查提供的Python代码，找出潜在问题和安全漏洞",
    expected_output="代码审查报告，包含问题列表和严重程度评级",
    agent=reviewer
)

optimize_task = Task(
    description="基于审查结果，提供优化后的代码版本",
    expected_output="优化后的代码及改动说明",
    agent=optimizer,
    context=[review_task]
)

crew = Crew(
    agents=[reviewer, optimizer],
    tasks=[review_task, optimize_task],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff(inputs={"code": "def process(data): ..."})

from crewai import Agent, Crew, Task, Process

triage_agent = Agent(
    role="客服分流专员",
    goal="快速分析客户问题并分配给合适的处理人员",
    backstory="经验丰富的客服分流专员，擅长快速判断问题类型和优先级",
    verbose=True
)

tech_support = Agent(
    role="技术支持工程师",
    goal="解决技术类客户问题",
    backstory="技术支持专家，精通产品技术细节和故障排除",
    verbose=True
)

billing_agent = Agent(
    role="账务处理专员",
    goal="处理账单和退款相关问题",
    backstory="账务处理专家，熟悉公司退费政策和账单系统",
    verbose=True
)

# 使用层级模式，Manager自动分配
crew = Crew(
    agents=[triage_agent, tech_support, billing_agent],
    tasks=[
        Task(description="处理客户工单", expected_output="工单处理结果", agent=triage_agent),
        Task(description="解决技术问题", expected_output="技术问题解决方案", agent=tech_support),
        Task(description="处理账务请求", expected_output="账务处理结果", agent=billing_agent),
    ],
    process=Process.hierarchical,
    manager_llm="gpt-4o",
    memory=True  # 记住客户历史交互
)