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

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

context

 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   )