测试Agent+Skill架构实战：如何构建高可扩展的自动化测试系统

2次阅读

没有评论

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

在实际工作中，我们经常会遇到这样的问题：

脚本复用率低：每个新项目都要重写大量相似代码，修改一个参数就要改动多处
多环境适配成本高：同一套测试用例需要针对不同环境（如 Dev/QA/Prod）维护多份配置
用例维护困难：随着业务复杂化，线性脚本变得臃肿脆弱，一个小改动就可能引发连锁报错

通过将测试能力拆分为独立的 Skill（技能单元），由智能 Agent（代理）统一调度，实现：

模块化：每个 Skill 只关注单一测试能力（如登录、下单）
可编排：Agent 通过 Orchestration（编排）动态组合 Skill 形成完整测试流
松耦合：Skill 之间通过标准契约交互，不关心具体实现

任务调度：根据测试计划调用对应 Skill，处理执行顺序和依赖关系
状态管理：维护测试上下文（Context）和共享数据（如 token、全局变量）
通信协议：标准化 Skill 间通信方式（推荐 HTTP/RPC+Protobuf）

输入输出契约：明确定义入参和返回值结构，例如：

class LoginSkill:
    input_schema = {'username': str, 'password': str}
    output_schema = {'token': str, 'expire_in': int}

幂等性保证：相同输入必须产生相同输出，避免随机性影响测试结果
版本兼容：通过语义化版本（SemVer）管理变更，旧版 Skill 需保持可用

from abc import ABC, abstractmethod
from typing import Dict, Any
import hashlib

class BaseSkill(ABC):
    """技能基类（抽象类）"""
    version = "1.0.0"
    description = ""

    @classmethod
    def fingerprint(cls) -> str:
        """生成技能唯一标识"""
        meta = f"{cls.__name__}-{cls.version}-{cls.description}"
        return hashlib.md5(meta.encode()).hexdigest()

    @abstractmethod
    async def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
        """执行入口"""
        pass

import importlib
from pathlib import Path

class SkillLoader:
    @staticmethod
    def load_from_dir(skill_dir: str) -> Dict[str, BaseSkill]:
        """从目录加载所有技能"""
        skills = {}
        for py_file in Path(skill_dir).glob("*.py"):
            module_name = py_file.stem
            try:
                module = importlib.import_module(f"skills.{module_name}")
                for attr in dir(module):
                    obj = getattr(module, attr)
                    if isinstance(obj, type) and issubclass(obj, BaseSkill) and obj != BaseSkill:
                        skills[obj.fingerprint()] = obj()
            except Exception as e:
                print(f"加载 {module_name} 失败: {e}")
        return skills

import asyncio
from concurrent.futures import ThreadPoolExecutor

class TestAgent:
    def __init__(self, max_workers=4):
        self.skill_pool = ThreadPoolExecutor(max_workers=max_workers)

    async def run_pipeline(self, skills: List[BaseSkill], context: dict):
        """异步执行技能链"""
        for skill in skills:
            try:
                # 将阻塞 IO 操作放入线程池执行
                result = await asyncio.get_event_loop().run_in_executor(
                    self.skill_pool,
                    lambda: skill.execute(context)
                )
                context.update(result)
            except Exception as e:
                print(f"技能 {skill.__class__.__name__} 执行失败: {e}")
                break
        return context

冷启动时预加载高频 Skill 到内存
对耗时初始化（如浏览器驱动）提前执行

from functools import wraps
import signal

class TimeoutException(Exception): pass

def timeout(seconds=10):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            def _handle_timeout(signum, frame):
                raise TimeoutException(f"操作超时({seconds}s)")

            signal.signal(signal.SIGALRM, _handle_timeout)
            signal.alarm(seconds)
            try:
                result = func(*args, **kwargs)
            finally:
                signal.alarm(0)
            return result
        return wrapper
    return decorator