CrewAI 上手攻略：多 Agent 自动化处理复杂任务，让 AI 像员工一样分工协作

 # Install CrewAI  
 pip install crewai crewai-tools
 
 # For additional tools  
 pip install 'crewai[tools]'

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

# Initialize tools  
search_tool = SerperDevTool()  
web_tool = WebsiteSearchTool()

# Create Agents  
researcher = Agent(  
    role='Research Analyst',  
    goal='Gather comprehensive information on {topic}',  
    backstory='You are an expert researcher with a keen eye for detail and accuracy.',  
    tools=[search_tool, web_tool],  
    verbose=True,  
    allow_delegation=False  
)

writer = Agent(  
    role='Content Writer',  
    goal='Create engaging, well-structured content about {topic}',  
    backstory='You are a skilled writer who transforms research into compelling narratives.',  
    verbose=True,  
    allow_delegation=False  
)

editor = Agent(  
    role='Editor',  
    goal='Refine and polish content to ensure quality and clarity',  
    backstory='You are a meticulous editor with an eye for grammar, style, and flow.',  
    verbose=True,  
    allow_delegation=False  
)

# Define Tasks  
research_task = Task(  
    description='Research {topic} and gather key facts, statistics, and insights.',  
    expected_output='A comprehensive research report with sources',  
    agent=researcher  
)

writing_task = Task(  
    description='Using the research, write a 500-word blog post about {topic}',  
    expected_output='A well-written blog post in markdown format',  
    agent=writer,  
    context=[research_task]  # Depends on research task  
)

editing_task = Task(  
    description='Edit the blog post for grammar, clarity, and engagement',  
    expected_output='A polished, publication-ready blog post',  
    agent=editor,  
    context=[writing_task]  
)

# Create Crew  
crew = Crew(  
    agents=[researcher, writer, editor],  
    tasks=[research_task, writing_task, editing_task],  
    process=Process.sequential,  # Tasks run in order  
    verbose=True  
)

# Execute  
result = crew.kickoff(inputs={'topic': 'Artificial Intelligence in Healthcare'})  
 print(result)

 from crewai import Agent, Task, Crew  
from crewai_tools import FileReadTool, CodeInterpreterTool

# Tools  
file_tool = FileReadTool()  
code_tool = CodeInterpreterTool()

# Agents with specific expertise  
architect = Agent(  
    role='Software Architect',  
    goal='Design scalable software architecture for {project}',  
    backstory='Senior architect with 15 years of experience in system design.',  
    verbose=True  
)

developer = Agent(  
    role='Python Developer',  
    goal='Write clean, efficient Python code',  
    backstory='Expert Python developer focused on best practices.',  
    tools=[code_tool],  
    verbose=True  
)

qa_engineer = Agent(  
    role='QA Engineer',  
    goal='Ensure code quality through comprehensive testing',  
    backstory='Detail-oriented QA engineer specializing in test automation.',  
    tools=[code_tool],  
    verbose=True  
)

# Tasks  
design_task = Task(  
    description='Design architecture for a {project} including component breakdown',  
    expected_output='Detailed architecture document with diagrams',  
    agent=architect  
)

development_task = Task(  
    description='Implement the core functionality based on the architecture',  
    expected_output='Working Python code with documentation',  
    agent=developer,  
    context=[design_task]  
)

testing_task = Task(  
    description='Write and execute unit tests for the developed code',  
    expected_output='Test suite with coverage report',  
    agent=qa_engineer,  
    context=[development_task]  
)

# Hierarchical process with manager agent  
dev_crew = Crew(  
    agents=[architect, developer, qa_engineer],  
    tasks=[design_task, development_task, testing_task],  
    process=Process.hierarchical,  # Manager coordinates tasks  
    manager_llm='gpt-4',  # Manager uses GPT-4  
    verbose=True  
)

 result = dev_crew.kickoff(inputs={'project': 'RESTful API for task management'})

 # Run crew asynchronously for better performance  
 result = await crew.kickoff_async(inputs={'topic': 'AI trends'})
 
 # Run specific tasks in parallel  
 from crewai import Task  
 task1 = Task(description='Research topic A', agent=researcher, async_execution=True)  
 task2 = Task(description='Research topic B', agent=researcher, async_execution=True)

 agent = Agent(  
     role='Decision Maker',  
     goal='Make strategic decisions',  
     human_input=True  # Will prompt for human feedback  
 )

 from crewai import Task  
from pydantic import BaseModel

class BlogPost(BaseModel):  
    title: str  
    content: str  
    tags: list[str]

task = Task(  
    description='Write a blog post',  
    expected_output='Blog post with title and tags',  
    agent=writer,  
    output_json=BlogPost,  # Structured output  
    output_file='output.json'  # Save to file  
 )