getting-started

CrewAI 入门与架构

了解如何选择合适的抽象、搭建项目脚手架，并将各部分连接起来。

已针对 crewai 1.15.23 进行验证，日期为 2026-10-01。 已于 2026-10-01 使用真实大模型进行实测。 先运行 crewai version；如果主版本或次版本与 1.15 不同，请使用 ask-docs 技能重新核对与版本相关的细节。

---

必须遵循的工作流——请先阅读

绝不要手动创建 crewAI 项目文件。 始终使用 CLI 搭建项目脚手架：

crewai create flow <project_name>

这不是可选项。 即使你只需要一个 crew，即使你对文件结构了如指掌，也要先运行 CLI，再修改生成的文件。切勿从头手动编写 main.py、crew.py、agents.yaml、tasks.yaml 或 pyproject.toml。

原因： CLI 会设置正确的导入语句、目录结构、pyproject.toml 配置和样板代码；手动操作时，这些内容很容易出现细微错误。以下参考资料将帮助你了解各部分的工作方式，以便你修改脚手架生成的代码，而不是跳过搭建脚手架的步骤。

工作流：

1. 运行 crewai create flow <name>（优先使用下划线；CLI 会将连字符和空格转换为下划线，用作文件夹名和包名）
2. 编辑生成的 YAML 和 Python 文件，以满足你的使用场景
3. 运行 crewai install，然后运行 crewai run

---

1. 选择合适的抽象

crewAI 有五种常见的抽象选项。选择最符合需求的简单方案：

层级	适用场景	开销	示例
LLM.call()	单次提示词调用、无需工具、结构化提取	最低	将电子邮件解析为字段
Agent.kickoff()	单个 Agent 使用工具并进行推理，无需多 Agent 协作	低	使用网页搜索研究某个主题
Crew.kickoff()	多个 Agent 协作完成相关任务	中	研究 + 撰写 + 审核工作流
Flow 包装 crews/agents/LLM 调用	具备状态、路由、条件判断和错误处理的生产应用	完整	带有分支逻辑的多步骤工作流
对话式 Flow	多轮对话；每条用户消息都会使用相同的会话 ID 重新运行 Flow	完整 + 实验性	具备路由聊天、研究和升级处理轮次的支持助手

决策流程图

Do you need tools or multi-step reasoning?
├── No  → LLM.call()
└── Yes
    └── Do you need multiple agents collaborating?
        ├── No  → Agent.kickoff()
        └── Yes
            └── Do you need state management, routing, or multiple crews?
                ├── No  → Crew (but still scaffold as a Flow for future-proofing)
                └── Yes → Flow + Crew(s)

Do users send multiple chat messages in one session?
└── Yes → Conversational Flow with handle_turn(message, session_id=...)

经验法则： 对于任何生产应用，始终从 Flow 开始。你可以在 Flow 步骤中嵌入 LLM.call()、Agent.kickoff() 或 Crew.kickoff()。这样你就能获得状态管理和错误处理能力，也能为后续扩展留出空间。

对于聊天应用，请从对话式 Flow 开始，不要试图让 Crew.kickoff() 或 Flow.kickoff() 充当聊天循环：每条用户消息都调用 flow.handle_turn(message, session_id=...)，或者使用 flow.chat() 启动本地终端 REPL。官方指南：https://docs.crewai.com/en/guides/flows/conversational-flows。

---

2. LLM.call()——直接调用 LLM

适用于简单的单轮任务，不需要工具或 Agent 推理。

from crewai import LLM
from pydantic import BaseModel

class EmailFields(BaseModel):
    sender: str
    subject: str
    urgency: str

llm = LLM(model="openai/gpt-4o")   # non-OpenAI providers need their extra, e.g. uv add "crewai[anthropic]"

# Plain call - returns a string
raw = llm.call(messages=[{"role": "user", "content": "Summarize this text..."}])
print(raw)  # str

# With response_model - returns the Pydantic object directly
result = llm.call(
    messages=[{"role": "user", "content": f"Extract fields from this email: {email_text}"}],
    response_model=EmailFields,
)
print(result.sender)   # str — access Pydantic fields directly
print(result.urgency)  # str

llm.call(..., response_format=Model) 会引发 TypeError: ... unexpected keyword argument 'response_format'. Did you mean 'response_model'?。在 LLM.call() 中，关键字是 response_model；在 Agent.kickoff() 中则是 response_format（见下一节）。

不适用场景： 如果你需要工具、多步骤推理或重试，请改用 Agent。

---

3. Agent.kickoff()——运行单个 Agent

适用于需要单个 Agent 使用工具并进行推理，但不需要多 Agent 协作的场景。

from crewai import Agent
from crewai_tools import SerperDevTool
from pydantic import BaseModel

class ResearchFindings(BaseModel):
    main_points: list[str]
    key_technologies: list[str]

researcher = Agent(
    role="AI Researcher",
    goal="Research the latest AI developments",
    backstory="Expert AI researcher with deep technical knowledge.",
    llm="openai/gpt-4o",       # Optional: defaults to the MODEL / MODEL_NAME / OPENAI_MODEL_NAME env var, else "gpt-4.1-mini"
    tools=[SerperDevTool()],   # needs SERPER_API_KEY
)

# Unstructured output
result = researcher.kickoff("What are the latest LLM developments?")
print(result.raw)            # str
print(result.usage_metrics)  # token usage dict

# Structured output with response_format
result = researcher.kickoff(
    "Summarize latest AI developments",
    response_format=ResearchFindings,
)
print(result.pydantic.main_points)

注意： Agent.kickoff() 会将结果封装在 LiteAgentOutput 中——通过 result.pydantic 访问结构化输出。这与直接返回 Pydantic 对象的 LLM.call() 不同。

不适用场景： 如果你需要多个 Agent 彼此传递上下文，请使用 Crew。

4. CLI 脚手架参考

如上所述：绝不要跳过 crewai create flow。 本节介绍 CLI 生成的内容，帮助你了解要修改什么，而不是让你手动重新创建。

crewai create flow my_project

命名：crewai create flow my-project 仍然有效——CLI 会创建名为 my_project 的文件夹和包——但请使用下划线，这样你输入的名称就与导入的包名一致。

此命令会生成以下内容（crewai 1.15.23）：

my_project/
├── src/my_project/
│   ├── crews/
│   │   └── content_crew/
│   │       ├── config/
│   │       │   ├── agents.yaml    # Agent definitions (role, goal, backstory)
│   │       │   └── tasks.yaml     # Task definitions (description, expected_output)
│   │       └── content_crew.py    # Crew class with @CrewBase
│   ├── tools/
│   │   └── custom_tool.py
│   └── main.py                    # Flow class with @start/@listen, plus kickoff() and plot()
├── .env                           # OPENAI_API_KEY placeholder - replace it, never commit it
├── AGENTS.md, CLAUDE.md, GEMINI.md, CURSOR.md   # coding-assistant instructions
├── README.md
└── pyproject.toml                 # [tool.crewai] type = "flow"; scripts kickoff, plot, ...

生成的脚手架仅依赖 crewai[tools]——crewai create flow 和 crewai create crew --classic 都不会添加提供商扩展，因此使用任何非 OpenAI 模型时，都要先运行 uv add "crewai[anthropic]"（或相应的扩展），再运行 crewai install。

生成的 main.py 使用绝对包路径导入——from my_project.crews.content_crew.content_crew import ContentCrew 和 from crewai.flow import Flow, listen, start。请保持这种风格；相对导入（from .crews...）会导致 uv run src/my_project/main.py 出错，也会导致 crewai deploy validate 失败（参见 deploy-to-amp 技能）。请确保 main.py 中的 kickoff() 返回 None：控制台脚本会执行 sys.exit(kickoff())，因此返回结果会导致进程以状态码 1 退出。有关 Flow 脚手架的更多信息，请参阅 build-flow 技能。

优先使用 crewai create flow。 仅当你确定永远不需要路由、状态或多个 crew 时，才使用 crewai create crew <name> --classic。如果不加 --classic，crewai create crew 会启动交互式向导，并生成 JSON 项目（crew.jsonc、agents/*.jsonc），而不是下面的 @CrewBase + YAML 布局；如果没有终端，则会打印 Aborted!。在脚本中添加 --skip-provider 可跳过提供商提示。

---

5. Crews：YAML 配置与 @CrewBase

该脚手架使用 YAML 定义 Agent 和任务，并通过 @CrewBase 类将它们连接起来。完整的双 Agent 示例请参阅：Crew YAML 配置。

# config/agents.yaml
researcher:
  role: >
    {topic} Senior Data Researcher
  goal: >
    Uncover cutting-edge developments in {topic}
  backstory: >
    You're a seasoned researcher with a knack for uncovering
    the latest developments in {topic}.
  # Optional overrides: llm: openai/gpt-4o, max_iter: 20, max_rpm: 10

# config/tasks.yaml
research_task:
  description: >
    Conduct thorough research about {topic}.
  expected_output: >
    A detailed report on the top 5 developments in {topic}.
  agent: researcher
  output_file: output/report.md

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

@CrewBase
class ResearchCrew:
    """Research crew."""

    agents_config = "config/agents.yaml"   # these two paths are also the defaults
    tasks_config = "config/tasks.yaml"

    @agent
    def researcher(self) -> Agent:          # method name == YAML key
        return Agent(config=self.agents_config["researcher"])

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

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

关键规则：

• {variable} 占位符会在运行时通过 crew.kickoff(inputs={...}) 替换
• expected_output 始终是一个字符串（绝不是 Pydantic 类名）；在 Python 的 @task 方法中设置 output_pydantic=Model
• agent 的值必须与 agents.yaml 中的 Agent 键匹配，方法名也必须与 YAML 键匹配（def researcher -> researcher:）
• 在 Process.sequential 中，每个任务都会自动接收所有先前任务的输出作为上下文
• 对于非顺序依赖，请使用 context=[other_task] 显式传递输出

---

6. Flows——生产应用的基础

Flows 是构建生产级 crewAI 应用的推荐方式。它们提供状态管理、条件路由、人工参与环节和持久化，并将 crews、agents 和 LLM 调用封装成连贯的工作流。本节介绍各抽象如何协同工作；完整的 Flow API（状态规则、or_/and_、@persist、检查点、@human_feedback、plot()）请使用 build-flow 技能。

基础 Flow — main.py

from crewai.flow import Flow, listen, start
from pydantic import BaseModel
from my_project.crews.research_crew.research_crew import ResearchCrew

class ResearchState(BaseModel):
    topic: str = ""
    report: str = ""

class ResearchFlow(Flow[ResearchState]):

    @start()
    def begin(self):
        print(f"Starting research on: {self.state.topic}")

    @listen(begin)
    def run_research(self):
        result = ResearchCrew().crew().kickoff(
            inputs={"topic": self.state.topic}
        )
        self.state.report = result.raw

def kickoff():
    flow = ResearchFlow()
    flow.kickoff(inputs={"topic": "AI Agents"})   # do not return the result

if __name__ == "__main__":
    kickoff()

要点：

• flow.kickoff(inputs={"topic": "AI Agents"}) 会填充 self.state.topic（键必须与 Pydantic 字段名匹配；未知键会被静默丢弃）。YAML 中的 {variable} 替换会在之后进行，即在 Flow 步骤中调用 crew.kickoff(inputs={"topic": self.state.topic}) 时。调用链为：Flow 输入 → 状态 → crew 输入 → YAML 替换。
• 每个 @listen 方法都会在其依赖项完成后运行
• 状态会在所有 Flow 步骤间持久保存——可用它在不同 crews 间传递数据

状态： 使用结构化状态（Flow[MyState]，搭配 Pydantic 模型，并为每个字段设置默认值）可实现类型安全和验证；class MyFlow(Flow) 则使用字典状态（self.state["topic"]），适用于临时原型。请勿自行声明 id——Flow 会添加该字段。

在 Flows 中使用 Agent.kickoff()（常见模式）

许多生产级 Flow 完全不使用 Crew，而是通过 Agent.kickoff() 编排单个 Agent。这样可以进行精细控制——每个 Flow 步骤都会创建一个专用 Agent（职责范围明确，工具特定），向其传递状态，并将 result.raw（或使用 response_format=Model 时的 result.pydantic）存回状态。Flow 负责编排；Agent 负责推理。下一个示例中的 handle_simple 步骤就采用了这种模式。

在 Flow 中，何时使用 Agent.kickoff()，何时使用 Crew.kickoff()：

使用 Agent.kickoff() 的场景	使用 Crew.kickoff() 的场景
每个步骤都是使用不同工具的独立 Agent	多个 Agent 需要协作完成一项任务
希望由 Flow 控制执行顺序	Agent 需要在一个步骤内相互传递上下文
步骤彼此独立，无需 Agent 间委派	需要带有管理者的层级式流程
希望最大限度地控制步骤之间传递的数据	子工作流自成一体且可复用

在 Flow 中混合使用抽象

Flow 可以在单个工作流中组合所有 crewAI 抽象：

from crewai import LLM, Agent
from crewai.flow import Flow, listen, router, start

class ProductFlow(Flow[ProductState]):

    @start()
    def classify_request(self):
        # LLM.call() for simple classification
        llm = LLM(model="openai/gpt-4o")
        self.state.category = llm.call(
            messages=[{"role": "user", "content": f"Classify as 'simple' or 'complex': {self.state.request}"}],
            response_model=Category,
        ).category

    @router(classify_request)
    def route_by_category(self):
        if self.state.category == "simple":
            return "quick_answer"
        return "deep_research"

    @listen("quick_answer")
    def handle_simple(self):
        # Agent.kickoff() for single-agent work
        agent = Agent(role="Helper", goal="Answer quickly", backstory="...")
        self.state.answer = agent.kickoff(self.state.request).raw

    @listen("deep_research")
    def handle_complex(self):
        # Crew.kickoff() for multi-agent collaboration
        result = ResearchCrew().crew().kickoff(
            inputs={"topic": self.state.request}
        )
        self.state.answer = result.raw

路由、汇合、持久化、人工反馈——简明说明

• @router(method) 返回一个字符串标签；@listen("label") 会运行对应分支。标签不能与监听该标签的方法名称相同，并且每个标签都必须有监听器（没有匹配项的标签会导致 Flow 静默结束）。
• or_(a, b) 会在第一个条件完成时触发一次。and_(a, b) 会在所有条件完成后触发一次。两者都从 crewai.flow 导入。
• @persist()（带括号；from crewai.flow import persist）会将状态保存到本地 SQLite 文件；使用 kickoff(restore_from_state_id=...) 恢复。
• @human_feedback(message=..., emit=[...], llm=..., default_outcome=...) 会暂停以供审核，并根据审核结果进行路由。审核者的文本位于 HumanFeedbackResult 的 result.feedback（或 self.last_human_feedback.feedback）中。
• flow.plot("my_flow.html", show=False) 会将交互式 HTML 写入临时目录并返回其路径——不会生成 PNG。

Flow Routing 和 build-flow 技能中提供了每项功能的测试模式。

使用 handle_turn() 处理对话流

如果产品是聊天会话，请使用对话式 Flow：例如支持助手、路由式研究助手、引导向导，或任何同一用户会发送多轮消息的界面。

核心模型：

• 每条用户消息都会启动一次新的 Flow 运行，但使用相同的会话 ID
• handle_turn(message, session_id=...) 会将用户消息追加到 state.messages，重置每轮执行跟踪，然后在内部调用 kickoff(inputs={"id": session_id})
• Flow.kickoff() 不接受 user_message= 或 session_id= 关键字参数（会触发 TypeError）
• 使用 route_turn() 和 @listen("ROUTE") 处理器来路由聊天轮次；也可以省略 route_turn()，让 LLM 路由器从处理器中进行选择
• 处理器返回的字符串会成为助手回复；如果想明确记录回复，请调用 append_assistant_message(reply)（不会重复追加）
• 将自有循环包装在 try/finally 中，并调用 finalize_session_traces()；flow.chat() 会为本地 REPL 执行此操作

从 crewai.flow 导入；crewai.experimental.conversational 是同一模块的弃用别名。@ConversationConfig 装饰器已设置 conversational = True，且 defer_trace_finalization 默认为 True。如果没有 llm，converse 路由会回复 "配置 LLM 后，我就可以继续对话。"

对于由 LLM 驱动的路由，无需额外配置：使用 @ConversationConfig(llm=...) 且不设置 route_turn() 时，路由器会自动启用，并根据 @listen("ROUTE") 处理器及其文档字符串构建路由目录——不要在路由器提示词中重复列出路由。只有在需要更改提示词、描述或回退选项时，才传入 RouterConfig。

对话流中提供了一个经过测试的最小示例（使用 route_turn() 搭配自定义处理器和 converse，并通过 try/finally 中的 handle_turn() 驱动），以及完整的生命周期、路由、流式传输、持久化和跟踪指南。

---

7. 使用 inputs 进行变量插值

{variable} 模式可让你重复使用 crew。

# Variables flow through: kickoff → YAML templates → agent/task prompts
crew.kickoff(inputs={
    "topic": "AI Agents",
    "current_year": "2025",
    "target_audience": "developers",
})

在 YAML 中，{topic} 和 {current_year} 会被替换（包括 Agent 的 role／goal／backstory，以及任务的 description／expected_output）：

research_task:
  description: >
    Research {topic} trends for {current_year},
    targeting {target_audience}.

常见错误：

• 调用 kickoff() 时未传入 inputs → 字面量 {variable} 会进入提示词
• 传入的 inputs 缺少 YAML 中引用的变量 → ValueError: Missing required template variable 'Template variable 'current_year' not found in inputs dictionary'
• 使用 Jinja2 语法 {{ }}，而不是单大括号 { } → crewAI 使用单大括号；{{ topic }} 会保留为字面量
• 传入的变量与任何 YAML 占位符都不匹配 → 会被静默忽略

---

8. 运行项目

# Install dependencies
crewai install

# Run the flow (or classic crew)
crewai run

或者直接从项目目录运行：

cd my_project
uv run kickoff                  # flow: the console script in pyproject.toml (run_crew for a classic crew)
uv run src/my_project/main.py   # also works with the scaffold's absolute imports

模型和密钥来自环境变量或 .env：设置 MODEL=anthropic/claude-haiku-4-5（或任何提供商字符串），并配置该提供商的密钥，然后使用 uv add "crewai[anthropic]" 添加提供商扩展。若未设置 MODEL 且未设置 llm=，Agent 将使用 gpt-4.1-mini，并需要 OPENAI_API_KEY。

9. 快速诊断清单

症状	可能原因	修复方法
{topic} 原样出现在 Agent 输出中	kickoff() 中没有 inputs=	传入 crew.kickoff(inputs={"topic": "..."})
ValueError: Missing required template variable	已提供 inputs，但缺少一个 YAML 占位符	传入 YAML 中用到的每个 {variable}
self.agents_config['name'] 上出现 KeyError	方法名与 YAML 键不匹配	确保 @agent def researcher 与 YAML 中的 researcher: 匹配
ModuleNotFoundError / ImportError: attempted relative import	使用了相对导入，或包路径与文件夹不匹配	使用脚手架的绝对导入形式：from my_project.crews.crew_name.crew_name import CrewClass
ImportError: cannot import name 'persist'	使用了 from crewai.flow.flow import persist	改用 from crewai.flow import persist
llm.call 出现 TypeError: ... unexpected keyword argument 'response_format'	LLM.call() 的关键字参数错误	使用 llm.call(messages, response_model=Model)
Crew 运行了，但 Flow 状态为空	未将结果写回 self.state	在 @listen 方法中将 Crew 输出赋给 self.state.field
Process.SEQUENTIAL 引发 AttributeError	枚举值使用了大写形式	使用小写形式：Process.sequential
Agent 忽略工具	工具分配给了 Agent，但任务需要这些工具	将工具移到任务级别，或确认 Agent 拥有正确的工具
Agent 编造搜索结果	未分配工具 — Agent 实际上无法搜索	添加 tools=[SerperDevTool()] 或等效工具；没有工具的 Agent 会产生幻觉数据
@listen 从未触发	监听器字符串与路由器的返回值不匹配，或 @listen("method_name") 中有拼写错误	@router 必须返回 @listen("label") 期望的确切字符串；方法链式调用时优先使用 @listen(method_ref)
两个上游步骤都已完成，但监听器只运行了一次	这是 or_ 的行为 — 第一个步骤完成时就会触发一次	如果需要等待所有上游步骤完成，请使用 and_()
crewai run 提示 Flow 出错，但输出看起来正确	main.py 中的 kickoff() 返回了一个值	返回 None
crewai create crew 打印 Aborted! 或显示向导问题	这是一个未使用 --classic 的 JSON 向导	使用 crewai create crew <name> --classic（或使用 crewai create flow）
出现 AuthenticationError、API key not found 或 OPENAI_API_KEY is required	缺少模型提供方的环境变量	在 .env 中设置提供方密钥（OPENAI_API_KEY、ANTHROPIC_API_KEY 等），并为搜索工具设置 SERPER_API_KEY
Anthropic native provider not available	未安装提供方扩展包	使用 uv add "crewai[anthropic]"（或对应的扩展包）
Agent 在结构化输出上无限重试	对大模型而言，Pydantic 模型过于复杂	简化模型、减少嵌套，或使用能力更强的 llm
Agent 循环到 max_iter 后仍未完成	任务描述过于含糊，或与 expected_output 冲突	将 expected_output 写得具体且可实现；降低 max_iter 以便更快失败
@router 返回值被忽略	方法未使用 @router 装饰	分支方法使用 @router(condition)，而不是 @listen(condition)
出现 AttributeError: ... 'feedback_text'	使用了旧版 HumanFeedbackResult 字段名	使用 result.feedback
Flow.kickoff(user_message=..., session_id=...) 失败	kickoff() 不接受对话相关的关键字参数	对聊天消息使用 flow.handle_turn(message, session_id=...)
聊天记录缺少助手回复	处理程序既未返回字符串，也未记录回复	返回回复字符串，或调用 self.append_assistant_message(reply)

---

参考资料

如需深入了解特定主题，请参阅：

• Crew YAML 配置 - 完整的 agents.yaml、tasks.yaml 和 @CrewBase crew.py 示例
• Flow 路由、持久化、流式传输与人工反馈 - @router、or_()、and_()、@persist、流式传输和 @human_feedback 模式
• 对话式 Flows - 多轮 Flow API，包含 handle_turn()、stream_turn()、chat()、ConversationConfig、路由器行为、持久化和追踪
• MCP 服务 — 优先使用官方 MCP 服务，而非原生工具；包含设置、DSL 集成和已知的官方服务
• 工具目录 - 常见的内置工具，包含导入项、环境变量和常见组合（不存在 MCP 服务时用作备用方案）

相关技能：

• build-flow - 完整的 Flow API：状态规则、路由器、or_/and_、@persist、检查点、@human_feedback、plot()
• check-crewai-api - 当前的 Agent/Task/Crew/LLM API 与记忆中的 0.x 形式对比
• connect-tools-and-mcp - 自定义工具、crewai_tools 中有哪些名称、mcps=[...] 形式和超时设置
• design-agent — Agent 的 Role-Goal-Backstory 框架、参数调整、工具分配、记忆与知识配置
• design-task — 任务 description/expected_output 最佳实践、护栏、结构化输出、依赖关系
• test-crewai-project - 使用 stub LLM 离线运行 crew 和 flow
• deploy-to-amp - CrewAI AMP 部署所需的项目结构和入口点
• call-deployed-crew - 通过 HTTP 调用已部署的 crew 或 flow
• ask-docs - 查询实时 CrewAI 文档，解答这些技能未涵盖的问题