多智能体系统测试实战：从零构建Agent测试方法论

1次阅读

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

多智能体系统 (MAS) 与单体 Agent 测试存在显著差异，主要体现在以下几个方面：

非确定性行为：由于多个 Agent 并行执行和交互，系统的整体行为难以完全预测
涌现现象(Emergent Phenomena)：简单个体交互可能产生复杂的全局行为模式
分布式状态：系统状态分散在各个 Agent 中，难以获取全局一致性视图
时序敏感性：消息传递延迟和顺序会影响系统行为

验证单个 Agent 的核心功能，重点关注：

决策逻辑
本地状态管理
消息处理能力

import unittest
from agent import MyAgent

class TestSingleAgent(unittest.TestCase):
    def test_decision_making(self):
        """验证 Agent 在特定输入下的决策输出"""
        agent = MyAgent(agent_id="test_agent")
        test_state = {"sensor_data": 0.75}
        expected_action = "move_forward"
        self.assertEqual(agent.decide(test_state), expected_action)

验证多个 Agent 的交互行为，重点关注：

消息协议一致性
协作流程正确性
资源竞争处理

模拟异常场景，验证系统容错能力：

网络分区
Agent 崩溃
消息丢失

from typing import Dict, Any
from queue import Queue

class MockAgent:
    def __init__(self, agent_id: str):
        self.id = agent_id
        self.inbox = Queue()  # 模拟消息队列
        self.behavior = {}    # 可配置的响应行为

    def configure_behavior(self, msg_type: str, response):
        """预定义对特定消息类型的响应"""
        self.behavior[msg_type] = response

    def process_message(self, msg: Dict[str, Any]):
        """处理接收到的消息"""
        if msg["type"] in self.behavior:
            return self.behavior[msg["type"]](msg)
        return None

class MessageBus:
    def __init__(self):
        self.agents = {}  # agent_id -> agent 实例
        self.message_log = []

    def register_agent(self, agent):
        self.agents[agent.id] = agent

    def send(self, sender_id: str, receiver_id: str, message: dict):
        """模拟消息传递，可注入延迟或丢失"""
        if receiver_id in self.agents:
            self.agents[receiver_id].inbox.put({
                "sender": sender_id,
                **message
            })
            self.message_log.append({"timestamp": time.time(),
                "sender": sender_id,
                "receiver": receiver_id,
                "message": message
            })

控制随机种子：确保每次测试运行的随机行为一致
场景录制回放：捕获典型交互序列作为测试用例
状态快照：在关键点保存系统状态用于断言

import random

def test_deterministic_behavior():
    # 固定随机种子
    random.seed(42)

    # 初始化测试环境
    bus = MessageBus()
    agent1 = MockAgent("agent1")
    agent2 = MockAgent("agent2")

    # 预定义行为
    agent1.configure_behavior("request", lambda _: {"type": "response", "data": "OK"})

    # 执行测试
    bus.register_agent(agent1)
    bus.register_agent(agent2)
    bus.send("agent2", "agent1", {"type": "request"})

    # 验证
    assert not agent1.inbox.empty()
    response = agent1.process_message(agent1.inbox.get())
    assert response["data"] == "OK"

使用 Lamport 时钟可视化消息时序：

sequenceDiagram
    participant A as Agent1
    participant B as Agent2
    participant C as Agent3

    A->>B: 消息 1 (逻辑时钟:1)
    B->>C: 消息 2 (逻辑时钟:2)
    C->>A: 消息 3 (逻辑时钟:3)
    A->>B: 消息 4 (逻辑时钟:4)