协作

如何在 CrewAI 团队中让 Agent 彼此协作、委派任务并高效沟通。

概览

CrewAI 中的协作机制使 Agent 能够像团队一样工作,通过委派任务和相互提问来利用彼此的专业能力。当设置 allow_delegation=True 时,Agent 会自动获得强大的协作工具。

快速开始:启用协作

  1. from crewai import Agent, Crew, Task
  2. # Enable collaboration for agents
  3. researcher = Agent(
  4. role="Research Specialist",
  5. goal="Conduct thorough research on any topic",
  6. backstory="Expert researcher with access to various sources",
  7. allow_delegation=True, # 🔑 Key setting for collaboration
  8. verbose=True
  9. )
  10. writer = Agent(
  11. role="Content Writer",
  12. goal="Create engaging content based on research",
  13. backstory="Skilled writer who transforms research into compelling content",
  14. allow_delegation=True, # 🔑 Enables asking questions to other agents
  15. verbose=True
  16. )
  17. # Agents can now collaborate automatically
  18. crew = Crew(
  19. agents=[researcher, writer],
  20. tasks=[...],
  21. verbose=True
  22. )

Agent 协作如何工作

当设置 allow_delegation=True 时,CrewAI 会自动为 Agent 提供两个强大的工具:

1. Delegate Work Tool

允许 Agent 将任务分配给具备特定专长的队友。

  1. # Agent automatically gets this tool:
  2. # Delegate work to coworker(task: str, context: str, coworker: str)

2. Ask Question Tool

允许 Agent 向同事提出具体问题,以获取所需信息。

  1. # Agent automatically gets this tool:
  2. # Ask question to coworker(question: str, context: str, coworker: str)

协作实战

下面是一个完整示例,展示 Agent 如何协作完成内容创作任务:

  1. from crewai import Agent, Crew, Task, Process
  2. # Create collaborative agents
  3. researcher = Agent(
  4. role="Research Specialist",
  5. goal="Find accurate, up-to-date information on any topic",
  6. backstory="""You're a meticulous researcher with expertise in finding
  7. reliable sources and fact-checking information across various domains.""",
  8. allow_delegation=True,
  9. verbose=True
  10. )
  11. writer = Agent(
  12. role="Content Writer",
  13. goal="Create engaging, well-structured content",
  14. backstory="""You're a skilled content writer who excels at transforming
  15. research into compelling, readable content for different audiences.""",
  16. allow_delegation=True,
  17. verbose=True
  18. )
  19. editor = Agent(
  20. role="Content Editor",
  21. goal="Ensure content quality and consistency",
  22. backstory="""You're an experienced editor with an eye for detail,
  23. ensuring content meets high standards for clarity and accuracy.""",
  24. allow_delegation=True,
  25. verbose=True
  26. )
  27. # Create a task that encourages collaboration
  28. article_task = Task(
  29. description="""Write a comprehensive 1000-word article about 'The Future of AI in Healthcare'.
  30. The article should include:
  31. - Current AI applications in healthcare
  32. - Emerging trends and technologies
  33. - Potential challenges and ethical considerations
  34. - Expert predictions for the next 5 years
  35. Collaborate with your teammates to ensure accuracy and quality.""",
  36. expected_output="A well-researched, engaging 1000-word article with proper structure and citations",
  37. agent=writer # Writer leads, but can delegate research to researcher
  38. )
  39. # Create collaborative crew
  40. crew = Crew(
  41. agents=[researcher, writer, editor],
  42. tasks=[article_task],
  43. process=Process.sequential,
  44. verbose=True
  45. )
  46. result = crew.kickoff()

协作模式

模式 1:研究 → 写作 → 编辑

  1. research_task = Task(
  2. description="Research the latest developments in quantum computing",
  3. expected_output="Comprehensive research summary with key findings and sources",
  4. agent=researcher
  5. )
  6. writing_task = Task(
  7. description="Write an article based on the research findings",
  8. expected_output="Engaging 800-word article about quantum computing",
  9. agent=writer,
  10. context=[research_task] # Gets research output as context
  11. )
  12. editing_task = Task(
  13. description="Edit and polish the article for publication",
  14. expected_output="Publication-ready article with improved clarity and flow",
  15. agent=editor,
  16. context=[writing_task] # Gets article draft as context
  17. )

模式 2:单任务协作

  1. collaborative_task = Task(
  2. description="""Create a marketing strategy for a new AI product.
  3. Writer: Focus on messaging and content strategy
  4. Researcher: Provide market analysis and competitor insights
  5. Work together to create a comprehensive strategy.""",
  6. expected_output="Complete marketing strategy with research backing",
  7. agent=writer # Lead agent, but can delegate to researcher
  8. )

分层协作

对于复杂项目,可以使用带有 manager Agent 的分层流程:

  1. from crewai import Agent, Crew, Task, Process
  2. # Manager agent coordinates the team
  3. manager = Agent(
  4. role="Project Manager",
  5. goal="Coordinate team efforts and ensure project success",
  6. backstory="Experienced project manager skilled at delegation and quality control",
  7. allow_delegation=True,
  8. verbose=True
  9. )
  10. # Specialist agents
  11. researcher = Agent(
  12. role="Researcher",
  13. goal="Provide accurate research and analysis",
  14. backstory="Expert researcher with deep analytical skills",
  15. allow_delegation=False, # Specialists focus on their expertise
  16. verbose=True
  17. )
  18. writer = Agent(
  19. role="Writer",
  20. goal="Create compelling content",
  21. backstory="Skilled writer who creates engaging content",
  22. allow_delegation=False,
  23. verbose=True
  24. )
  25. # Manager-led task
  26. project_task = Task(
  27. description="Create a comprehensive market analysis report with recommendations",
  28. expected_output="Executive summary, detailed analysis, and strategic recommendations",
  29. agent=manager # Manager will delegate to specialists
  30. )
  31. # Hierarchical crew
  32. crew = Crew(
  33. agents=[manager, researcher, writer],
  34. tasks=[project_task],
  35. process=Process.hierarchical, # Manager coordinates everything
  36. manager_llm="gpt-4o", # Specify LLM for manager
  37. verbose=True
  38. )

协作最佳实践

1. 清晰定义角色

  1. # ✅ Good: Specific, complementary roles
  2. researcher = Agent(role="Market Research Analyst", ...)
  3. writer = Agent(role="Technical Content Writer", ...)
  4. # ❌ Avoid: Overlapping or vague roles
  5. agent1 = Agent(role="General Assistant", ...)
  6. agent2 = Agent(role="Helper", ...)

2. 有策略地启用委派

  1. # ✅ Enable delegation for coordinators and generalists
  2. lead_agent = Agent(
  3. role="Content Lead",
  4. allow_delegation=True, # Can delegate to specialists
  5. ...
  6. )
  7. # ✅ Disable for focused specialists (optional)
  8. specialist_agent = Agent(
  9. role="Data Analyst",
  10. allow_delegation=False, # Focuses on core expertise
  11. ...
  12. )

3. 共享上下文

  1. # ✅ Use context parameter for task dependencies
  2. writing_task = Task(
  3. description="Write article based on research",
  4. agent=writer,
  5. context=[research_task], # Shares research results
  6. ...
  7. )

4. 清晰的任务描述

  1. # ✅ Specific, actionable descriptions
  2. Task(
  3. description="""Research competitors in the AI chatbot space.
  4. Focus on: pricing models, key features, target markets.
  5. Provide data in a structured format.""",
  6. ...
  7. )
  8. # ❌ Vague descriptions that don't guide collaboration
  9. Task(description="Do some research about chatbots", ...)

协作问题排查

问题:Agent 没有发生协作

现象: Agent 各自独立工作,没有发生任务委派

  1. # ✅ Solution: Ensure delegation is enabled
  2. agent = Agent(
  3. role="...",
  4. allow_delegation=True, # This is required!
  5. ...
  6. )

问题:来回沟通过多

现象: Agent 提问过于频繁,导致进度缓慢

  1. # ✅ Solution: Provide better context and specific roles
  2. Task(
  3. description="""Write a technical blog post about machine learning.
  4. Context: Target audience is software developers with basic ML knowledge.
  5. Length: 1200 words
  6. Include: code examples, practical applications, best practices
  7. If you need specific technical details, delegate research to the researcher.""",
  8. ...
  9. )

问题:委派循环

现象: Agent 之间来回反复委派,无法结束

  1. # ✅ Solution: Clear hierarchy and responsibilities
  2. manager = Agent(role="Manager", allow_delegation=True)
  3. specialist1 = Agent(role="Specialist A", allow_delegation=False) # No re-delegation
  4. specialist2 = Agent(role="Specialist B", allow_delegation=False)

高级协作功能

自定义协作规则

  1. # Set specific collaboration guidelines in agent backstory
  2. agent = Agent(
  3. role="Senior Developer",
  4. backstory="""You lead development projects and coordinate with team members.
  5. Collaboration guidelines:
  6. - Delegate research tasks to the Research Analyst
  7. - Ask the Designer for UI/UX guidance
  8. - Consult the QA Engineer for testing strategies
  9. - Only escalate blocking issues to the Project Manager""",
  10. allow_delegation=True
  11. )

监控协作过程

  1. def track_collaboration(output):
  2. """Track collaboration patterns"""
  3. if "Delegate work to coworker" in output.raw:
  4. print("🤝 Delegation occurred")
  5. if "Ask question to coworker" in output.raw:
  6. print("❓ Question asked")
  7. crew = Crew(
  8. agents=[...],
  9. tasks=[...],
  10. step_callback=track_collaboration, # Monitor collaboration
  11. verbose=True
  12. )

记忆与学习

启用 Agent 记忆,让它们记住过去的协作过程:

  1. agent = Agent(
  2. role="Content Lead",
  3. memory=True, # Remembers past interactions
  4. allow_delegation=True,
  5. verbose=True
  6. )

启用记忆后,Agent 可以从过去的协作中学习,并随着时间推移不断优化其委派决策。

下一步

  • 尝试这些示例:先从基础协作示例开始
  • 实验不同角色:测试不同的 Agent 角色组合
  • 观察交互过程:使用 verbose=True 查看协作如何发生
  • 优化任务描述:清晰的任务会带来更好的协作效果
  • 逐步扩展:对于复杂项目,尝试分层流程

协作能够把单个 AI Agent 转变为强大的团队,使它们可以共同应对复杂、多维度的挑战。