当我们开发的智能体应用逐渐复杂时，单进程模式往往会遇到性能瓶颈和可扩展性问题。今天要聊的 AutoGen 分布式智能体运行时，就像为智能体系统搭建了一条 "跨进程高速公路"，让多个智能体能够在不同进程间高效通信。不过需要提前说明：这是一项实验性功能，API 可能会有变动，但正是这种前沿特性，能让我们提前掌握分布式 AI 系统的构建方法。

一、分布式运行时的核心架构：主机与工作节点的协同机制

1. 架构组成与角色分工

分布式智能体运行时采用典型的主从架构，由两个核心组件构成：

  • 主机服务（Host Service）：相当于智能体世界的 "交通枢纽"
    • 维护所有工作节点的连接状态
    • 负责消息路由（确保消息送达正确的智能体）
    • 管理直接消息的会话（类似 RPC 调用的上下文）
  • 工作节点运行时（Worker Runtime）：智能体的 "执行引擎"
    • 运行具体的智能体代码
    • 向主机服务注册所支持的智能体
    • 处理消息并执行相应的业务逻辑

2. 跨进程通信的关键流程

  1. 工作节点启动后向主机服务注册
  2. 主机服务维护一张 "智能体 - 工作节点" 映射表
  3. 当需要发送消息时：
    • 主机服务根据映射表找到目标工作节点
    • 通过 gRPC 协议跨进程传递消息
    • 工作节点处理消息并返回结果

类比理解：主机服务就像快递中转站，工作节点是各地的快递点，智能体则是等待接收包裹的用户，中转站根据地址簿（映射表）把包裹送到正确的快递点。

二、实战准备：环境搭建与依赖安装

1. 安装必要的扩展包

分布式运行时需要额外的依赖，通过以下命令安装：

bash

pip install "autogen-ext[grpc]"

注意：这里的中括号语法是 pip 的特性，用于安装指定的扩展子集，grpc表示我们需要 gRPC 相关的组件。

2. 启动主机服务（以默认端口为例）

python

from autogen_ext.runtimes.grpc import GrpcWorkerAgentRuntimeHost

# 创建主机服务实例（监听本地50051端口）
host = GrpcWorkerAgentRuntimeHost(address="localhost:50051")

# 在后台启动服务（非阻塞方式）
host.start()
print("主机服务已启动，等待工作节点连接...")

关键细节：

  • address参数格式为 "主机名：端口"，生产环境中应使用可访问的 IP
  • start()方法会在后台线程中运行服务，不阻塞主线程
  • 默认使用 gRPC 协议，性能优于 HTTP，适合生产环境

三、智能体开发：从消息定义到处理逻辑

1. 定义跨进程传递的消息类型

python

from dataclasses import dataclass
from autogen_core import DefaultTopicId, MessageContext, RoutedAgent, default_subscription, message_handler

@dataclass
class MyMessage:
    """自定义消息类型，需确保跨进程可序列化"""
    content: str

注意事项：

  • 使用 Python 的 dataclass 装饰器定义消息，确保结构清晰
  • 消息类型需能被 protobuf 序列化（跨语言场景尤其重要）
  • 避免在消息中包含不可序列化的对象（如文件句柄、网络连接）

2. 实现智能体类及其消息处理逻辑

python

@default_subscription  # 订阅默认主题，接收所有消息
class MyAgent(RoutedAgent):
    def __init__(self, name: str) -> None:
        super().__init__("My agent")
        self._name = name
        self._counter = 0  # 消息计数器
    
    @message_handler  # 注册消息处理函数
    async def my_message_handler(self, message: MyMessage, ctx: MessageContext) -> None:
        self._counter += 1
        # 达到5条消息后停止发送
        if self._counter > 5:
            return
        
        # 构造响应消息并打印
        content = f"{self._name}: Hello x {self._counter}"
        print(content)
        
        # 发布新消息到默认主题
        await self.publish_message(
            MyMessage(content=content),
            DefaultTopicId()
        )

代码解析：

  • @default_subscription装饰器使智能体订阅默认主题，能接收所有发布的消息
  • message_handler装饰器将方法注册为消息处理器
  • publish_message方法用于跨进程发送消息，第一个参数是消息对象，第二个是主题 ID
  • 计数器逻辑演示了智能体的状态维护能力

四、实战部署：分进程运行的完整流程

1. 主机服务的独立部署（host_service.py）

python

# 主机服务必须单独运行在一个进程中
from autogen_ext.runtimes.grpc import GrpcWorkerAgentRuntimeHost
import asyncio

async def start_host():
    # 创建主机服务（监听本地50051端口，生产环境需改为服务器IP）
    host = GrpcWorkerAgentRuntimeHost(address="0.0.0.0:50051")
    host.start()
    print(f"主机服务已启动，监听端口50051")
    
    # 保持主机运行（生产环境可使用stop_when_signal）
    try:
        while True:
            await asyncio.sleep(1)
    except KeyboardInterrupt:
        await host.stop()
        print("主机服务已停止")

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

部署方式：

bash

# 在第一个终端运行主机服务
python host_service.py

关键输出：

plaintext

主机服务已启动，监听端口50051

2. 工作节点 1 的独立部署（worker1.py）

python

# 工作节点1必须在另一个进程中运行
from autogen_ext.runtimes.grpc import GrpcWorkerAgentRuntime
from dataclasses import dataclass
from autogen_core import DefaultTopicId, RoutedAgent, default_subscription, message_handler
import asyncio

@dataclass
class MyMessage:
    content: str

@default_subscription
class MyAgent(RoutedAgent):
    def __init__(self, name: str) -> None:
        super().__init__("worker1_agent")
        self._name = name
        self._counter = 0
    
    @message_handler
    async def process_message(self, msg: MyMessage, ctx):
        self._counter += 1
        if self._counter > 5:
            return
        content = f"[WORKER1] {self._name}: Hello x {self._counter}"
        print(content)
        await self.publish_message(MyMessage(content=content), DefaultTopicId())

async def start_worker1():
    # 连接到主机服务（IP需与主机实际地址一致）
    worker = GrpcWorkerAgentRuntime(host_address="localhost:50051")
    await worker.start()
    
    # 注册智能体到工作节点1
    await MyAgent.register(
        worker,
        "worker1_agent",  # 智能体在主机中的唯一标识
        lambda: MyAgent("Node1")  # 工厂函数创建实例
    )
    print("工作节点1已启动，智能体注册完成")

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

部署方式：

bash

# 在第二个终端运行工作节点1
python worker1.py

关键输出：

plaintext

工作节点1已启动，智能体注册完成

3. 工作节点 2 的独立部署（worker2.py）

python

# 工作节点2在第三个进程中运行（可部署在另一台机器）
from autogen_ext.runtimes.grpc import GrpcWorkerAgentRuntime
from dataclasses import dataclass
from autogen_core import DefaultTopicId, RoutedAgent, default_subscription, message_handler
import asyncio

@dataclass
class MyMessage:
    content: str

@default_subscription
class MyAgent(RoutedAgent):
    def __init__(self, name: str) -> None:
        super().__init__("worker2_agent")
        self._name = name
        self._counter = 0
    
    @message_handler
    async def process_message(self, msg: MyMessage, ctx):
        self._counter += 1
        if self._counter > 5:
            return
        content = f"[WORKER2] {self._name}: Hello x {self._counter}"
        print(content)
        await self.publish_message(MyMessage(content=content), DefaultTopicId())

async def start_worker2():
    # 连接到主机服务（若部署在另一台机器，此处改为主机IP）
    worker = GrpcWorkerAgentRuntime(host_address="192.168.1.100:50051")
    await worker.start()
    
    # 注册智能体到工作节点2
    await MyAgent.register(
        worker,
        "worker2_agent",
        lambda: MyAgent("Node2")
    )
    print("工作节点2已启动，智能体注册完成")
    
    # 从工作节点2发送初始消息（触发分布式通信）
    await worker.publish_message(
        MyMessage(content="分布式启动测试"),
        DefaultTopicId()
    )

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

部署方式：

bash

# 在第三个终端或另一台机器运行工作节点2
python worker2.py

关键输出：

plaintext

工作节点2已启动，智能体注册完成
[WORKER2] Node2: Hello x 1

五、分布式通信的实时观测：跨进程消息流转演示

1. 多进程交互的控制台输出

当三个进程全部启动后，你将看到：

  • 工作节点 2 终端：首先输出自己的第一条消息
  • 工作节点 1 终端：随后接收到消息并输出自己的响应
  • 两个工作节点交替输出，直到各发送 5 条消息

plaintext

# 工作节点1终端输出
[WORKER1] Node1: Hello x 1
[WORKER1] Node1: Hello x 2
...
[WORKER1] Node1: Hello x 5

# 工作节点2终端输出
[WORKER2] Node2: Hello x 1
[WORKER2] Node2: Hello x 2
...
[WORKER2] Node2: Hello x 5

2. 分布式通信的核心流程解析

  1. 工作节点 2 发送消息：

    python

    await worker2.publish_message(MyMessage(...), DefaultTopicId())

  2. 消息路由过程：

    • 工作节点 2 将消息发送到主机服务
    • 主机服务查看路由表（所有智能体订阅默认主题）
    • 主机将消息广播到所有已注册的工作节点

  3. 消息处理过程：

    • 工作节点 1 和工作节点 2 各自接收消息
    • 每个智能体独立处理消息并增加计数器
    • 处理完成后，智能体将新消息发回主机服务
    • 主机再次路由消息，形成分布式消息循环

六、进阶部署：跨主机场景的实战配置

1. 主机服务部署在服务器 A

python

# 服务器A的host_service.py（修改地址为0.0.0.0）
host = GrpcWorkerAgentRuntimeHost(address="0.0.0.0:50051")

启动命令：

bash

# 在服务器A运行
nohup python host_service.py &

2. 工作节点部署在服务器 B

python

# 服务器B的worker.py（修改host_address为主机IP）
worker = GrpcWorkerAgentRuntime(host_address="192.168.1.100:50051")

启动命令：

bash

# 在服务器B运行
python worker.py

3. 跨主机通信的关键配置

  • 防火墙设置：确保服务器 A 的 50051 端口可被访问
  • 网络连通性：两台服务器需在同一局域网或公网可达
  • IP 地址配置：
    • 主机服务使用0.0.0.0监听所有网络接口
    • 工作节点使用主机的实际 IP 地址

七、分布式系统的监控与管理

1. 查看当前连接状态

python

# 在主机服务中添加监控代码
print(f"当前连接的工作节点数: {len(host.worker_connections)}")
for worker_id, conn in host.worker_connections.items():
    print(f"工作节点 {worker_id}: 状态={conn.status}, 智能体数={len(conn.agents)}")

2. 动态添加工作节点

python

# 在运行时动态添加新工作节点（无需重启主机）
new_worker = GrpcWorkerAgentRuntime(host_address="localhost:50051")
await new_worker.start()
await MyAgent.register(new_worker, "worker3_agent", lambda: MyAgent("Node3"))

3. 故障恢复机制

python

# 工作节点断线重连（AutoGen内置支持）
worker = GrpcWorkerAgentRuntime(host_address="localhost:50051", max_reconnect_attempts=10)

自动重连策略：

  • 首次断线后等待 1 秒重连
  • 每次重连失败后等待时间翻倍
  • 达到最大尝试次数后停止重连

八、跨语言支持：Protobuf 在分布式场景中的关键作用

当需要跨语言部署智能体时（如 Python 与 Java 混合架构），必须遵循以下原则：

  1. 消息类型共享：所有跨智能体消息必须使用共享的 protobuf 模式
  2. 序列化一致性：不同语言的实现需保证消息序列化结果一致
  3. 类型映射规范：建立各语言与 protobuf 的类型映射表（如 Python 的 dataclass 对应 protobuf 的 message）

示例 protobuf 定义（与 MyMessage 对应）：

protobuf

syntax = "proto3";

package autogen.messages;

message MyMessage {
    string content = 1;
}

跨语言实现步骤：

  1. 使用 protoc 编译器生成各语言的消息类
  2. 在非 Python 环境中，实现与 RoutedAgent 等价的消息处理逻辑
  3. 确保消息发布 / 订阅的主题 ID 一致

结语：从单机到分布式的思维转变

通过分进程、跨主机的实战部署，我们终于看清了 AutoGen 分布式运行时的真实面貌 —— 它不是代码逻辑的简单拆分，而是物理进程的隔离与协同。这种部署模式为智能体系统带来了真正的可扩展性，但也要求我们掌握分布式系统的调试与管理技巧。

如果本文对你有帮助，别忘了点赞收藏，关注我，一起探索更高效的开发方式～

# 分布式 # 人工智能
Logo
深开鸿 技术专区

更多推荐

cover

【计算机网路】计算机网络概述：从 ARPAnet 到万物互联

avatar 深开鸿 技术专区
cover

Flutter 与开源鸿蒙（OpenHarmony）实战：构建下一代跨平台应用的完整指南

avatar 深开鸿 技术专区

Copier在分布式系统中的应用：确保跨节点数据一致性的复制方案

在当今的分布式系统架构中，数据一致性是确保系统可靠性的关键挑战。Copier作为一个高效的Golang结构体复制库，为分布式环境下的数据复制提供了简单而强大的解决方案。这个开源工具能够帮助开发者在不同节点间实现结构体数据的无缝复制，确保数据在分布式网络中的一致性传输。## 什么是Copier及其核心价值Copier是一个专门为Golang设计的结构体复制工具，它能够智能地处理不同类型结构体

avatar 深开鸿 技术专区