AG2多智能体框架入门指南：从零搭建你的第一个智能体系统

共计 2298 个字符，预计需要花费 6 分钟才能阅读完成。

为什么需要多智能体系统？

传统单智能体在复杂场景中常遇到三大天花板：

• 任务分解困难：像物流调度这种需要动态路径规划的任务，单智能体难以同时处理数百个订单的实时状态
• 资源瓶颈：所有决策集中在单个节点，CPU/ 内存成为性能瓶颈
• 容错率低：主节点崩溃直接导致系统瘫痪

去年我们电商大促时，原基于 RLlib 的推荐系统就因单智能体过载导致 30% 请求超时。改用 AG2 后，通过将用户画像分析、商品匹配、排序策略拆解到不同智能体，吞吐量提升了 4 倍。

框架选型对比

AG2 的 stigmergy 通信机制是其核心优势——智能体通过环境中的数字痕迹（如公共黑板）间接通信，避免直接耦合。

手把手实现第一个智能体

1. 环境准备

pip install ag2==0.3.2  # 推荐使用虚拟环境

2. 定义智能体

from ag2 import Agent, agent

@agent(role="bidder", version="1.0")
class AuctionAgent(Agent):
    def __init__(self, agent_id, budget):
        super().__init__(agent_id)
        self.budget = budget  # 初始预算
        self.current_bid = 0  # 当前出价

    def on_message(self, sender, message):
        """处理拍卖消息的核心逻辑"""
        if message["type"] == "bid_update":
            return self._handle_bid(message["price"])

    def _handle_bid(self, new_price):
        """BDI 模型决策逻辑"""
        # Belief: 当前市场价格
        # Desire: 以合理价格中标
        # Intention: 动态调整出价策略
        if new_price < self.budget * 0.8:
            self.current_bid = new_price + random.randint(10,50)
            return {"action": "raise", "amount": self.current_bid}

3. 实现消息路由

# 创建主题交换机
from ag2 import MessageBroker
broker = MessageBroker()
broker.create_topic("auction_room")

# 订阅消息
auction_agent.subscribe("auction_room", 
                       callback=lambda msg: print(f"收到报价: {msg}"),
                       error_handler=lambda e: logging.error(f"消息处理失败: {e}"))

# 发布消息
broker.publish(
    topic="auction_room",
    message={"type": "bid_update", "price": 120},
    timeout=5.0  # 必须设置超时!
)

避坑指南

1. 死锁问题

现象：智能体 A 等待 B 的响应，同时 B 也在等待 A
解决：所有阻塞调用必须设置超时参数：

response = agent.query("other_agent", 
                      query={"request": "data"},
                      timeout=3.0)  # 关键参数！

2. 心跳丢失

现象：网络波动导致智能体被错误标记为离线
解决：调整心跳检测参数：

@agent(heartbeat_interval=2,  # 心跳间隔(秒)
       max_missed=3)         # 最大允许丢失次数

3. 消息堆积

现象：高并发下消息队列爆满
解决：实现背压控制：

broker = MessageBroker(
    max_queue_size=1000,  # 队列最大长度
    overflow_policy="drop_oldest"  # 满时丢弃最旧消息
)

性能实测数据

测试环境：AWS c5.2xlarge 10 节点集群

当节点超过 300 时建议采用分片策略：

# 按智能体 ID 哈希分片
shard_id = hash(agent_id) % 3  
broker.create_topic(f"auction_room_{shard_id}")

竞拍智能体 Demo 扩展

建议读者尝试以下扩展：

1. 增加「虚假报价」检测逻辑
2. 实现组合拍卖（combinatorial auction）
3. 添加强化学习模块动态调整出价策略

启动示例：

if __name__ == "__main__":
    # 创建 5 个竞拍者
    agents = [AuctionAgent(f"bidder_{i}", budget=1000) 
              for i in range(5)]

    # 启动模拟拍卖
    for round in range(10):
        broker.publish("auction_room", 
                      {"type": "start_round", "item": f"lot_{round}"})

实际部署时，我们发现智能体的初始化顺序很关键——建议先启动消息中间件，再逐个启动智能体。遇到连接问题时，先用 ag2-cli healthcheck 工具诊断网络状态。

多智能体开发就像组建足球队，既要让每个球员（智能体）各司其职，又要确保战术（通信协议）配合默契。AG2 提供的工具链让这个过程变得 surprisingly enjoyable——至少比我们当年用 ZeroMQ 手撸通信协议要幸福多了。