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

Microsoft AutoGen：Agent、Team、消息、终止、运行时和扩展

AutoGen 的核心不是“让多个模型互相聊天”，而是把 Agent 系统拆成几个可替换的运行时构件：Agent 负责处理消息并维护状态，消息负责表达事实，Team 负责组织协作，终止条件负责定义何时停止，Runtime 负责寻址、生命周期和投递，Extension 负责接入模型、工具、执行器和分布式基础设施。

截至 2026 年 9 月，AutoGen 官方文档将 Python 体系分为三个主要层次：

• AgentChat：面向应用开发的高层 API，提供 AssistantAgent、RoundRobinGroupChat、SelectorGroupChat、Swarm 等现成组件。
• Core：面向可扩展、多 Agent、事件驱动系统的底层 API，提供 Agent、Runtime、消息、Topic、Subscription 等基础抽象。
• Extensions：连接模型服务、MCP、代码执行器、分布式 Runtime 等外部系统的实现集合。(microsoft.github.io)

因此，理解 AutoGen 时不能只从 AssistantAgent 开始。AssistantAgent 解决的是“一个 Agent 如何调用模型”，而 Core 解决的是“多个有状态实体如何在一个可寻址、可调度、可扩展的系统中协作”。

---

一、先建立心智模型：Agent 不是提示词，Team 不是消息列表

AutoGen Core 中的 Agent 可以形式化为一个有状态的消息处理实体：

$$A = (id, state, handlers, effects)$$

其中：

• id 是 Agent 的唯一地址；
• state 是 Agent 的内部状态，例如任务进度、缓存、会话上下文；
• handlers 是它能够处理的消息类型及对应逻辑；
• effects 是处理消息产生的外部影响，例如发送消息、发布事件、调用 API 或执行代码。

当 Agent 收到消息 $m$ 时，可以抽象为：

$$(state_{t+1}, E_t) = \delta(state_t, m_t)$$

其中：

• $\delta$ 是 Agent 的处理函数；
• $state_{t+1}$ 是处理后的新状态；
• $E_t$ 是外部副作用集合，例如新消息或数据库更新。

这个定义有两个直接后果。

第一，模型不是 Agent 的全部。模型只是 Agent 内部可能使用的一个推理组件。一个确定性的规则处理器、代码执行器或外部服务适配器也可以实现为 Agent。

第二，消息是 Agent 之间的唯一通信媒介。Agent 不应直接持有另一个 Agent 的 Python 对象并调用其方法；它们通过 Runtime 进行寻址和投递。AutoGen 官方将 Agent 定义为能够通过消息通信、维护自己的状态，并在收到消息或状态变化后执行动作的软件实体。(microsoft.github.io)

Team 则是对多个 Agent 及其协作策略的封装。它通常包含：

$$Team = (Participants, Scheduler, Context, Termination)$$

其中：

• Participants 是参与者；
• Scheduler 决定下一步由谁处理；
• Context 是团队共享或传递的对话上下文；
• Termination 决定团队何时结束。

所以，Team 不是简单的 list[Agent]。它还必须回答一个控制流问题：

下一条消息应该交给谁，交给多少次，哪些消息可见，什么时候认为任务完成？

---

二、AutoGen 的三层结构

1. AgentChat：用预制协作模式快速构建应用

AgentChat 是建立在 Core 之上的高层 API，适合先验证任务流程。它提供的常见 Team 包括：

Team	调度方式	适合场景
RoundRobinGroupChat	按固定顺序轮流发言	反思、审稿、简单协作
SelectorGroupChat	每轮由模型选择下一位 Agent	动态路由、能力匹配
Swarm	通过 HandoffMessage 转交控制权	专家接力、状态转移
MagenticOneGroupChat	面向开放式任务的通用多 Agent 编排	Web、文件和复杂任务

这些 Team 都继承自 BaseGroupChat，可以配置参与者、终止条件、最大轮数、Runtime 和自定义消息类型。(microsoft.github.io)

2. Core：事件驱动和运行时基础设施

Core API 更接近消息系统，而不是聊天 API。开发者通常通过 RoutedAgent 和 @message_handler 定义 Agent，再将 Agent 类型注册到 Runtime。

Core API 的优点是：

• Agent 逻辑和消息投递解耦；
• 可以在单进程或分布式 Runtime 中运行；
• 可以使用点对点消息或发布订阅；
• 可以自行定义消息类型、路由规则和生命周期。

代价是：开发者需要自己设计协议、状态机、终止信号、幂等行为和错误处理。官方也明确指出，Core API 更灵活但更不具备约束性；如果只是快速构建多 Agent 应用，应优先考虑 AgentChat。(microsoft.github.io)

3. Extensions：把框架边界连接到现实系统

Extensions 是 AutoGen 与外部系统之间的适配层，例如：

• OpenAI 等模型客户端；
• MCP Workbench；
• Docker 代码执行器；
• Assistant API Agent；
• gRPC 分布式 Agent Runtime。

官方将这些实现放在独立扩展包中，而不是把所有第三方能力直接塞进核心包。这样做的原因是：核心抽象保持稳定，外部依赖可以独立发布、升级和替换。(microsoft.github.io)

---

三、Agent：类型、实例和地址必须区分

AutoGen 中最容易混淆的是“Agent 类”“Agent 类型”和“Agent 实例”。

1. Agent 类

这是 Python 中的实现类，例如：

class InvoiceAgent(RoutedAgent):
    ...

它描述行为，但不是 Runtime 中的地址。

2. Agent 类型

Agent 类型是注册到 Runtime 的字符串，例如：

"invoice_agent"

它关联一个工厂函数。不同 Agent 类型可以由同一个 Python 类实现，只是构造参数不同。

3. Agent 实例

实例由：

$$AgentId = (AgentType, AgentKey)$$

唯一确定。例如：

("invoice_agent", "tenant-42")
("invoice_agent", "tenant-99")

这两个实例使用相同的 Agent 类和 Agent 类型，但分别拥有自己的状态。官方将 Agent ID 定义为由 Agent Type 和 Agent Key 组成的地址；Runtime 在第一次向该地址投递消息时，可以根据工厂函数自动创建实例。(microsoft.github.io)

这个模型天然适合多租户和会话隔离：

AgentType = "review_agent"
AgentKey  = review_request_id

于是每个审查请求都可以拥有一个独立的 review_agent 实例。状态不会因为多个请求复用同一个 Python 对象而混在一起。

需要注意，官方文档中提到的 Agent 实例分页进出能力目前并不是已经实现的通用能力。因此，不能把“Runtime 管理生命周期”误解为“所有实例都自动持久化、自动恢复、自动淘汰”。(microsoft.github.io)

---

四、消息：纯数据、可序列化、带协议语义

AutoGen Core 中的消息是可序列化对象，通常使用 dataclass 或 Pydantic BaseModel 定义。消息本身应该是纯数据，不应包含业务逻辑。(microsoft.github.io)

from dataclasses import dataclass


@dataclass
class ReviewRequest:
    request_id: str
    document: str


@dataclass
class ReviewResult:
    request_id: str
    approved: bool
    reason: str

消息至少要表达三类信息：

1. 业务载荷：例如文档内容、订单号、分类结果；
2. 关联信息：例如 request_id、correlation_id、causation_id；
3. 协议版本或幂等字段：例如 schema_version、message_id、attempt。

一个生产消息不应只有：

@dataclass
class Message:
    content: str

因为当系统出现重试、并发、乱序或跨租户投递时，仅凭 content 无法判断：

• 这条消息属于哪个请求；
• 它由谁产生；
• 它是否已经处理过；
• 它是否是旧版本消息；
• 它是否应该覆盖当前状态。

可以改为：

from dataclasses import dataclass


@dataclass
class ReviewRequest:
    message_id: str
    correlation_id: str
    causation_id: str | None
    schema_version: int
    sequence: int
    document: str

这里：

• correlation_id 标识一次端到端业务流程；
• causation_id 指向产生当前消息的上一条消息；
• sequence 表示同一关联流程中的逻辑顺序；
• message_id 用于幂等去重。

这几个字段不是 AutoGen 强制要求的统一标准，而是事件驱动系统中常见的工程协议。AutoGen 提供消息传输能力，但不会替应用自动推导业务关联关系。

---

五、直接消息和广播消息

AutoGen Core 提供两种基本通信方式：

直接消息：sender ─────────────> receiver AgentId
广播消息：sender ──> Topic ──> 多个订阅者

1. 直接消息：明确请求和响应

直接消息需要指定接收方的 AgentId：

await runtime.send_message(
    ReviewRequest(
        message_id="m-001",
        correlation_id="r-001",
        causation_id=None,
        schema_version=1,
        sequence=1,
        document="合同内容",
    ),
    AgentId("review_agent", "r-001"),
)

它适合：

• 请求—响应；
• 明确的任务委派；
• 已经完成路由选择的消息；
• 需要知道确定接收方的控制流。

如果一个分类 Agent 已经选择了“法律审查 Agent”，就可以直接把消息发送到：

AgentId("legal_review_agent", request_id)

直接消息的代价是发送方必须知道接收者的类型和键，因此耦合更强。

2. 广播消息：发布到 Topic

广播消息不指定具体 Agent，而是发布到 Topic。Topic 由：

$$Topic = (TopicType, TopicSource)$$

组成。

例如：

("invoice_created", "tenant-42/order-1001")

其中：

• TopicType 表示事件类别；
• TopicSource 表示该类别下的具体业务范围或实例。

Subscription 再将 Topic 映射到 Agent。没有订阅者时，消息不会送达；有多个订阅者时，每个匹配的订阅者都会收到消息。(microsoft.github.io)

3. 类型订阅和多租户隔离

类型订阅可以抽象为：

$$TopicType \rightarrow AgentType$$

例如：

"invoice_created" -> "fraud_agent"
"invoice_created" -> "accounting_agent"

当 Topic Source 为 tenant-42/order-1001 时，Runtime 可以将其映射为：

AgentId("fraud_agent", "tenant-42/order-1001")
AgentId("accounting_agent", "tenant-42/order-1001")

这意味着 Topic Source 不只是过滤条件，还可以成为 Agent Key，从而自动形成按租户、请求或订单隔离的 Agent 实例。官方文档将这种模式用于多租户场景：同一 Agent 类型通过不同的 topic source 产生不同 Agent 实例。(microsoft.github.io)

---

六、事件驱动 Agent 的消息流

一个事件驱动的发票处理系统可以设计为：

sequenceDiagram
    participant API as API
    participant RT as Agent Runtime
    participant T as Topic<br/>invoice_created/tenant-42
    participant F as Fraud Agent
    participant A as Accounting Agent
    participant O as Orchestrator

    API->>RT: publish InvoiceCreated
    RT->>T: resolve subscriptions
    T->>F: deliver InvoiceCreated
    T->>A: deliver InvoiceCreated
    F->>RT: publish FraudChecked
    A->>RT: publish AccountingChecked
    RT->>O: deliver result events
    O->>O: check correlation_id and completion

这里有一个关键区别：

广播只保证“把事件交给匹配的订阅者”，不天然形成请求—响应协议。

如果发布消息的 Agent 的 Handler 返回了一个结果，这个返回值不会自动成为发布者的响应。官方明确区分了广播和直接消息：广播是一种单向发布订阅机制，不能用于请求—响应；发布消息的返回值也不会被接收方自动返回给发布者。(microsoft.github.io)

因此，广播场景需要显式定义结果事件：

@dataclass
class FraudChecked:
    correlation_id: str
    approved: bool


@dataclass
class AccountingChecked:
    correlation_id: str
    valid: bool

Orchestrator 通过 correlation_id 聚合两个结果：

$$complete(r) = fraud\_checked(r) \land accounting\_checked(r)$$

如果两个 Agent 的处理顺序不固定，Orchestrator 就不能依赖“先收到 Fraud，再收到 Accounting”。它必须按关联 ID 保存部分结果：

state[correlation_id] = {
    "fraud": ...,
    "accounting": ...,
}

当两个字段都存在时才生成最终事件。

这就是事件驱动 Agent 中的最终一致性：系统不要求所有消费者同时完成，而是允许状态分阶段到达，最终通过聚合条件形成一致结果。

顺序不是广播的默认语义

即使某个消费者看到了消息 A、B，也不应直接假设：

publish(A) -> handler(A) 完成 -> publish(B)

就等价于所有 Agent 都按 A、B 顺序处理。不同订阅者可能有不同处理速度；分布式 Runtime 还可能存在跨进程传输、重试和暂时不可用。

需要强顺序时，应显式设计：

• 同一 AgentKey 串行消费；
• sequence 字段；
• 前置事件完成后再发布后继事件；
• 幂等去重；
• 乱序缓存和超时处理。

AutoGen 提供通信和订阅抽象，但不会替应用自动提供跨消费者的全局顺序保证。

---

七、一个可运行的 Core 示例：检查和递减任务

下面的例子不依赖模型服务，展示 Core 的完整生命周期：

1. 定义消息；
2. 定义 Agent；
3. 注册 Agent 类型；
4. 启动 Runtime；
5. 发送消息；
6. Agent 继续发布消息；
7. Runtime 变为空闲；
8. 关闭 Runtime。

import asyncio
from dataclasses import dataclass
from typing import Callable

from autogen_core import (
    AgentId,
    DefaultTopicId,
    MessageContext,
    RoutedAgent,
    SingleThreadedAgentRuntime,
    default_subscription,
    message_handler,
)


@dataclass
class NumberMessage:
    value: int


@default_subscription
class Modifier(RoutedAgent):
    def __init__(self, modify: Callable[[int], int]) -> None:
        super().__init__("modify a number")
        self._modify = modify

    @message_handler
    async def handle(
        self,
        message: NumberMessage,
        ctx: MessageContext,
    ) -> None:
        new_value = self._modify(message.value)
        print(f"Modifier: {message.value} -> {new_value}")

        await self.publish_message(
            NumberMessage(new_value),
            DefaultTopicId(),
        )


@default_subscription
class Checker(RoutedAgent):
    def __init__(self, stop_at: Callable[[int], bool]) -> None:
        super().__init__("check a number")
        self._stop_at = stop_at

    @message_handler
    async def handle(
        self,
        message: NumberMessage,
        ctx: MessageContext,
    ) -> None:
        if self._stop_at(message.value):
            print(f"Checker: stop at {message.value}")
            return

        print(f"Checker: continue from {message.value}")
        await self.publish_message(
            NumberMessage(message.value),
            DefaultTopicId(),
        )


async def main() -> None:
    runtime = SingleThreadedAgentRuntime()

    await Modifier.register(
        runtime,
        "modifier",
        lambda: Modifier(lambda value: value - 1),
    )

    await Checker.register(
        runtime,
        "checker",
        lambda: Checker(lambda value: value <= 1),
    )

    runtime.start()

    await runtime.send_message(
        NumberMessage(5),
        AgentId("checker", "default"),
    )

    await runtime.stop_when_idle()
    await runtime.close()


if __name__ == "__main__":
    asyncio.run(main())

安装依赖：

pip install -U autogen-core
python demo.py

预期输出的顺序可能体现消息调度过程，但核心结果是：

Checker: continue from 5
Modifier: 5 -> 4
Checker: continue from 4
Modifier: 4 -> 3
Checker: continue from 3
Modifier: 3 -> 2
Checker: continue from 2
Modifier: 2 -> 1
Checker: stop at 1

这个例子中的初始消息直接发送给：

AgentId("checker", "default")

Checker 处理后发布到默认 Topic。由于两个类都使用 @default_subscription，发布消息会被匹配到相应订阅者。Runtime 会在第一次投递消息时创建对应 Agent 实例。官方 Quick Start 使用了同样的“注册—启动—发送—等待空闲”生命周期。(microsoft.github.io)

这个示例还暴露了一个重要边界：Modifier 和 Checker 订阅同一个 Topic，而发布者不会再次收到自己发布的消息；AutoGen 特意避免了发布者因为自身订阅而形成无限自循环。(microsoft.github.io)

---

八、Runtime：不是线程池，而是寻址、投递和生命周期边界

Runtime 至少承担四件事：

1. 消息投递：把直接消息或广播消息送到目标 Agent；
2. 身份管理：解析 AgentId、Topic 和 Subscription；
3. 生命周期管理：根据 Agent 类型和工厂函数创建实例；
4. 运行控制：启动、停止、等待空闲、关闭资源。

1. SingleThreadedAgentRuntime

SingleThreadedAgentRuntime 适合单进程本地应用。它可以在后台处理消息：

runtime.start()

也可以等待所有未处理消息完成：

await runtime.stop_when_idle()

还可以停止后台处理：

await runtime.stop()

stop() 不会取消已经在执行中的 Handler，而是停止继续处理后台任务；之后可以再次 start()。关闭资源则使用：

await runtime.close()

这些操作语义不同：

操作	语义
start()	开始后台处理
stop()	停止后台处理，不取消正在执行的 Handler
stop_when_idle()	等待队列和正在处理的 Handler 都空闲
close()	释放 Runtime 资源

官方文档明确区分了 stop()、stop_when_idle() 和 close()，不能把它们当作同一个“停止函数”。(microsoft.github.io)

2. 分布式 Runtime

分布式 Runtime 面向多进程、多机器甚至不同语言实现的 Agent。官方描述的结构包括 Host Service、Worker 和 Gateway：

Worker A ── Gateway ──┐
Worker B ── Gateway ──┼── Host Service
Worker C ── Gateway ──┘

Host Service 协调跨 Worker 的通信并维护连接状态；Worker 承载 Agent，并向 Host Service 广告自己能够运行的 Agent 类型。Agent 的实现方式应尽量保持不变，以便从单机 Runtime 切换到分布式 Runtime。(microsoft.github.io)

但“接口相同”不等于“运行语义完全相同”。跨进程后会引入新的故障路径：

• Worker 暂时不可达；
• 消息投递超时；
• Handler 已产生副作用但响应丢失；
• 消息重试导致重复执行；
• Agent 状态未持久化而无法恢复；
• 版本不兼容导致消息反序列化失败。

因此，分布式 Agent 必须把 Handler 设计成幂等或可补偿：

$$process(m) = \begin{cases} ignore, & message\_id \in processed \\ apply(m), & otherwise \end{cases}$$

实际系统可以将 message_id 和处理结果写入持久化存储，再执行外部副作用。否则，下面的失败路径会产生重复扣款：

收到支付消息
  -> 调用支付网关成功
  -> 进程崩溃
  -> Runtime 重试同一消息
  -> 再次调用支付网关

AutoGen Runtime 提供的是执行和通信基础设施，不会替应用自动完成外部副作用的 exactly-once 语义。

---

九、AgentChat 中的 Agent 和 Core Agent 不是同一个生命周期模型

AgentChat 的典型使用方式是应用代码直接创建 Agent：

from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(model="gpt-4o")

assistant = AssistantAgent(
    name="assistant",
    model_client=model_client,
)

而 Core 的惯用方式是：

await MyAgent.register(
    runtime,
    "my_agent",
    lambda: MyAgent(),
)

然后由 Runtime 根据 AgentId 创建实例。

这两个层次不能直接混用。官方明确指出，AgentChat 中的 AssistantAgent 通常由应用代码创建，不直接由 Core Runtime 管理；如果要在 Core 中使用它，需要用一个 RoutedAgent 作为包装器，将 Core 消息转换为 AgentChat 消息，再调用被包装 Agent。(microsoft.github.io)

包装器的基本结构如下：

from dataclasses import dataclass

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_core import MessageContext, RoutedAgent, message_handler


@dataclass
class UserRequest:
    text: str


class AssistantWrapper(RoutedAgent):
    def __init__(self) -> None:
        super().__init__("wrap an AgentChat assistant")

        self._assistant = AssistantAgent(
            name="assistant",
            model_client=OpenAIChatCompletionClient(
                model="gpt-4o",
            ),
        )

    @message_handler
    async def handle(
        self,
        message: UserRequest,
        ctx: MessageContext,
    ) -> None:
        response = await self._assistant.on_messages(
            [
                TextMessage(
                    content=message.text,
                    source="user",
                )
            ],
            cancellation_token=ctx.cancellation_token,
        )

        print(response.chat_message)

这个包装器的意义不是“多写一层代码”，而是明确两个边界：

Core Runtime 管理：地址、注册、投递、生命周期
AgentChat Agent 管理：模型调用、对话上下文、单步响应

如果不区分这两个生命周期，常见错误是把一个有状态的 AgentChat Agent 放进多个 Core Agent 实例之间共享，最终造成会话上下文串线。

---

十、Team：固定调度、动态选择和交接

1. RoundRobinGroupChat

Round-robin 是最确定的调度方式：

A -> B -> C -> A -> B -> C

它适合：

• 主 Agent 生成草稿，批评 Agent 审查；
• 多轮反思；
• 每个参与者都必须获得一次发言机会的流程。

它的问题也很明确：即使下一步实际只需要某个专家，调度器仍可能让不相关 Agent 发言，造成额外模型调用和无效上下文。

2. SelectorGroupChat

Selector Group Chat 在每轮消息后，通过 ChatCompletion 模型选择下一位发言者。它适合动态路由，但“由模型选择”不等于“能力匹配已经正确”。

可以将选择建模为：

$$a^* = \arg\max_{a \in A} \left( R(a, x) - \lambda C(a) - \mu H(a) \right)$$

其中：

• $x$ 是当前任务；
• $R(a,x)$ 是 Agent $a$ 对任务的能力匹配分；
• $C(a)$ 是成本；
• $H(a)$ 是失败或不确定性风险；
• $\lambda,\mu$ 是业务权重。

实际系统不应只把 Agent 名称交给选择模型，而应提供结构化能力描述：

capabilities = {
    "legal_agent": {
        "domains": ["contract", "compliance"],
        "operations": ["review", "risk_summary"],
        "risk_level": "high",
    },
    "finance_agent": {
        "domains": ["invoice", "payment"],
        "operations": ["reconcile", "calculate"],
        "risk_level": "medium",
    },
}

分类、能力匹配、动态选择、回退和评测应该分成不同阶段：

输入
  -> 分类：判断任务类别
  -> 候选生成：筛出具备能力的 Agent
  -> 选择：考虑上下文、负载、成本和风险
  -> 执行
  -> 失败回退
  -> 记录路由决策并评测

反例是直接让模型从所有 Agent 中自由选择。只要描述相似、上下文过长或某个 Agent 名称更醒目，就可能出现错误路由。动态选择应有候选集、置信度阈值和回退路径，而不是把路由完全交给自然语言。

3. Swarm 和 HandoffMessage

Swarm 使用 HandoffMessage 表达 Agent 之间的控制权转移。它更接近显式状态机：

triage -> billing -> human_review

与广播不同，handoff 的重点不是“所有人都知道发生了什么”，而是：

当前任务接下来由谁继续负责。

因此，handoff 消息应携带足够的上下文和关联 ID，但不应默认把所有历史对话复制给所有 Agent。对于敏感数据和大上下文，显式传递任务摘要比无边界共享完整历史更可控。AgentChat 官方将 Swarm 定义为通过 HandoffMessage 表达 Agent 间过渡的 Team 模式。(microsoft.github.io)

---

十一、终止：完成、停止、取消是三个不同概念

终止设计是多 Agent 系统中最容易出错的部分。至少要区分：

1. 正常完成：任务达到成功条件；
2. 优雅停止：允许当前 Agent 完成当前轮，再结束；
3. 立即取消：立刻终止执行，状态可能不一致；
4. 失败结束：因为异常或预算耗尽而停止。

1. AgentChat 的终止条件

AgentChat 的 TerminationCondition 是有状态的可调用条件。它接收自上次检查以来产生的消息，如果满足条件则返回 StopMessage；一旦触发，在再次使用前需要 reset()。终止条件可以通过 | 和 & 组合。(microsoft.github.io)

from autogen_agentchat.conditions import (
    MaxMessageTermination,
    TextMentionTermination,
)

termination = (
    MaxMessageTermination(20)
    | TextMentionTermination("TERMINATE")
)

其逻辑是：

$$stop = count \ge 20 \lor "TERMINATE" \in output$$

如果使用：

termination = (
    MaxMessageTermination(20)
    & TextMentionTermination("TERMINATE")
)

则变为：

$$stop = count \ge 20 \land "TERMINATE" \in output$$

这两个条件的差异很大。OR 适合“任一安全边界触发就停止”；AND 则要求两个条件都成立，可能导致模型已经达到最大消息数却仍不停止。

2. 外部优雅停止

ExternalTermination 用于从应用外部请求 Team 停止：

from autogen_agentchat.conditions import ExternalTermination

external = ExternalTermination()

team = RoundRobinGroupChat(
    [assistant],
    termination_condition=external,
)

# 另一个协程中
external.set()

这种停止不会粗暴打断当前 Agent 的轮次，而是在当前轮结束后停止，使最后一条消息能够正常广播并保留团队状态。(microsoft.github.io)

3. CancellationToken 的立即取消

如果使用取消令牌：

from autogen_core import CancellationToken

token = CancellationToken()

task = asyncio.create_task(
    team.run(
        task="process a long task",
        cancellation_token=token,
    )
)

token.cancel()

调用方会收到 asyncio.CancelledError。这和外部优雅停止不同：立即取消可能使 Team 处于不一致状态，也可能没有重置终止条件。(microsoft.github.io)

因此，停止策略应按故障类型选择：

场景	推荐机制
用户点击停止	ExternalTermination
达到最大轮数	MaxMessageTermination
模型明确输出完成标记	TextMentionTermination 或结构化结果
服务关闭、超时、进程终止	CancellationToken
Core 事件流结束	显式终止消息或 Runtime 停止条件

4. Core 中的终止消息

Core 没有强制规定唯一的全局终止协议。常见做法是定义终止消息：

from dataclasses import dataclass


@dataclass
class Termination:
    correlation_id: str
    reason: str

某个 Agent 发布 Termination 后，由 Runtime 的干预处理器或外围控制器观察它，再调用：

await runtime.stop()

官方提供了基于 InterventionHandler 监听终止消息的示例，但该示例明确适用于 SingleThreadedAgentRuntime。(microsoft.github.io)

这说明一个重要事实：

“某个 Agent 说完成了”不等于整个 Runtime 已经停止。

在并行或事件驱动系统中，可能还有其他消息正在队列中，或者其他消费者尚未完成。真正的终止条件应由流程控制器判断：

$$terminate(r) = success(r) \lor timeout(r) \lor budget\_exceeded(r) \lor cancelled(r) \lor fatal\_error(r)$$

而不是简单检测某个文本是否包含 DONE。

---

十二、状态、重置和恢复

Team 和 Agent 都可能有状态。状态包括：

• 对话历史；
• 工具调用上下文；
• 当前发言者；
• 已完成的子任务；
• 终止条件内部计数；
• 去重集合；
• 外部系统的任务状态。

如果下一次运行是完全无关的新任务，应执行：

await team.reset()

官方说明，reset() 会清理 Team 及其 Agent 的状态，并调用各 Agent 的 on_reset()。如果下一次任务是上一次任务的延续，则可以不重置，直接继续运行。(microsoft.github.io)

这里有一个常见误区：

await team.reset()

不等于：

删除外部数据库中的订单状态
删除已经发送的邮件
撤销已经调用的 API

它主要重置框架内部状态。外部副作用必须由业务补偿事务负责。

如果需要跨进程恢复，则应保存：

1. Team 配置；
2. Agent 状态；
3. 对话或事件日志；
4. 终止条件状态；
5. 未完成任务的关联 ID；
6. 幂等去重信息。

只保存最后一条自然语言消息通常不足以恢复一个有状态工作流，因为“下一步由谁执行”“哪些消费者已经完成”“该消息是否已产生副作用”都可能丢失。

---

十三、错误处理：Handler 异常、无法处理和外部失败

Core 的消息 Handler 应明确区分几类错误。

1. 消息类型不匹配

如果 Agent 无法处理某条消息，应抛出 CantHandleException，而不是静默忽略。这样 Runtime 或上层控制器可以区分：

没有对应能力

和：

已经匹配到能力，但执行失败

官方建议从 RoutedAgent 开始，通过 @message_handler 按消息类型路由；如果消息无法处理，可以抛出 CantHandleException。(microsoft.github.io)

2. 业务执行失败

例如模型调用、数据库请求或工具执行失败：

try:
    result = await call_external_service()
except TimeoutError:
    await self.publish_message(
        RetryableFailure(...),
        topic_id,
    )
except PermissionError:
    await self.publish_message(
        FatalFailure(...),
        topic_id,
    )

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

• 可重试错误：超时、临时网络失败、服务限流；
• 不可重试错误：参数错误、权限错误、协议版本错误；
• 需要人工介入：金额异常、合规决策、工具副作用不确定。

3. 重试与幂等

重试逻辑必须和消息 ID 配合：

if message.message_id in self.processed_ids:
    return

await perform_effect(message)
self.processed_ids.add(message.message_id)

但如果 perform_effect() 成功后进程在写入 processed_ids 前崩溃，仍然存在重复执行窗口。因此，真正的幂等记录通常需要和业务副作用放在同一个事务边界，或使用外部系统提供的幂等键。

---

十四、扩展：不是继承一个类就完成了

AutoGen 扩展的本质是：

\text{External System} \leftrightarrow \text{AutoGen Interface}
]

例如模型扩展需要把模型服务的请求、流式响应、工具调用、错误和使用量转换为 AutoGen 能理解的接口。

一个合格的扩展通常需要考虑：

• 对应的 Core 或 AgentChat 接口；
• 输入输出类型；
• 异步调用和取消；
• 超时与重试；
• 资源关闭；
• 配置序列化；
• 版本兼容；
• 日志和可观测性；
• 外部副作用的幂等。

官方建议扩展尽可能实现 autogen_core 提供的通用接口，并在 pyproject.toml 中对 AutoGen 版本设置合理约束。例如：

[project]
name = "autogen-example-extension"
version = "0.1.0"
dependencies = [
    "autogen-core>=0.4,<0.5"
]

官方还建议使用类型标注，并鼓励独立扩展以 autogen- 作为包名前缀；第一方支持的扩展放在 autogen-ext，其他项目可以作为独立包发布。(microsoft.github.io)

扩展版本约束尤其重要。Runtime、消息模型和组件配置一旦发生不兼容，问题可能不是导入失败，而是运行到特定消息才出现反序列化或路由错误。因此：

扩展版本
  -> 依赖的 AutoGen 版本
  -> 消息协议版本
  -> 外部服务 API 版本

应当分别管理，不能只锁定一个 Python 包版本。

---

十五、组件配置和可复现性

AgentChat 的 Agent、Team、模型客户端和终止条件可以被序列化为配置，之后再加载。例如，Team 配置可以包含：

• 组件 Provider；
• 组件类型；
• 版本；
• 描述；
• 具体配置参数。

官方文档展示了通过 dump_component() 导出组件配置，再通过 load_component() 从 JSON 加载 Team 的方式。(microsoft.github.io)

这对生产系统有两个价值。

第一，运行配置可以进入版本控制，而不是完全隐藏在 Python 代码构造函数中。

第二，评测可以固定：

模型配置
Agent 指令
工具集合
Team 调度器
终止条件

否则，同一任务在不同日期运行时，可能无法判断结果变化来自模型、提示词、路由器还是终止逻辑。

对于动态路由系统，评测不应只看最终答案，还应记录：

{
  "correlation_id": "r-001",
  "predicted_category": "contract",
  "candidate_agents": ["legal_agent", "finance_agent"],
  "selected_agent": "legal_agent",
  "fallback": false,
  "termination_reason": "completed",
  "message_count": 6
}

这样才能分别评估：

• 分类准确率；
• 能力匹配准确率；
• 动态选择成功率；
• 回退触发率；
• 无效消息比例；
• 平均完成轮数；
• 终止条件是否过早或过晚。

---

十六、常见误解和诊断路径

误解一：多个 Agent 就一定比一个 Agent 好

Team 适合需要协作和不同专业能力的复杂任务，但它需要更多脚手架、调度和终止控制。简单任务如果单个 Agent 已经能够稳定完成，增加 Team 只会增加上下文、调用次数和故障面。官方也建议先优化单 Agent，再在确实需要协作时引入 Team。(microsoft.github.io)

误解二：发布消息可以等待返回值

广播是单向发布订阅，不是 RPC。需要响应时使用直接消息，或者定义带 correlation_id 的结果事件。

误解三：stop() 会取消所有任务

SingleThreadedAgentRuntime.stop() 不会取消正在执行的 Handler。需要立即中断时，应使用取消机制；需要保持状态一致时，应优先使用优雅停止。

误解四：终止条件是无状态函数

AgentChat 的终止条件是有状态的，触发后需要 reset() 才能再次使用。如果复用一个已经触发过的终止条件启动下一轮，可能出现 Team 一开始就停止的现象。

误解五：同一个 Agent 类型等于同一个 Agent 实例

Agent 类型只是工厂注册名。真正的实例由 (AgentType, AgentKey) 确定。多租户系统如果错误地把所有请求都使用 "default" 作为 Agent Key，就会把不同会话状态合并到同一个实例。

误解六：Runtime 提供 exactly-once

Runtime 可以投递消息，但不能自动保证外部副作用只发生一次。支付、发货、发邮件等操作必须通过幂等键、事务、去重表或补偿机制保证。

---

十七、选择 AutoGen 层次的判断方法

可以按以下问题选择 API 层次：

选择 AgentChat，如果：

• 目标是快速验证多 Agent 协作；
• Team 模式比较标准；
• 主要工作是模型、工具和对话上下文；
• 不需要自己管理底层事件路由。

选择 Core，如果：

• Agent 是长期运行的有状态服务；
• 需要 Topic、Subscription 和事件驱动流程；
• 需要自定义消息协议和路由；
• 需要在单进程和分布式 Runtime 之间切换；
• 需要对生命周期、故障和并发拥有更细控制。

编写 Extension，如果：

• 需要接入新的模型供应商；
• 需要接入企业内部工具或 MCP 服务；
• 需要新的代码执行环境；
• 需要自定义 Runtime、传输层或持久化机制。

最终可以把 AutoGen 的职责边界概括为：

Agent       = 谁处理状态和消息
Message     = 传递什么事实
Topic       = 事件属于哪个范围
Subscription= 谁对该事件感兴趣
Team        = 多个 Agent 如何协作
Termination = 何时结束协作
Runtime     = 如何寻址、创建、投递和运行
Extension   = 如何连接外部系统

当这几个边界被明确后，AutoGen 就不再只是一个“多 Agent 聊天框架”，而是一套可以从单步模型调用逐步扩展到事件驱动、动态路由、分布式执行和可评测工作流的 Agent 工程基础设施。

---

系列导航与关联阅读

• 系列入口：Agent 工程完整路线：从运行循环、记忆与协议到安全、评测和生产交付
• 上一篇：PydanticAI：类型化依赖、工具、结构化输出、图和测试
• 下一篇：CrewAI：Agent、Task、Crew、Flow、状态和生产边界
• 延伸：多 Agent 路由：分类、能力匹配、动态选择、回退和评测
• 延伸：事件驱动 Agent：Topic、消费者、关联 ID、顺序和最终一致性

官方资料

• PydanticAI Documentation
• Microsoft AutoGen Documentation

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