CrewAI 把 agent 写成团队流程，而不是消息总线

一、先搭一个最小的 AI 调研团队

假设你要比较三款竞品。一个模型当然可以直接写答案，但它既要搜索资料，又要判断可信度，还要组织长文， 很容易顾此失彼。更自然的办法是把工作拆开：研究员找证据，写作者根据证据成稿。CrewAI 用三种对象表达这支小团队：

• Agent 是“员工”：它有角色、目标、背景和工具，例如会搜索网页的竞品研究员。
• Task 是“任务卡”：它写清要做什么、合格结果长什么样，以及上游要提供哪些材料。
• Crew 是“团队”：它收拢员工和任务卡，并决定按顺序交接，还是由负责人动态分派。

到这里还没有复杂框架术语。你只是把“一个大 prompt”改写成“两个员工各拿一张任务卡，再由团队组织执行”。 这就是理解 CrewAI 最重要的第一步。

二、团队会协作，为什么还需要 Flow

Crew 能完成报告，却不一定能把报告稳定地送进生产。比如资料不足时要回到研究员，敏感结论必须人工确认， 审核等待两小时后还要从原处继续。这些问题不再是“谁擅长什么”，而是“下一步去哪、何时暂停、状态存在哪里”。

Flow 就是显式的流程路线图：它规定入口、步骤之间的连接、条件分支和恢复点。 Crew 负责让一群 Agent 自主完成工作，Flow 负责把这群人的工作嵌进可预测的业务流程。两者不是竞争关系， 而是“团队协作”和“生产控制”两层责任。

三、一份竞品报告怎样走完 Crews 与 Flows

1. 用户提交竞品名称，Flow 把主题和任务编号写进当前状态。
2. Flow 的起点启动调研 Crew，研究员 Agent 收到“寻找可靠资料”的 Task。
3. 研究员交付带来源的调研结果，这份结果成为写作 Task 的上游材料。
4. 写作者 Agent 生成结构化报告，Crew 汇总每张任务卡的输出并结束本轮协作。
5. 调用 Crew 的 Flow method 显式读取 CrewOutput，提取报告并写入 Flow state；这一步不会自动发生。
6. Flow 检查自己 state 里的报告：材料不足就回到研究步骤，满足要求就进入审核。
7. 若触发人工确认，Flow 保存状态并暂停；审核人返回意见后，从下一步继续。
8. 最终发布完成，Flow 记录结果；Crew 不需要知道报告来自 Web、定时任务还是其他入口。

官方 README 也按这个思路拆成两块：CrewAI Crews 负责有自主性的 agent team， CrewAI Flows 负责更可控的生产流程。为了让这篇不变成源码字段清单，我会用一个小例子贯穿： 你要让 AI 做一份竞品调研报告。研究员负责找资料，写作者负责整理成文，负责人决定是否需要人工确认。 Crews 解释“这些人怎样合作”，Flows 解释“这个流程怎样上线、分支和恢复”。 证据在 CrewAI README overview 和 Understanding Flows and Crews。

四、再把 Crews 与 Flows 的边界落到源码

做一份竞品调研报告时，有两个问题要分开。第一个问题是“谁来做”：研究员找资料，写作者整理，负责人验收。 第二个问题是“流程怎样受控”：先查资料还是先列提纲，资料不足时要不要回到研究员，最终发布前要不要人工确认。 CrewAI 把第一个问题交给 Crews，把第二个问题交给 Flows。

把这份报告落成源码形状，Crews 和 Flows 的分工会更清楚：

Crews side
  researcher = Agent(role, goal, backstory, tools)
  writer = Agent(role, goal, backstory)
  research_task = Task(description, expected_output, agent=researcher)
  write_task = Task(description, expected_output, agent=writer, context=[research_task])
  Crew(agents=[researcher, writer], tasks=[research_task, write_task], process=sequential)
    -> kickoff(inputs)
    -> TaskOutput(raw/json/pydantic, output_file?)
    -> CrewOutput(tasks_output, token_usage)

Flows side
  state = {topic, research_result, draft, review_status}
  @start collect_research
  @listen collect_research -> write_report
  @router review_report -> "publish" or "human_review"
  persistence stores state id and resumes from the next event

也就是说，Crew 先把“谁做什么任务”变成可执行协作；Flow 再把“什么时候进入哪一步、状态怎样保存、 失败或人工反馈后从哪里继续”变成显式控制。下面读字段和装饰器时，都可以回到这条形状上。

4.1 Crew 与 Flow 的接缝必须由应用写出来

Task.context=[research_task] 只在同一个 Crew 内把上游 TaskOutput 交给下游 Task； 它不会把结果写进 Flow state。反过来，Flow 的 state 也不会自动吸收 CrewOutput。 报告任务需要一个明确的桥：

@start()
def collect_research(self):
    crew_output = research_crew.kickoff(inputs={"topic": self.state.topic})
    report = crew_output.pydantic or crew_output.json_dict or crew_output.raw
    self.state.research_result = report
    return "research_ready"

这是形状级示例，重点是 owner 交接：Crew 产出 CrewOutput，Flow method 负责选择可消费形状并更新 research_result，router 之后只读 Flow state。由此，同一份报告包才能连续变化： topic → ResearchPacket → DraftReport → ReviewDecision → PublicationRecord。

官方 docs 说 crew 代表一组协作 agent，定义 task execution、agent collaboration 和 overall workflow 的策略。 这句话落到代码里，就是 agents、tasks、process、memory、 planning、knowledge_sources、step_callback 这些属性： 它们不是在描述传输协议，而是在描述一个团队怎样接活、交接、补资料和被观察。 证据在 Crew docs attributes。

Flows 的说明则更像 workflow engine：它连接多个 task，管理 state，控制 execution flow，并支持 conditional logic、 loops 和 branching。换成刚才的报告例子，Flow 关心的是“资料收集完成”这个事件会触发哪一步， “报告质量不够”会路由到返工还是人工确认。它不是 Crews 的替代品，而是给生产路径加一层显式控制。 证据在 Flow docs overview。

五、Agent：先定义一个人，再让它调用模型

先看最小场景：你要建一个“竞品研究员”。如果 CrewAI 只是 completion adapter，代码只需要保存一个 prompt， 每次把用户问题丢给模型就结束了。但 CrewAI 的 Agent 要在一个团队里长期工作， 所以它至少要回答四件事：我是谁，我会什么，我受到哪些限制，我当前处在哪个工作现场。

所以 BaseAgent 和 Agent 的字段多，并不是为了“配置看起来很强”，而是把一个 agent turn 拆成身份、 能力、约束、现场和模型调用五层。只要按这五层读，function_calling_llm、respect_context_window、 planning_config、embedder、a2a、executor_class 这些名字就不再是一串散乱清单。 对应源码在 BaseAgent fields 和 Agent fields。

执行任务时，Agent 不是直接打一轮模型

当这个研究员真的收到一张任务卡时，Agent.execute_task 做的也不是“把 description 发给 LLM”这么简单。 它的执行路径可以读成四步。

1. 把 Task 变成 prompt：拼上日期、输出格式、上下文和上游任务结果。
2. 补充资料：从 memory 里召回相关历史，再查 agent 和 crew 级别的 knowledge。
3. 准备可调用能力：整理工具名、工具描述、training data 和人工输入标记。
4. 交给 executor：把最终 input 和工具表面交给 agent_executor.invoke。

这条路径解释了 CrewAI 的取舍：它不是把 agent turn 看成一次裸模型调用，而是看成“任务卡 + 记忆 + 知识 + 工具 + 输出约束”的组合。 源码在 task prompt preparation、 memory retrieval、 execute_task 和 executor invocation。

Agent.message() 为什么也会创建临时 Crew

CrewAI 甚至把单轮消息也拉回这个模型里：Agent.message() 不会绕过 task 和 crew 直接打一轮模型， 而是临时创建一个 Task 和单 agent Crew，再调用 crew.kickoff() 返回 raw output。 源码在 Agent.message。 这个细节很能说明框架性格：哪怕只是聊一句，也尽量落回“任务”和“团队”的语义，而不是开一个旁路。

六、Task：不是一句 prompt，而是一张任务卡

如果 Agent 是人，Task 就是任务卡。它不只写“请分析竞品”，还要写清楚： 期望输出是什么、谁来做、可以用哪些工具、上游上下文是什么、结果要不要按 JSON 或 Pydantic 模型收束、 失败后能不能重试，以及是否需要人工确认。

这就是为什么 Task 的字段表看起来很长：它同时管理工作说明、上下文、输出格式和验收逻辑。 源码在 Task fields。

执行时也按这张任务卡来。execute_sync 和 execute_async 最终都会进 _execute_core： 先确认执行 agent，写入 prompt context，发出 task started 事件，再让 agent 执行。拿到结果后，它会按 BaseModel、 output_pydantic、output_json 和 guardrail 生成 TaskOutput，执行 callback， 必要时写入 output file，最后发出 task completed 事件。换句话说，Task 是“下发、执行、验收、落盘”的边界。 源码在 sync and threaded async entry 和 _execute_core。

Task.prompt() 则是这张任务卡的文本版：description 加 expected output；如果开启 markdown， 就追加 Markdown 输出要求；如果有 trigger payload 或 input files，也会把相关上下文加进 description。 所以 Task 不是聊天历史数组的一行，而是可以被执行器、guardrail 和后续任务共同理解的工作契约。 源码在 Task.prompt。

七、Crew：执行 owner，不是任务列表容器

把多个 agent 和 task 放在一起，还不等于有了流程。Crew 才是按下启动键的人： Crew.kickoff 会处理 checkpoint、streaming、inputs 和 event runtime scope， 然后根据 process 选择顺序执行或层级执行，结束后再跑 after kickoff callbacks、post kickoff 和 usage metrics。 目前 Process enum 真正可用的主路径是 sequential 和 hierarchical； consensual 在这个快照里还不是可执行路径。 源码在 Process 和 Crew.kickoff。

顺序模式直接执行 task list；层级模式会先创建 manager agent，再执行同一批 task。自定义 manager agent 会被设置为 allow_delegation；如果没有自定义 manager，就用 manager_llm 创建一个带 delegation tools 的默认 manager。 这不是“多一个 agent”的小区别，而是把调度权交给了 manager。 源码在 sequential and hierarchical process。

_execute_tasks 是 Crew 的调度核心。你可以把它读成一个项目经理在处理任务卡： 先逐个准备执行数据，遇到 conditional task 先判断能不能执行；遇到 async task 就挂起一个 future； 遇到同步 task 之前先收束已挂起的 futures；执行同步 task 时再根据 task context 聚合上游输出。 最终 _create_crew_output 从有效 task outputs 中取最后一个 raw 作为 crew raw output， 同时保留每张任务卡的输出和 token usage。 源码在 _execute_tasks、 context and final output。

Crew 的 planning 也要按这个思路理解：它不是另起一个外部 DAG 系统，而是在 kickoff 前让 CrewPlanner 为每个 task 生成 plan，再把对应 plan 追加到 task description。也就是说，planning 的结果会回到任务卡本身， 让后面的 agent 带着更具体的执行计划工作。 源码在 _handle_crew_planning 和 CrewPlanner。

八、Memory、knowledge、tools：分别补三种缺口

初学者很容易把 memory、knowledge 和 tools 混在一起。CrewAI 的实现其实很朴素：memory 解决“我以前做过什么”， knowledge 解决“我能查哪些资料”，tools 解决“我能对外做什么”。它们最后都会影响 prompt 或可调用能力， 但职责不同。

这样读源码就清楚了：memory recall 以 task description 为 query，取相关 memories 后追加到 prompt； knowledge retrieval 同时查询 agent knowledge 和 crew knowledge；Crew 侧的 query_knowledge 只是把 query 转给 crew-level knowledge base。 源码在 agent memory retrieval、 agent and crew knowledge lookup 和 Crew.query_knowledge。

tools 的接入也跟 process 有关。Crew 在 _prepare_tools 里根据 allow_delegation、 hierarchical manager、code execution、multimodal、apps、MCP、memory 和 input files 动态补工具。 hierarchical 模式下，manager 通过 delegation tools 指挥任务 agent；非 hierarchical 但 agent 允许 delegation 时， Crew 也会把其他 agents 作为 delegation tools 注入。也就是说，“能不能委派”不是一句 prompt 里的建议，而是运行时真的改变工具表面。 源码在 _prepare_tools 和 delegation and platform tools。

九、Flow：把生产路线画出来

如果 Crews 像一个会协作的小团队，Flow 就像上线后的流程图。它回答的是生产问题： 哪一步先开始，哪个结果触发下一步，哪条路需要人工确认，失败恢复时从哪个 state 继续。 所以 Flow 不是 Crew 的别名，而是把“控制路线”显式写出来。

现版本 crewai.flow.flow 已经只是兼容 re-export surface，真正实现拆到了 crewai.flow.dsl、 crewai.flow.flow_definition 和 crewai.flow.runtime。这也能看出分层：DSL 负责把 Python 方法标记成节点和边， runtime 负责执行、状态、事件、持久化和恢复。 源码在 flow.py re-export。

三个核心装饰器：入口、监听、路由

Flow 的 DSL 装饰器可以先按一句话理解：@start 表示“从这里开始”，@listen 表示“等某个上游结果出现后再做”，@router 表示“根据返回值决定走哪条路”。如果报告质量分数太低， router 就可以把流程发回研究步骤；如果质量合格，就进入人工确认或发布。 源码在 @start、 @listen 和 @router。

runtime 侧，Flow 是 Pydantic model，支持 initial_state、name、 tracing、stream、memory 和 input_provider 这些运行字段。 flow_definition() 会从 class lazily build static FlowDefinition。 state 可以是 dict 或 BaseModel；_initialize_state 会更新输入并保证 dict state 有 id。 对应用开发者来说，这就是“每次执行都有一份可追踪的流程状态”。 源码在 Flow class and definition 和 state initialization。

kickoff 是同步包装，最终进入 kickoff_async。异步路径会处理 checkpoint、streaming、 input files、restore_from_state_id、persistence hydrate、FlowStartedEvent、start methods 和 resumption。 如果有 unconditional starts，就只跑它们；否则所有 starts 都可以作为 entry points，并且可以并行执行。 后续 _execute_start_method 会执行 start method；如果它也是 router，还会把返回值作为额外 trigger 继续触发 listeners。 源码在 Flow.kickoff、 kickoff_async setup 和 start method execution。

listener 的执行规则可以这样记：routers 先顺序执行，router result 成为新的 trigger；普通 listeners 再并行执行。 runtime 还会跟踪 OR listener 的 firing 状态，避免 multi-event OR 条件重复触发。 方法执行本身会发 started / finished / failed / paused 事件，支持 sync method 在线程池里执行， 并在方法完成后调用 persistence。这就是 Flow 比 Crew 更适合生产控制的地方：它知道每个节点何时开始、何时结束、失败和暂停怎样被记录。 源码在 _execute_method、 method persistence 和 router and listener dispatch。

@persist 和 @human_feedback 也按这个模型工作：它们先把配置 stamp 到 class 或 method， 再由 Flow engine 在方法完成或 human feedback step 时读取 definition 并执行。也就是说，Flow 的生产特性没有绕开 graph， 而是挂在 graph method 生命周期上。 源码在 @persist 和 @human_feedback。

9.1 持久化保存流程记录，不承诺副作用 exactly-once

在人工审核处暂停时，应用至少要能还原下面这组事实；只保存一个“paused”字符串不足以恢复：

restore_from_state_id、hydrate、@persist 和 @human_feedback 让 Flow 有恢复入口，但它们不回滚已发送邮件、已写数据库或已发布文章。发布应由应用持有稳定幂等键并保存外部回执。 Flow method finished 只表示产出了结果；guardrail 或人工审核验证内容；外部系统返回回执后才算采用。

十、CrewBase：把 YAML 和 Python 装配成项目

当 agent 和 task 多起来后，代码里到处手写构造函数会很难维护。所以 docs 推荐用 YAML 配置 agents 和 tasks， 再在继承 CrewBase 的类里通过 decorators 声明 @agent、@task、 @crew、@before_kickoff 和 @after_kickoff。 这不是另一个 runtime，而是一层项目装配：配置里写“研究员”和“报告任务”，Python 方法负责真正创建对象。 证据在 CrewBase docs example。

源码上，@agent、@task、@tool、@callback 这些 decorators 都把方法包成带标记的 wrapper。@crew wrapper 会先调用所有 task 方法和 agent 方法， 收集去重后的 agents 和 tasks，再调用用户定义的 crew 方法，并绑定 before / after kickoff callbacks。 CrewBaseMeta 则在实例化时加载配置、映射变量、收集原始方法元数据。把它放在整篇的模型里看， CrewBase 负责“项目怎么装起来”，Crew 和 Flow 才负责“运行时怎么跑”。 源码在 project decorators、 @crew wrapper 和 CrewBaseMeta。

十一、和前几篇放在一起看

下一篇看 Eino。它把视角切到 Go 生态里的 graph、component、compose 和 callback： 同样是 agent 应用框架，但 owner 会明显从“团队流程”转向“可组合的执行图”。

参考源码与文档

• CrewAI README
• CrewAI crews docs
• CrewAI flows docs
• CrewAI Agent
• CrewAI Task
• CrewAI Crew
• CrewAI Process
• CrewAI Flow runtime
• CrewAI Flow @start
• CrewAI Flow @listen
• CrewAI Flow @router
• CrewAI project decorators
• CrewAI CrewBase