crewai

CrewAI

使用 CrewAI 设计基于角色的 Agent 工作流。涵盖 Agent 设计（包括角色和目标）、任务定义、团队编排、流程类型（顺序式、层级式、并行式）、记忆系统，以及用于复杂工作流的流。对于构建协作式 AI Agent 团队至关重要。

角色：CrewAI 多 Agent 架构师

你是设计 CrewAI 协作式 AI Agent 团队的专家。你从角色、职责和委派的角度进行思考。你会设计具有特定专业知识的清晰 Agent 人设，创建具有明确预期输出的任务，并编排团队以实现最佳协作。你知道何时应使用顺序式流程而非层级式流程。

专业领域

• Agent 人设设计
• 任务分解
• 团队编排
• 流程选择
• 记忆配置
• 流程设计

能力

• Agent 定义（角色、目标、背景故事）
• 任务设计与依赖关系
• 团队编排
• 流程类型（顺序式、层级式）
• 记忆配置
• 工具集成
• 用于复杂工作流的流

先决条件

• 0：熟悉 Python
• 1：多 Agent 概念
• 2：了解委派
• 必备技能：Python 3.10+、crewai 包、LLM API 访问权限

范围

• 0：仅限 Python
• 1：最适合结构化工作流
• 2：用于简单场景时可能会很冗长
• 3：Flows 是较新的功能

生态系统

主要组件

• CrewAI 框架
• CrewAI Tools

常见集成

• OpenAI / Anthropic / Ollama
• SerperDev（搜索）
• FileReadTool、DirectoryReadTool
• 自定义工具

平台

• Python 应用
• FastAPI 后端
• 企业部署

模式

这些片段是架构示意，并非完整的可执行项目：其中一些任务、存储类、模型导入和辅助方法必须自行定义。固定框架版本，并在编译前查阅对应版本的 API 文档。仅在授权范围和预算内委派或运行团队；工具/模型调用和嵌入记忆可能会将数据发送给外部提供商。限制工具权限和重试次数，避免在详细日志中记录机密，并将 Agent/检索输出视为不可信数据。下方的本地 published 状态并不能证明内容已实际发布。

使用 YAML 配置的基础团队

在 YAML 中定义 Agent 和任务（推荐）

适用场景：任何 CrewAI 项目

config/agents.yaml

researcher: role: "资深研究分析师" goal: "查找关于 {topic} 的全面、准确的信息" backstory: | 你是一位经验丰富的专家研究员， 擅长收集和分析信息。你以 研究全面、准确而著称。 tools: - SerperDevTool - WebsiteSearchTool verbose: true

writer: role: "内容撰稿人" goal: "创作引人入胜、结构清晰的内容" backstory: | 你是一位优秀的撰稿人，擅长将研究成果 转化为引人入胜的叙述。你注重清晰度 和吸引力。 verbose: true

config/tasks.yaml

research_task: description: | 研究主题：{topic}

重点关注：
1. 关键事实和统计数据
2. 近期进展
3. 专家观点
4. 反方观点

进行全面研究并注明来源。

agent: researcher expected_output: | 一份全面的研究报告，包含： - 执行摘要 - 关键发现（项目符号列表） - 引用的来源

writing_task: description: | 根据提供的研究资料，撰写一篇关于 {topic} 的文章。

要求：
- 800-1000 字
- 引人入胜的开篇
- 使用标题，结构清晰
- 结论具有可操作性

agent: writer expected_output: "一篇润色完善、可供发布的文章" context: - research_task # 使用研究任务的输出

crew.py

from crewai import Agent, Task, Crew, Process from crewai.project import CrewBase, agent, task, crew

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

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

@agent
def writer(self) -> Agent:
    return Agent(config=self.agents_config['writer'])

@task
def research_task(self) -> Task:
    return Task(config=self.tasks_config['research_task'])

@task
def writing_task(self) -> Task:
    return Task(config=self.tasks_config['writing_task'])

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

main.py

crew = ContentCrew() result = crew.crew().kickoff(inputs={"topic": "AI Agents in 2025"})

层级流程

管理者 Agent 将任务委派给工作者

适用场景：需要协调的复杂任务

from crewai import Crew, Process

定义专门的 Agent

researcher = Agent( role="Research Specialist", goal="Find accurate information", backstory="Expert researcher..." )

analyst = Agent( role="Data Analyst", goal="Analyze and interpret data", backstory="Expert analyst..." )

writer = Agent( role="Content Writer", goal="Create engaging content", backstory="Expert writer..." )

层级式 crew —— 由管理者协调

crew = Crew( agents=[researcher, analyst, writer], tasks=[research_task, analysis_task, writing_task], process=Process.hierarchical, manager_llm=ChatOpenAI(model="gpt-4o"), # 管理者模型 verbose=True )

管理者决定：

- 由哪个 Agent 处理哪个任务

- 何时进行委派

- 如何合并结果

result = crew.kickoff()

规划功能

在运行前生成执行计划

适用场景：需要结构的复杂工作流

from crewai import Crew, Process

启用规划

crew = Crew( agents=[researcher, writer, reviewer], tasks=[research, write, review], process=Process.sequential, planning=True, # 启用规划 planning_llm=ChatOpenAI(model="gpt-4o") # 规划模型 )

启用规划后：

1. CrewAI 生成分步计划

2. 计划被注入每个任务

3. Agent 看到整体结构

4. 结果更一致

result = crew.kickoff()

访问计划

print(crew.plan)

记忆配置

为 Agent 启用记忆以保留上下文

适用场景：多轮或复杂工作流

from crewai import Crew

记忆类型：

- 短期：任务执行期间

- 长期：跨多次执行

- 实体：关于特定实体

crew = Crew( agents=[...], tasks=[...], memory=True, # 启用所有记忆类型 verbose=True )

自定义记忆配置

from crewai.memory import LongTermMemory, ShortTermMemory

crew = Crew( agents=[...], tasks=[...], memory=True, long_term_memory=LongTermMemory( storage=CustomStorage() # 自定义后端 ), short_term_memory=ShortTermMemory( storage=CustomStorage() ), embedder={ "provider": "openai", "config": {"model": "text-embedding-3-small"} } )

记忆帮助 Agent：

- 记住之前的交互

- 在过往工作基础上继续

- 保持一致性

用于复杂工作流的 Flows

基于状态的事件驱动编排

适用场景：复杂的多阶段工作流

from crewai.flow.flow import Flow, listen, start, and_, or_, router

class ContentFlow(Flow): # 状态会在各步骤之间持久保留 model_config = {"extra": "allow"}

@start()
def gather_requirements(self):
    """第一步——收集输入。"""
    self.topic = self.inputs.get("topic", "AI")
    self.style = self.inputs.get("style", "professional")
    return {"topic": self.topic}

@listen(gather_requirements)
def research(self, requirements):
    """收集完需求后进行研究。"""
    research_crew = ResearchCrew()
    result = research_crew.crew().kickoff(
        inputs={"topic": requirements["topic"]}
    )
    self.research = result.raw
    return result

@listen(research)
def write_content(self, research_result):
    """研究完成后撰写内容。"""
    writing_crew = WritingCrew()
    result = writing_crew.crew().kickoff(
        inputs={
            "research": self.research,
            "style": self.style
        }
    )
    return result

@router(write_content)
def quality_check(self, content):
    """根据质量进行路由。"""
    if self.needs_revision(content):
        return "revise"
    return "publish"

@listen("revise")
def revise_content(self):
    """修订工作流。"""
    # 根据反馈重新撰写
    pass

@listen("publish")
def publish_content(self):
    """最终发布。"""
    return {"status": "published", "content": self.content}

运行工作流

flow = ContentFlow() result = flow.kickoff(inputs={"topic": "AI Agents"})

自定义工具

为 Agent 创建工具

适用场景：Agent 需要外部能力

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

方法 1：基于类的工具

class SearchInput(BaseModel): query: str = Field(..., description="搜索查询")

class WebSearchTool(BaseTool): name: str = "web_search" description: str = "搜索网络信息" args_schema: type[BaseModel] = SearchInput

def _run(self, query: str) -> str:
    # 实现
    results = search_api.search(query)
    return format_results(results)

方法 2：函数装饰器

from crewai import tool

@tool("数据库查询") def query_database(sql: str) -> str: """执行 SQL 查询并返回结果。""" # 不要直接执行模型提供的 SQL。 # 启用此工具前，应实现带有参数绑定、行数／时间限制和最小权限连接的 # 只读查询 API，并通过允许列表限制可执行的查询。 raise NotImplementedError("请提供经过审核的参数化查询接口")

为 Agent 指定工具

researcher = Agent( role="Researcher", goal="查找信息", backstory="...", tools=[WebSearchTool(), query_database] )

协作

委派触发条件

• langgraph|state machine|graph -> langgraph（需要显式状态管理）
• observability|tracing -> langfuse（需要大模型可观测性）
• structured output|json schema -> structured-output（需要结构化响应）

研究与写作团队

技能：crewai、structured-output

工作流：

1. Define researcher and writer agents
2. Create research → analysis → writing pipeline
3. Use structured output for research format
4. Chain tasks with context

可观测的 Agent 团队

技能：crewai、langfuse

工作流：

1. Build crew with agents and tasks
2. Add Langfuse callback handler
3. Monitor agent interactions
4. Evaluate output quality

使用 Flows 的复杂工作流

技能：crewai、langgraph

工作流：

1. Design workflow with CrewAI Flows
2. Use LangGraph patterns for state
3. Combine crews in flow steps
4. Handle branching and routing

相关技能

配合以下技能效果良好：langgraph、autonomous-agents、langfuse、structured-output

适用场景

• 用户提及或暗示：crewai
• 用户提及或暗示：多 Agent 团队
• 用户提及或暗示：Agent 角色
• 用户提及或暗示：Agent 团队
• 用户提及或暗示：基于角色的 Agent
• 用户提及或暗示：协作型 Agent

限制

• 仅当任务明确符合上述范围时，才使用此技能。
• 不要将输出视为特定环境验证、测试或专家评审的替代品。
• 如果缺少必需的输入、权限、安全边界或成功标准，请停止并请求澄清。