Agent 工程体系 · 第 49/98 篇。内容以 2026 年 9 月可验证的公开规范和稳定接口为基线；框架版本敏感能力会明确标注，不把实验行为写成通用保证。

CrewAI：Agent、Task、Crew、Flow、状态和生产边界

CrewAI 的核心价值不在于把多个大语言模型调用包装成“团队”，而在于它把 Agent 的自主决策、Task 的任务契约、Crew 的协作运行时和 Flow 的确定性控制放进了同一套 Python 编程模型。

这几个对象解决的是不同层次的问题：

• Agent：谁具备什么能力，以及如何行动。
• Task：要完成什么工作，输入和输出是什么。
• Crew：一组 Agent 如何通过一组 Task 协作。
• Flow：多个步骤如何按事件、条件、循环和状态推进。
• 状态：一次执行过程中已经知道什么、完成了什么、下一步是什么。
• 生产边界：哪些决策可以交给模型，哪些必须由代码、策略或人工控制。

CrewAI 文档将 Crews 定位为偏自主的多 Agent 协作，将 Flows 定位为事件驱动、可控制、可持久化的工作流；两者也可以组合使用。(docs.crewai.com)

---

一、先建立分层模型：模型调用不等于 Agent 系统

一个普通的 LLM 调用可以抽象为：

$$y = M(x, c)$$

其中：

• $x$ 是当前输入；
• $c$ 是上下文；
• $M$ 是模型；
• $y$ 是输出。

它通常是一次性的：输入确定后调用模型，返回结果。

一个 Agent 则至少增加了三个要素：

$$A = (G, P, T, S)$$

其中：

• $G$：目标，即 Agent 要达到的结果；
• $P$：策略或行为设定，例如角色、背景、约束；
• $T$：工具集合；
• $S$：执行状态。

Agent 的一次行动不再只是“生成文本”，而是：

$$a_t = \pi(G, P, T, S_t, o_t)$$

其中：

• $o_t$ 是当前观察结果，例如用户输入、工具返回值或其他 Agent 的结果；
• $\pi$ 是由模型和运行时共同实现的策略；
• $a_t$ 可以是回复、工具调用、委派任务或终止。

执行工具后，状态发生变化：

$$S_{t+1} = \delta(S_t, a_t, r_t)$$

其中 $r_t$ 是工具执行结果，$\delta$ 是状态转移函数。

这一区分很重要。模型负责提出下一步行动，但以下问题必须由 Agent 运行时或业务代码处理：

1. 工具是否允许调用；
2. 参数是否符合 schema；
3. 调用是否超时；
4. 失败后是否重试；
5. 结果是否可信；
6. 是否允许继续执行；
7. 是否需要人工审批。

因此，CrewAI 的 Agent 不是一个“更长的 Prompt”，而是一个带有目标、角色、工具和执行策略的运行单元。

---

二、Agent：能力边界，而不是任务本身

2.1 Agent 的组成

在 CrewAI 中，一个 Agent 通常包含：

• role：角色；
• goal：目标；
• backstory：背景和行为上下文；
• tools：可以调用的工具；
• llm：使用的模型；
• allow_delegation：是否允许向其他 Agent 委派；
• 输出格式、记忆、知识和执行相关配置。

一个最小 Agent 如下：

from crewai import Agent

researcher = Agent(
    role="技术资料研究员",
    goal="从指定资料中提取可验证的技术事实",
    backstory=(
        "你擅长阅读技术文档，区分规范保证、实现行为和经验判断。"
        "不得把猜测写成事实。"
    ),
    tools=[],
    allow_delegation=False,
    verbose=True,
)

这里的 role 不是权限系统中的角色，goal 也不是可验证的业务约束。它们主要用于构造模型上下文，影响 Agent 的行为倾向。

例如，下面两个 Agent 的权限可能完全相同，但行为不同：

researcher = Agent(
    role="资料研究员",
    goal="收集证据并标注来源",
    backstory="优先保守判断，不足之处明确说明。",
)

writer = Agent(
    role="技术作者",
    goal="把已验证事实组织成结构清晰的文章",
    backstory="不得自行补充未经研究员确认的版本敏感信息。",
)

从工程角度看，role 和 backstory 不能代替真正的授权。下面的配置并不能阻止 Agent 访问数据库：

Agent(
    role="只读分析员",
    goal="分析销售数据",
    tools=[delete_customer_tool],  # 真实权限仍然在工具上
)

真正的安全边界应位于工具实现、服务端授权和数据访问层，而不是 Prompt。

2.2 Agent 的工具边界

工具是 Agent 从“生成建议”变成“执行动作”的入口。工具至少应该定义：

from typing import Any

class ReadOrderTool:
    name = "read_order"
    description = "按订单号读取订单的只读信息"

    def run(self, order_id: str) -> dict[str, Any]:
        if not order_id.startswith("OD-"):
            raise ValueError("非法订单号")
        return {"order_id": order_id, "status": "paid"}

把工具交给 Agent：

reader = Agent(
    role="订单查询员",
    goal="只读取订单信息并解释状态",
    backstory="不得修改订单，不得猜测不存在的字段。",
    tools=[ReadOrderTool()],
    allow_delegation=False,
)

工具层需要自行处理：

• 参数校验；
• 身份和租户校验；
• 超时；
• 幂等；
• 速率限制；
• 脱敏；
• 审计；
• 可重试和不可重试错误。

Agent 可以决定“想调用 read_order”，但不能因此获得绕过服务端授权的能力。

2.3 allow_delegation 的实际含义

允许委派意味着 Agent 可以把部分工作交给其他 Agent。它不等于自动获得一个可靠的任务规划器。

例如：

manager = Agent(
    role="项目协调员",
    goal="协调研究和写作工作",
    backstory="只有在确实需要专业能力时才委派。",
    allow_delegation=True,
)

如果没有清晰的 Agent 集合、Task 目标和输出约束，委派可能出现：

• Agent A 委派给 Agent B；
• Agent B 再委派回 Agent A；
• 多个 Agent 重复做相同工作；
• 委派结果没有格式，无法被后续任务消费；
• 协作成本高于直接完成任务。

因此，委派是 Agent 的一种行动能力，不是任务依赖图的替代品。

---

三、Task：任务契约和数据生产单元

3.1 Task 不只是 Prompt

Task 描述一次明确的工作。它至少应回答：

1. 输入是什么；
2. 要完成什么；
3. 输出必须满足什么条件；
4. 由哪个 Agent 执行；
5. 结果是否供其他 Task 使用；
6. 失败如何处理。

最小示例：

from crewai import Task

research_task = Task(
    description=(
        "分析主题：{topic}。"
        "只使用提供的资料，提取核心概念、关键边界和一个反例。"
    ),
    expected_output=(
        "输出一份 Markdown 研究摘要，包含："
        "定义、证据、边界、反例四个部分。"
    ),
    agent=researcher,
)

description 主要说明工作内容，expected_output 主要说明交付形态。二者不能替代程序化校验。

如果后续代码要求 JSON，就不应只写：

expected_output="输出 JSON"

而应使用结构化输出或额外校验。否则模型可能返回：

下面是 JSON：
{"status": "ok"}

前缀文本就可能导致解析失败。

3.2 Task 之间的依赖

假设有三个任务：

• $T_1$：收集资料；
• $T_2$：从资料中提取事实；
• $T_3$：生成最终报告。

依赖关系为：

$$T_1 \rightarrow T_2 \rightarrow T_3$$

这表示：

$$start(T_2) \geq finish(T_1)$$

并且：

$$input(T_2) = output(T_1)$$

如果存在两个相互独立的研究任务：

$$T_1 \parallel T_2$$

则只有在以下条件同时成立时才适合并行：

$$readset(T_1) \cap writeset(T_2) = \varnothing$$

$$readset(T_2) \cap writeset(T_1) = \varnothing$$

也就是说，两个任务不能互相依赖，也不能并发写同一个不可合并的资源。

在 CrewAI 中，Task 可以通过上下文接收前序任务的输出。一个典型结构是：

extract_task = Task(
    description="从资料中提取事实，并为每条事实保留来源。",
    expected_output="事实列表，每条包含 claim 和 source。",
    agent=researcher,
)

write_task = Task(
    description="根据上游事实列表，生成技术说明。",
    expected_output="结构化 Markdown 文档，不得引入上游未出现的事实。",
    agent=writer,
    context=[extract_task],
)

这里 context=[extract_task] 表示数据依赖，而不是简单地把两个任务排在列表中。依赖明确后，运行时才有机会进行顺序调度、错误传播和结果追踪。

3.3 Task 输出的三个层次

Task 输出可以分为三个层次：

文本输出

适合供人阅读，但不适合稳定驱动程序：

expected_output="一段 Markdown 总结"

JSON 输出

适合接口边界，但仍需校验：

expected_output="""
JSON 对象，字段为：
- claims: 字符串数组
- confidence: 0 到 1 之间的数字
"""

类型化输出

更适合后续 Task 或 Flow 使用。概念上应类似：

from pydantic import BaseModel, Field

class Claim(BaseModel):
    text: str
    source: str
    confidence: float = Field(ge=0, le=1)

class ResearchResult(BaseModel):
    claims: list[Claim]

然后要求 Task 返回该模型。CrewAI 文档将结构化输出和 Pydantic 作为 Agent/Task 的能力之一，但具体参数名和版本行为需要以锁定版本的 API 文档为准；不能因为模型声明了类型，就认为事实已经被验证。(docs.crewai.com)

类型校验只能检查：

confidence 是否为数字
source 是否存在
claims 是否为数组

它不能检查：

source 是否真的支持 text
text 是否与 source 矛盾

后者仍需要检索验证、规则校验或人工审核。

---

四、Crew：一组 Agent 和 Task 的协作运行时

4.1 Crew 解决什么问题

Crew 是 Agent 和 Task 的组合，并负责执行过程。它通常包含：

• Agent 集合；
• Task 集合；
• 执行过程；
• 模型配置；
• 日志和回调；
• 记忆或知识配置；
• 最终输出。

一个端到端的最小示例：

from crewai import Agent, Crew, Process, Task

researcher = Agent(
    role="资料研究员",
    goal="提取准确、可追溯的事实",
    backstory="遇到证据不足时必须明确说明。",
    allow_delegation=False,
)

writer = Agent(
    role="技术作者",
    goal="把研究结果写成结构清晰的技术文章",
    backstory="只使用研究结果中的事实，不自行扩大结论。",
    allow_delegation=False,
)

research_task = Task(
    description=(
        "研究 CrewAI 中 Agent、Task、Crew、Flow 和状态之间的关系。"
        "给出定义、数据流、失败边界和一个反例。"
    ),
    expected_output="带有定义、数据流、失败边界和反例的研究摘要。",
    agent=researcher,
)

write_task = Task(
    description=(
        "根据研究摘要撰写一篇中文技术文章。"
        "必须区分框架保证、常见实现和工程建议。"
    ),
    expected_output="完整 Markdown 技术文章。",
    agent=writer,
    context=[research_task],
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff()
print(result)

运行前需要安装 CrewAI，并配置所使用模型提供商的 API 密钥。由于模型、工具和版本都会影响行为，以上代码展示的是核心生命周期，不应被理解为固定的生产依赖清单。

执行数据流可以表示为：

flowchart LR
    I[输入 inputs] --> T1[Research Task]
    T1 --> O1[Research Output]
    O1 --> T2[Writing Task]
    T2 --> O2[Final Output]
    A1[Researcher Agent] -.执行.-> T1
    A2[Writer Agent] -.执行.-> T2

关键点是：Agent 执行 Task，Task 产生输出，后续 Task 消费输出，Crew 负责把这些对象组织成一次运行。

4.2 Process.sequential

顺序过程的语义比较直接：

Task 1 完成
    ↓
Task 2 获得 Task 1 输出
    ↓
Task 3 获得前序结果

其优点是可解释、容易调试、依赖明确。缺点是关键路径较长：

$$L_{\text{sequential}} = \sum_{i=1}^{n} latency(T_i)$$

如果 $T_1$、$T_2$、$T_3$ 没有数据依赖，却被强制顺序执行，就会增加延迟和成本。

4.3 Hierarchical：管理者调度，而不是静态流水线

层级过程通常引入一个管理 Agent，由管理者分配任务、检查结果和协调执行。它更接近：

$$manager \rightarrow \{worker_1, worker_2, \ldots, worker_n\}$$

这适合任务边界不完全确定、需要动态分派的场景，例如开放式研究。

但层级调度也引入新的不确定性：

• 管理者可能错误理解任务；
• 任务分配可能不均衡；
• 子任务可能重复；
• 管理者可能接受不完整结果；
• 失败后的重规划可能没有终止条件。

因此，层级过程适合“路径未知”的问题，不适合关键业务中的无约束审批、扣款、删除和权限变更。

4.4 Crew 的输出不是业务事实

Crew 返回的最终输出只是一次 Agent 协作的结果。它不天然具有以下属性：

• 真实性；
• 完整性；
• 唯一性；
• 可审计性；
• 事务一致性；
• 业务授权。

如果 Crew 生成了：

{"refund_amount": 1000}

这并不意味着系统可以直接退款。正确的数据流应是：

Crew 生成退款建议
    ↓
业务规则校验
    ↓
权限校验
    ↓
人工审批或策略审批
    ↓
幂等退款接口
    ↓
持久化操作结果

Crew 适合产生“建议、草稿、分析、候选计划”，而不是自动获得业务系统的最终写权限。

---

五、Flow：把 Agent 协作放进显式控制流

5.1 为什么需要 Flow

Crew 更关注“多个 Agent 如何完成一个目标”，Flow 更关注“整个业务流程下一步允许做什么”。

一个 Flow 通常具备：

• 起始步骤；
• 监听前置步骤完成的步骤；
• 路由步骤；
• 条件分支；
• 循环；
• 状态；
• 持久化和恢复；
• 错误处理；
• 可观测性。

CrewAI 文档将 Flow 描述为支持 start、listen、router 等步骤，并用于状态管理、持久化和长流程恢复。(docs.crewai.com)

一个简化示例：

from pydantic import BaseModel
from crewai.flow.flow import Flow, listen, router, start

class ReviewState(BaseModel):
    request: str = ""
    category: str = ""
    draft: str = ""
    approved: bool = False

class ReviewFlow(Flow[ReviewState]):

    @start()
    def receive(self, request: str) -> str:
        self.state.request = request
        return request

    @router(receive)
    def classify(self, request: str) -> str:
        if "退款" in request or "付款" in request:
            self.state.category = "financial"
            return "financial"

        self.state.category = "general"
        return "general"

    @listen("financial")
    def financial_review(self, request: str) -> str:
        self.state.draft = "进入财务审核队列"
        return self.state.draft

    @listen("general")
    def general_reply(self, request: str) -> str:
        self.state.draft = "进入普通客服处理"
        return self.state.draft

flow = ReviewFlow()
result = flow.kickoff(inputs={"request": "用户申请退款"})
print(result)
print(flow.state.model_dump())

需要注意的是，Flow API 在不同 CrewAI 版本中可能存在导入路径、状态声明和持久化装饰器差异。上例用于说明当前文档中的控制流模型；生产代码应固定 CrewAI 版本，并用该版本的 API 测试运行。CrewAI 文档本身同时提供多个版本文档索引，说明版本差异不能被忽略。(docs.crewai.com)

5.2 Flow 与 Crew 的组合

Flow 不需要替代 Crew。更合理的组合是：

Flow：接收请求
  ↓
Flow：分类和权限初检
  ↓
Flow：路由到财务、客服或技术分支
  ↓
Crew：在某个分支内进行研究、分析和写作
  ↓
Flow：校验输出
  ↓
Flow：人工审批或调用业务 API

代码结构可以是：

class SupportFlow(Flow[ReviewState]):

    @start()
    def receive(self, request: str):
        self.state.request = request
        return request

    @router(receive)
    def route(self, request: str):
        if "退款" in request:
            return "financial"
        return "general"

    @listen("general")
    def run_general_crew(self, request: str):
        crew = build_general_support_crew()
        self.state.draft = str(
            crew.kickoff(inputs={"request": request})
        )
        return self.state.draft

    @listen("financial")
    def run_financial_crew(self, request: str):
        crew = build_financial_analysis_crew()
        self.state.draft = str(
            crew.kickoff(inputs={"request": request})
        )
        return self.state.draft

Flow 控制“是否进入某个分支”，Crew 控制“分支内部如何完成复杂认知工作”。这也是两者最重要的边界。

5.3 Flow 的路由不是多 Agent 路由的全部

多 Agent 路由可以形式化为：

$$R(x) = \arg\max_{a \in A} score(x, capability(a))$$

其中：

• $x$ 是请求；
• $A$ 是候选 Agent 集合；
• capability(a) 是 Agent 的能力描述；
• $R(x)$ 是选择结果。

但生产路由不能只依赖模型判断，还需要：

$$R'(x) = \begin{cases} a_{\text{selected}}, & \text{分类置信度足够且 Agent 可用}\\ a_{\text{fallback}}, & \text{Agent 不可用或置信度不足}\\ human, & \text{风险等级超过阈值} \end{cases}$$

例如：

def safe_route(category: str, confidence: float, healthy: dict[str, bool]) -> str:
    if confidence < 0.8:
        return "human"

    if category == "financial":
        return "financial" if healthy.get("financial", False) else "human"

    if category == "technical":
        return "technical" if healthy.get("technical", False) else "general"

    return "general"

模型可以负责分类，代码负责阈值、可用性和回退。这种分工比让一个“总管 Agent”自由决定所有路径更容易审计。

---

六、状态：执行上下文、记忆和业务事实不是一回事

6.1 状态的定义

状态是某次执行在时间 $t$ 的可恢复上下文：

$$S_t = (input,\ progress,\ outputs,\ errors,\ approvals,\ metadata)$$

例如：

class ReviewState(BaseModel):
    request: str
    category: str | None = None
    draft: str | None = None
    retry_count: int = 0
    approved: bool = False
    last_error: str | None = None

状态的价值在于：流程不必依赖内存中的局部变量才能继续运行。

一次状态变化可能是：

S0:
request="申请退款"
category=None
draft=None
approved=False

S1:
request="申请退款"
category="financial"
draft=None
approved=False

S2:
request="申请退款"
category="financial"
draft="退款建议 100 元"
approved=False

S3:
request="申请退款"
category="financial"
draft="退款建议 100 元"
approved=True

每次变更都应能说明：

• 谁写入；
• 为什么写入；
• 是否经过校验；
• 是否可重放；
• 是否可恢复。

6.2 状态、短期记忆、长期记忆

这三个概念经常混淆。

执行状态

回答：

这一次 Flow 运行到哪里了？

例如：

step=financial_review
retry_count=1
draft=...

对话历史或短期记忆

回答：

当前 Agent 在本次会话中看过哪些消息？

它通常用于维持上下文，但不一定适合作为业务流程的唯一事实源。

长期记忆或知识库

回答：

系统跨多次运行保存了哪些可检索信息？

例如用户偏好、历史文档、领域知识。它通常需要检索、排序和时效管理。

因此：

“用户上次说过希望退款”

可能来自对话历史；

“退款审批已完成”

必须来自业务系统或经过授权的状态存储，而不能来自模型记忆。

6.3 持久化与恢复

如果 Flow 运行中断，恢复需要满足至少三个条件：

1. 状态已持久化；
2. 当前步骤和输入可确定；
3. 重试不会造成不可逆副作用。

可以将恢复表示为：

$$resume(run\_id) = execute(step(S_t), S_t)$$

但对于写操作，必须考虑幂等性。假设 Flow 在调用退款接口后、保存 approved=True 前崩溃：

1. 退款接口成功
2. 进程崩溃
3. approved 状态尚未保存
4. 恢复时再次调用退款接口

如果没有幂等键，可能发生重复退款。

正确做法是：

idempotency_key = f"refund:{request_id}"

payment_api.refund(
    request_id=request_id,
    amount=amount,
    idempotency_key=idempotency_key,
)

然后再记录：

refund_status = succeeded

Flow 的恢复能力不能自动把外部 API 变成事务系统。持久化解决的是“流程从哪里继续”，幂等解决的是“继续执行是否会重复产生副作用”。

---

七、Plan-and-Execute：任务分解、依赖、重规划和进度

CrewAI 的顺序 Task 适合预先知道步骤的流程，但开放式问题往往需要动态规划。

7.1 任务规划的形式化表示

将一个复杂目标表示为任务图：

$$G = (V, E)$$

其中：

• $V$ 是任务节点；
• $E$ 是依赖边；
• $u \rightarrow v$ 表示任务 $v$ 依赖任务 $u$。

例如“比较三个 Agent 框架”可以拆成：

P0 收集框架列表
 ├── P1 阅读框架 A 文档
 ├── P2 阅读框架 B 文档
 └── P3 阅读框架 C 文档
       ↓
P4 统一比较维度
       ↓
P5 生成报告

只有在 P1、P2、P3 之间没有共享写入时，才适合并发。

7.2 可执行计划必须包含验收条件

一个不完整的计划是：

1. 搜索资料
2. 分析资料
3. 写报告

一个可执行计划应包含：

{
  "id": "P1",
  "description": "读取框架 A 官方文档",
  "depends_on": [],
  "acceptance": [
    "至少提取 Agent、Task、Flow 的定义",
    "每个结论保留来源",
    "记录版本或更新时间"
  ],
  "status": "pending",
  "attempts": 0
}

验收条件是 Task 和状态系统之间的连接点。没有验收条件，done 只表示模型说“完成了”，不表示输出满足要求。

7.3 进度状态不是百分比

很多系统把进度表示为：

完成 60%

但这通常缺乏语义。更可靠的是显式状态：

pending
running
blocked
succeeded
failed
cancelled

一个任务的转移可以表示为：

stateDiagram-v2
    [*] --> pending
    pending --> running: scheduler dispatch
    running --> succeeded: acceptance passed
    running --> failed: non-retryable error
    running --> blocked: dependency unavailable
    failed --> pending: retry allowed
    running --> cancelled: timeout or cancellation
    succeeded --> [*]
    cancelled --> [*]

其中：

• failed 表示执行失败；
• blocked 表示当前任务可能没错，但依赖不可用；
• succeeded 必须由验收条件确认；
• pending 不代表一定会执行，可能等待资源或人工批准。

7.4 何时需要重规划

重规划不是“模型再想一次”，而是根据新事实改变任务图：

$$G_{t+1} = replan(G_t, observation_t)$$

例如：

原计划：
P1 查找官方文档
P2 比较 API
P3 写报告

观察：
官方文档没有描述某个关键行为

重规划：
P1a 查找官方 API reference
P1b 创建最小复现程序
P1c 将“未确认”写入风险清单
P2 只比较已验证能力
P3 降低结论强度

重规划应有边界：

• 最大规划轮数；
• 最大任务数；
• 最大总预算；
• 最大时间；
• 不允许降低关键验收标准；
• 高风险动作仍需人工或策略批准。

否则系统可能不断“重新研究”，表现为高 token 消耗但没有进展。

---

八、并发、故障传播和回退

8.1 并发不等于同时调用模型

并发包含至少三层：

1. 任务并发：多个独立 Task 同时执行；
2. 工具并发：一个 Agent 同时查询多个外部资源；
3. 请求并发：多个用户运行相同 Flow。

每层都可能受不同限制：

• 模型供应商速率限制；
• 工具服务连接数；
• 数据库连接池；
• 租户级配额；
• 单次运行预算。

如果三个独立研究 Task 各调用五次工具，并发度从 1 增至 3，吞吐可能提高，但下游请求数也可能从 5 增至 15。没有限流时，并行会放大故障。

8.2 错误分类

不要把所有异常都交给 Agent 重试。至少区分：

可重试：
- 临时网络错误
- 429 限流
- 短暂服务不可用
- 超时

不可重试：
- 参数校验失败
- 权限不足
- 资源不存在
- 业务规则拒绝
- 输出结构不合法且重复修复无效

需要人工：
- 高金额操作
- 证据冲突
- 身份不确定
- 风险分类置信度不足

简单的错误处理结构：

def run_with_policy(fn, *, max_retries: int = 2):
    for attempt in range(max_retries + 1):
        try:
            return fn()
        except TimeoutError:
            if attempt == max_retries:
                raise
        except PermissionError:
            raise
        except ValueError:
            raise

真正的生产实现还需要退避、熔断、超时和指标，但核心原则是：重试策略由错误类型决定，而不是由 Agent 的自我判断决定。

8.3 回退路径

回退不是简单地换一个 Agent。回退必须明确：

主路径：技术 Agent
    ↓失败
次路径：通用 Agent
    ↓仍失败
人工队列

回退时要记录：

{
  "run_id": "run-123",
  "primary_route": "technical",
  "fallback_route": "general",
  "reason": "technical_agent_timeout",
  "attempts": 2
}

否则用户只会看到一个模糊答案，运维人员无法判断系统是否发生了能力降级。

---

九、一个完整的混合示例：Flow 控制，Crew 分析

下面构造一个“技术问题分流与回答”的结构：

from pydantic import BaseModel
from crewai import Agent, Crew, Process, Task
from crewai.flow.flow import Flow, listen, router, start


class SupportState(BaseModel):
    request: str = ""
    category: str = ""
    answer: str = ""
    route_reason: str = ""


def build_technical_crew() -> Crew:
    analyst = Agent(
        role="技术问题分析员",
        goal="识别问题根因，并给出可验证的排查步骤",
        backstory=(
            "你必须区分已知事实、合理推断和待验证假设。"
            "不得声称已经执行了没有执行的命令。"
        ),
        allow_delegation=False,
    )

    task = Task(
        description=(
            "分析以下用户问题：{request}。"
            "输出：问题分类、可能原因、验证命令、风险和下一步。"
        ),
        expected_output=(
            "Markdown 格式，包含“分类”“原因”“验证”“风险”“下一步”五节。"
        ),
        agent=analyst,
    )

    return Crew(
        agents=[analyst],
        tasks=[task],
        process=Process.sequential,
        verbose=True,
    )


class SupportFlow(Flow[SupportState]):

    @start()
    def receive(self, request: str):
        self.state.request = request
        return request

    @router(receive)
    def classify(self, request: str):
        lower = request.lower()

        if any(word in lower for word in ["报错", "异常", "日志", "timeout"]):
            self.state.category = "technical"
            self.state.route_reason = "检测到故障排查意图"
            return "technical"

        self.state.category = "general"
        self.state.route_reason = "未检测到技术故障关键词"
        return "general"

    @listen("technical")
    def answer_technical(self, request: str):
        crew = build_technical_crew()
        result = crew.kickoff(inputs={"request": request})
        self.state.answer = str(result)
        return self.state.answer

    @listen("general")
    def answer_general(self, request: str):
        self.state.answer = (
            "该请求不属于技术故障排查范围，请转交通用客服流程。"
        )
        return self.state.answer


if __name__ == "__main__":
    flow = SupportFlow()
    output = flow.kickoff(
        inputs={"request": "服务启动时报 timeout，应该如何排查？"}
    )

    print(output)
    print(flow.state.model_dump())

执行路径是：

request
  ↓
receive
  ↓
classify
  ├── technical → technical Crew → answer
  └── general   → fixed response

这个设计中：

• Flow 负责分类和分支；
• Crew 负责技术问题内部的分析；
• Agent 负责解释和调用工具的决策；
• Task 规定输出契约；
• State 保存分类、路由原因和最终回答。

如果要接入真实日志工具，应把工具权限和日志脱敏放在工具实现中，而不是要求 Agent 自觉遵守。

---

十、CrewAI 与 Semantic Kernel 的概念映射

Semantic Kernel 也提供 Agent、插件、内核、线程和编排等抽象。其文档将 Kernel 描述为管理 AI 服务和插件的中心组件；Agent Framework 则提供 Agent、Agent Thread 和多 Agent 编排模式。(learn.microsoft.com)

可以做如下概念映射：

CrewAI	Semantic Kernel 中接近的概念	主要差异
Agent	Agent	都表示具有目标和能力的执行实体
Tool	Plugin / Function	都把外部能力暴露给模型
Task	一次 Agent 工作或编排步骤	CrewAI 更强调任务交付契约
Crew	多 Agent 协作单元	CrewAI 直接提供团队式抽象
Flow	Process 或 Agent Orchestration	都强调流程与编排，但 API 和生命周期不同
Flow State	Process state / Agent Thread 相关状态	状态所在层次和持久化方式不同
Process.sequential	Sequential orchestration	都表达顺序执行
并发任务	Concurrent orchestration	都需要处理结果聚合与故障传播

Semantic Kernel 的 Process Framework 是事件驱动的步骤模型，文档明确说明该框架仍处于实验阶段，可能发生变化。(learn.microsoft.com) 其 Agent Orchestration 也标注为实验性能力。(learn.microsoft.com)

因此，不能把两套框架的名字直接替换：

CrewAI 的 Flow ≠ Semantic Kernel 的 Process
CrewAI 的 Crew ≠ Semantic Kernel 的 AgentThread
CrewAI 的 Task ≠ 任意一个 Kernel Function

更准确的比较方式是看四个问题：

1. 谁负责调度；
2. 状态由谁拥有；
3. 工具调用如何授权；
4. 失败后如何恢复。

如果应用需要 Python 中快速构造 Agent 团队，CrewAI 的 Crew 模型更直接；如果应用已经围绕 Kernel、Plugin、服务注入、过滤器和 OpenTelemetry 建设，Semantic Kernel 的组件模型可能更自然。Semantic Kernel 官方定位也强调其插件、服务、Agent 和流程的组合能力。(learn.microsoft.com)

---

十一、生产边界：哪些事情不能只交给 CrewAI

11.1 不要让模型拥有最终授权

错误边界：

用户请求
  ↓
Agent 判断可以退款
  ↓
直接调用退款 API

至少应改为：

用户请求
  ↓
Agent 提取退款意图和金额
  ↓
代码校验订单、身份、金额和政策
  ↓
人工或策略审批
  ↓
带幂等键调用退款 API

Agent 可以负责非结构化信息理解，但授权必须由确定性代码和服务端策略完成。

11.2 不要把 Flow 状态当作数据库真相

Flow 状态适合保存运行上下文：

当前分类
当前步骤
临时结果
重试次数
审批等待状态

它不应替代订单、库存、支付、权限等领域系统。领域事实应有自己的存储、版本和审计。

11.3 不要用多个 Agent 掩盖任务定义不清

如果单个 Task 无法说明：

• 输入；
• 输出；
• 验收标准；
• 失败条件；

增加 Agent 数量通常只会增加不一致和成本。

多 Agent 的收益来自能力分工，例如：

研究员：收集证据
分析员：归纳和比较
审校员：检查冲突
作者：组织表达

如果四个 Agent 都在“自由回答同一个问题”，那不是协作，而是并行生成多个未经约束的答案。

11.4 不要把演示级可运行误认为生产级可靠

一个 crew.kickoff() 成功返回，只能说明：

• 代码被调用；
• 模型返回了结果；
• 运行时没有抛出未处理异常。

它不能说明：

• 结果正确；
• 工具没有误操作；
• 任务没有遗漏；
• 结果可重复；
• 失败可以恢复；
• 成本满足预算；
• 数据符合隐私要求。

生产验收应至少覆盖：

输入校验
模型和工具超时
限流与重试
结构化输出校验
事实和业务规则校验
幂等副作用
状态持久化
人工接管
日志与追踪
版本锁定
回归评测

---

十二、常见误解和诊断方法

误解一：Agent 越多，结果越好

反例：

三个 Agent 分别研究同一问题，
最后由第四个 Agent 投票。

如果三个 Agent 共享同一个错误来源，投票不会产生事实验证，只会产生一致的错误。

诊断方式：

• 比较每个 Agent 的证据集合；
• 检查是否存在独立来源；
• 检查最终结论是否引用了任何可验证材料；
• 统计重复调用和无效委派。

误解二：Task 成功就代表业务成功

Task 的成功通常只表示执行过程正常结束。应区分：

execution_success
acceptance_success
business_success

例如：

{
  "execution_success": true,
  "acceptance_success": false,
  "business_success": false,
  "reason": "输出缺少来源字段"
}

误解三：有了状态就能自动恢复

如果状态只保存：

{"step": "refund"}

却没有保存：

{
  "request_id": "...",
  "idempotency_key": "...",
  "external_effect_status": "unknown"
}

系统恢复时仍无法判断退款是否已经发生。

误解四：路由器选择了 Agent，就完成了路由

真正的路由还应检查：

• Agent 是否健康；
• 是否有能力处理当前输入；
• 当前租户是否允许；
• 是否超过风险阈值；
• 是否有可用回退；
• 是否需要人工处理。

误解五：Flow 越确定，结果越确定

Flow 可以确定执行顺序，却不能保证模型输出确定。需要区分：

$$deterministic(control\ flow) \neq deterministic(model\ output)$$

Flow 可以保证“先分类、再调用 Crew、最后校验”，但不能单独保证分类结果和文章内容完全一致。

---

结语：用 Crew 表达协作，用 Flow 表达边界

CrewAI 的几个核心对象可以归纳为一条执行链：

Agent 定义能力
  ↓
Task 定义工作和交付
  ↓
Crew 组织多 Agent 协作
  ↓
Flow 控制业务路径
  ↓
State 保存执行上下文
  ↓
外部系统完成真实业务提交

其中最重要的工程判断不是“应该创建几个 Agent”，而是：

哪一部分需要模型的开放式判断？
哪一部分必须由显式状态机控制？
哪一部分必须经过确定性校验？
哪一部分涉及不可逆副作用？

开放式研究、资料归纳、草稿生成可以交给 Crew；分类、依赖、重试、回退、审批和恢复应由 Flow 与代码控制；支付、删除、权限变更等不可逆操作必须落在经过授权、校验和幂等保护的业务服务上。

这就是 CrewAI 从 Agent 演示走向生产系统时最重要的边界。

---

系列导航与关联阅读

• 系列入口：Agent 工程完整路线：从运行循环、记忆与协议到安全、评测和生产交付
• 上一篇：Microsoft AutoGen：Agent、Team、消息、终止、运行时和扩展
• 下一篇：Semantic Kernel Agent：Plugin、Process、Memory、编排和服务集成
• 延伸：Agent Plan-and-Execute：任务分解、依赖、重规划和进度状态
• 延伸：多 Agent 路由：分类、能力匹配、动态选择、回退和评测

官方资料

• CrewAI Documentation
• Semantic Kernel Documentation

本文依据 Agent、模型、协议与框架官方资料重新梳理；正文、示例与生产清单由 WR BLOG 编写。