第 18 章 · 多 Crew 编排与复用

本章目标：掌握在 Flow 中串联多个 Crew 的组合模式，学会用共享 state 在 Crew 之间传递产物，能按业务域拆分 crews 包并设计可复用的企业级项目结构。

18.1 什么时候需要多个 Crew

单个 Crew 内的角色共享同一个任务链与上下文，一旦业务出现以下特征就该拆分：

阶段性质不同：调研（探索、多工具）与写作（纯生成）混在一个 Crew 里，max_iter、模型档位、温度都无法分别调优；
复用诉求：同一个"研究 Crew"要被周报、月报、竞品分析三条流水线共用；
失败隔离：某一步挂了只想重跑该阶段，而不是整条链。

拆分原则：一个 Crew 对应一个内聚的业务能力（研究能力、写作能力、审核能力），Flow 负责把它们串成端到端流程。这正是官方 crewai create flow 项目的设计意图——flow 工程里天然有一个 crews/ 目录。

18.2 组合模式一：Flow 方法直接调用多个 Crew

最直接的编排：每个 Flow 步骤实例化一个 Crew 并 kickoff，产物写入 state。沿用第 17 章脚手架的形态：

python

# src/my_pipeline/main.py —— 研究 Crew + 写作 Crew 的串联
import os
from pydantic import BaseModel
from crewai.flow.flow import Flow, listen, start
from my_pipeline.crews.research_crew.research_crew import ResearchCrew
from my_pipeline.crews.writing_crew.writing_crew import WritingCrew

class PipelineState(BaseModel):
    topic: str = ""
    research: str = ""      # 研究产物
    article: str = ""       # 写作产物

class ContentPipeline(Flow[PipelineState]):

    @start()
    def receive_topic(self):
        self.state.topic = "AI Agent 框架对比"

    @listen(receive_topic)
    def run_research(self):
        # 阶段一：研究 Crew（探索型：工具多、迭代上限高）
        out = ResearchCrew().crew().kickoff(
            inputs={"topic": self.state.topic}
        )
        self.state.research = out.raw       # 产物进 state，供下游使用
        return out.raw

    @listen(run_research)
    def run_writing(self):
        # 阶段二：写作 Crew（生成型：无搜索工具、温度稍高）
        out = WritingCrew().crew().kickoff(
            inputs={"research": self.state.research}
        )
        self.state.article = out.raw
        return out.raw

if __name__ == "__main__":
    print(ContentPipeline().kickoff())

三个工程要点：

每个 Crew 的 inputs 键要与它自己的 YAML 占位符对应——接口契约显式化是可维护性的关键；
state 里只放跨阶段的"正式产物"，中间草稿留在各 Crew 的任务输出里；
若阶段二失败只需重跑，可以配合第 16 章的持久化从 run_writing 前的快照恢复，不必重做研究。

18.3 组合模式二：轻步骤 + 重 Crew 的混合流水线

并非每步都值得一个 Crew。成熟项目通常是"轻量 LLM/单 Agent 步骤打头阵，重 Crew 干核心活"：

python

# hybrid_flow.py —— 轻重混合的典型布局
from pydantic import BaseModel
from crewai.flow.flow import Flow, listen, router, start

class TriageState(BaseModel):
    request: str = ""
    category: str = ""
    answer: str = ""

class SupportPipeline(Flow[TriageState]):

    @start()
    def triage(self):
        """分诊：一个便宜的小步骤决定走哪条流水线"""
        text = self.state.request.lower()
        self.state.category = (
            "billing" if "发票" in text or "账单" in text else "tech"
        )

    @router(triage)
    def route(self):
        return self.state.category          # "billing" 或 "tech"

    @listen("billing")
    def billing_flow(self):
        from my_pipeline.crews.billing_crew.billing_crew import BillingCrew
        out = BillingCrew().crew().kickoff(
            inputs={"question": self.state.request})
        self.state.answer = out.raw
        return out.raw

    @listen("tech")
    def tech_flow(self):
        from my_pipeline.crews.tech_crew.tech_crew import TechCrew
        out = TechCrew().crew().kickoff(
            inputs={"question": self.state.request})
        self.state.answer = out.raw
        return out.raw

这种"分诊 → 路由 → 专职 Crew"的结构就是客服工单系统的标准解法：分诊用规则或单次 LLM 调用（快且便宜），重活交给领域 Crew。

18.4 按业务域组织 crews 包

脚手架生成的 flow 项目目录天然支持多 Crew 平铺：

text

my_pipeline/
├── pyproject.toml
└── src/my_pipeline/
    ├── main.py                  # Flow 编排层
    ├── crews/
    │   ├── research_crew/
    │   │   ├── config/
    │   │   │   ├── agents.yaml
    │   │   │   └── tasks.yaml
    │   │   └── research_crew.py
    │   ├── writing_crew/
    │   │   ├── config/*.yaml
    │   │   └── writing_crew.py
    │   └── review_crew/
    │       └── ...
    └── tools/
        └── custom_tool.py       # 跨 Crew 共用的自定义工具

企业级项目的四条结构纪律：

每个 crew 自包含：config YAML + 定义文件齐备，可独立测试（ResearchCrew().crew().kickoff(...) 单独跑通）；
工具下沉：多个 crew 共用的工具放进顶层 tools/ 包，禁止互相 import 对方的私有模块；
LLM 配置集中：base_url/api_key/model 档位收敛到一个 llm_config.py，避免散落硬编码；
契约即文档：每个 crew 的 README 写明它需要的 inputs 键和产出的 outputs 结构。

新版脚手架还支持 JSON-first 形态的内嵌 crew（crews/research_crew/crew.jsonc + agents/researcher.jsonc），在 Flow 步骤中用官方提供的加载器读取：

python

from pathlib import Path
from crewai.project import load_crew

# JSON-first 内嵌 crew 的标准装载方式
crew, default_inputs = load_crew(
    Path(__file__).parent / "crews" / "research_crew" / "crew.jsonc"
)
result = crew.kickoff(inputs={**default_inputs, "topic": "AI Agents"})

18.5 复用与配置共享

多 Crew 之后马上会遇到重复配置问题。两个实用手法：

python

# llm_config.py —— 集中管理模型档位
import os
from crewai import LLM

def fast_llm():     # 分诊、摘要等轻任务
    return LLM(model="openai/deepseek-chat",
               base_url="https://api.deepseek.com/v1",
               api_key=os.getenv("DEEPSEEK_API_KEY"),
               temperature=0.2)

def deep_llm():     # 研究分析等重任务
    return LLM(model="openai/deepseek-chat",
               base_url="https://api.deepseek.com/v1",
               api_key=os.getenv("DEEPSEEK_API_KEY"),
               temperature=0.7,
               max_tokens=4096)

python

# 各 crew.py 中统一引用，档位调整只改一处
from my_pipeline.llm_config import deep_llm

@agent
def researcher(self) -> Agent:
    return Agent(config=self.agents_config["researcher"], llm=deep_llm())

另一个层面是产物 schema 复用：把 Crew 间传递的数据结构定义成公共 Pydantic 模型（如 ResearchReport），上游用 output_pydantic 输出、下游在 description 中声明按此结构消费，接口就具备了编译期检查能力。

18.6 端到端示例：三 Crew 内容工厂

把本章所有模式收拢成一个完整骨架（省略 YAML 细节，聚焦编排）：

python

# factory.py —— 研究 → 写作 → 审核的三段式内容工厂
from pydantic import BaseModel
from crewai.flow.flow import Flow, listen, router, start
from my_pipeline.llm_config import fast_llm
from my_pipeline.crews.research_crew.research_crew import ResearchCrew
from my_pipeline.crews.writing_crew.writing_crew import WritingCrew
from my_pipeline.crews.review_crew.review_crew import ReviewCrew

class FactoryState(BaseModel):
    topic: str = ""
    draft: str = ""
    verdict: str = ""
    final: str = ""

class ContentFactory(Flow[FactoryState]):

    @start()
    def intake(self):
        self.state.topic = "向量数据库选型指南"

    @listen(intake)
    def stage_research(self):
        out = ResearchCrew().crew().kickoff(inputs={"topic": self.state.topic})
        self.state.draft = out.raw

    @listen(stage_research)
    def stage_write(self):
        out = WritingCrew().crew().kickoff(inputs={"research": self.state.draft})
        self.state.draft = out.raw

    @listen(stage_write)
    def stage_review(self):
        out = ReviewCrew().crew().kickoff(inputs={"draft": self.state.draft})
        self.state.verdict = out.raw      # 约定输出 "PASS" 或修改意见

    @router(stage_review)
    def gate(self):
        return "publish" if "PASS" in self.state.verdict else "revise"

    @listen("publish")
    def done(self):
        self.state.final = self.state.draft
        return f"已发布：《{self.state.topic}》"

    @listen("revise")
    def back_to_writer(self):
        # 审核不通过：携带意见回到写作阶段（简化版重写回路）
        out = WritingCrew().crew().kickoff(inputs={
            "research": self.state.draft,
            "feedback": self.state.verdict,
        })
        self.state.final = out.raw
        return "修订后发布"

这个例子覆盖了多 Crew 编排的全部要素：阶段拆分、state 接力、质量闸门路由、带反馈的重试回路。第 23 章的综合实战会把它扩展成可直接运行的项目。

18.7 本章小结

拆分信号：阶段性质不同、需要复用、需要失败隔离；一个 Crew = 一个内聚业务能力；
组合模式：Flow 步骤顺序调用多个 Crew；轻重混合时用轻量分诊 + @router 引流到专职 Crew；
目录纪律：crews/ 下自包含、工具下沉 tools/、LLM 配置集中、inputs/outputs 契约文档化；
JSON-first 内嵌 crew 用 load_crew() 装载；经典 YAML crew 直接实例化调用；
质量闸门（review → router → 带反馈回写）是多 Crew 流水线的标配容错手段。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. 把"调研"和"写作"拆成两个 Crew 的核心理由不包括？

2. Flow 中串联两个 Crew 时，上游产物传给下游的标准方式是？

3. 关于 crews 包的组织纪律，正确的是？

4. "审核 Crew 给出 PASS 或修改意见，不通过则带意见重写"，正确的实现骨架是？

🛠️ 动手实践

把本章的内容工厂补全为可运行项目：用 --classic 脚手架生成三个 crew，编写各自的 agents.yaml/tasks.yaml 与统一的 llm_config.py。
为研究 Crew 和写作 Crew 分别设置不同的模型参数（低温低迭代 vs 高温高产出），用 usage_metrics 记录每个阶段的 token 分布。
在内容工厂中加入第三条路由：审核连续两次不通过则转"人工处理"分支（提示：在 state 里加 retry_count，在 gate 里判断）。

至此多 Crew 编排已成体系。接下来为团队接入外部能力生态：第 19 章 · MCP 集成。

第 18 章 · 多 Crew 编排与复用 ​

18.1 什么时候需要多个 Crew ​

18.2 组合模式一：Flow 方法直接调用多个 Crew ​

18.3 组合模式二：轻步骤 + 重 Crew 的混合流水线 ​

18.4 按业务域组织 crews 包 ​

18.5 复用与配置共享 ​

18.6 端到端示例：三 Crew 内容工厂 ​

18.7 本章小结 ​

🧪 随堂测验

🛠️ 动手实践 ​

第 18 章 · 多 Crew 编排与复用

18.1 什么时候需要多个 Crew

18.2 组合模式一：Flow 方法直接调用多个 Crew

18.3 组合模式二：轻步骤 + 重 Crew 的混合流水线

18.4 按业务域组织 crews 包

18.5 复用与配置共享

18.6 端到端示例：三 Crew 内容工厂

18.7 本章小结

🛠️ 动手实践