让AI替你写测试脚本：探索接口自动化测试智能体

本文介绍了使用Google-ADK框架构建智能体实现接口测试自动化的方法。通过安装Python3.11及以上版本和requests库，利用ADK框架的核心组件（如LlmAgent、SequentialAgent等）创建具备决策能力的测试智能体。文章详细展示了如何配置智能体角色、工具和指令，并提供了完整的执行流程代码示例。此外，还介绍了构建接口测试工具的关键实现，包括会话池配置、重试策略和文件处理等

可遇_不可求

383人浏览 · 2025-12-02 07:01:30

可遇_不可求 · 2025-12-02 07:01:30 发布

介绍

本文探索如何使用Google-ADK框架构建的智能体能够通过自然语言交互实现接口测试的自动化编排与执行。本文将详细探讨如何利用Google-ADK（Agent Development Kit）框架，结合requests库构建一个具备智能决策能力的接口自动化测试智能体。

安装

使用python3.11以上版本
```
pip install google-adk
```
安装requests
```
pip install requests
```

Google-ADK介绍

Agent Development Kit（ADK）是一个灵活的模块化框架，它将软件开发原则应用于AI代理创建。它旨在简化从简单任务到复杂系统的代理工作流的构建、部署和编排。

核心架构组件

ADK基于几个核心抽象构建：

Agent（代理）：定义身份、指令和工具的蓝图
Tool（工具）：代理可调用的能力，如Python函数
Runner（运行器）：编排"推理-行动"循环的引擎
Session（会话）：单个连续对话的状态
Memory（记忆）：跨会话的长期回忆
Artifact Service（工件服务）：管理文件等非文本数据

ADK中的Agent

LlmAgent（LLM代理）

LlmAgent是ADK的核心组件，作为应用的"思考"部分，利用大语言模型进行推理、理解自然语言、制定决策、生成响应和与工具交互。它的行为是非确定性的，使用LLM解释指令和上下文，动态决定如何进行、使用哪些工具或将控制权转移给其他代理。

SequentialAgent（顺序代理）

SequentialAgent是工作流代理，按照指定顺序执行其子Agent 。当需要执行固定、严格顺序的任务时使用，确保代码编写、审查、重构等步骤按 dependable 顺序执行。

ParallelAgent（并行代理）

ParallelAgent是工作流代理，并发执行其子Agent，显著加速可独立执行的任务。适用于多源数据检索或重计算等场景，其中并行化可带来显著的性能提升。

LoopAgent（循环代理）

LoopAgent是工作流代理，重复执行其子Agent直到满足特定终止条件。循环在达到可选的max_iterations时停止，或者如果任何子Agent返回带有escalate=True的事件时停止。

Custom Agent（自定义代理）

自定义代理通过直接继承BaseAgent并实现自己的控制流来提供终极灵活性。当需要实现独特的编排逻辑、特定的控制流或标准类型未涵盖的专门集成时使用。

创建智能体

使用google-adk创建一个LlmAgent。

下面我们将展示如何使用google-adk创建一个LlmAgent，并为其配置相应的工具和指令。

from google.adk.agents import Agent
from google.adk.tools import google_search
from google.adk.agents import LlmAgent
from google.adk.models.lite_llm import LiteLlm

root_agent = Agent(
    name="api_test_agent",
    model="gemini-2.5-flash", 
    instruction="You are a helpful assistant. Answer user questions using Google Search when needed.",
    description="An assistant that can search the web.",
    tools=[google_search]
)
# 或者使用LlmAgent创建LLM代理(Qwen、DeepSeek等)的智能体
root_agent = LlmAgent(
    name="api_test_agent",
    model=LiteLlm(
        model="qwen-plus-latest", 
        api_key="your api keys", 
        api_base="https://dashscope.aliyuncs.com/compatible-mode/v1"
    ), 
    instruction="你是一位高级接口自动化测试工程师，你的工作是根据用户输入的接口文档，分析创建接口测试脚本。并执行接口测试",
    description="提示词",
    tools=[api_test_tools]
)

在上述代码中，我们首先尝试创建了一个普通的Agent，为其指定了名称、使用的模型、指令、描述以及可调用的工具。然后，我们展示了如何使用LlmAgent创建一个更复杂的智能体。这个智能体被赋予了高级接口自动化测试工程师的角色，它能够根据用户输入的接口文档，自动分析并创建接口测试脚本，然后执行测试。通过配置LiteLlm模型，我们可以指定使用特定的大语言模型，并提供相应的API密钥和基础URL。同时，我们还为智能体配置了api_test_tools工具，使其能够实际执行接口测试。

执行智能体

创建好智能体后，我们需要一个运行器来执行它。下面是一个完整的执行智能体的示例代码：

from google.adk.runner import Runner
from google.adk.utils.context_utils import Aclosing
from google.genai import types

async def run_agent():  
    try:
        runner = Runner(agent=root_agent, app_name="test_app")  
        session = await runner.session_service.create_session(  
            app_name="test_app", user_id="test_user"  
        )  

        # 运行代理  
        async with Aclosing(
            runner.run_async(
                user_id="test_user",
                session_id=session.id,
                new_message=types.Content(role="user", parts=[types.Part(text="Hello!")]) 
            )
        ) as agen:
            async for event in agen:
                sse_event = event.model_dump_json(
                    exclude_node=True, by_alias=True
                )
                yield f"data: {sse_event}\n\n"
            except Exception as e:
                yield f"data: {{'error': '{str(e)}'}}\n\n"

在这个示例中，我们首先创建了一个Runner实例，并指定了要执行的智能体和应用名称。然后，我们创建了一个会话，为智能体与用户的交互提供了上下文。接下来，我们使用run_async方法模拟用户发送消息，启动智能体的执行过程。在执行过程中，我们异步迭代处理代理返回的事件，并将事件以SSE（Server-Sent Events）格式输出。如果在执行过程中出现异常，我们会捕获异常并将错误信息输出。

构建接口工具

为了实现接口测试的自动化，我们需要构建一系列接口工具，这些工具将被智能体调用以完成实际的测试任务。下面是一个完整的接口工具实现：

import os
import mimetypes
import json
from functools import lru_cache
from typing import Dict, Any, Optional, Tuple
from common.yaml_tool import read_yaml
from common.token_api import get_token
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 配置会话池和重试策略
session = requests.Session()
retry_strategy = Retry(
    total=3,
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["GET", "POST"],
    backoff_factor=1
)
session.mount("https://", HTTPAdapter(max_retries=retry_strategy))
session.mount("http://", HTTPAdapter(max_retries=retry_strategy))

@lru_cache(maxsize=1)
def get_config() -> Dict[str, Any]:
    """缓存配置文件读取结果"""
    return read_yaml('config/config.yaml')

def get_mime_type(filename: str) -> str:
    """自动检测文件MIME类型"""
    mime_type, _ = mimetypes.guess_type(filename)
    return mime_type or 'application/octet-stream'

def safe_file_opener(file_path: str):
    """安全打开文件（上下文管理器）"""
    return open(file_path, 'rb')

async def api_test_tools(test_body: Dict[str, Any]) -> Dict[str, Any]:
    """
    智能接口测试执行器
    
    Args:
        test_body: 测试请求规范，包含：
            - data_type: 参数类型 (params|json|form-data|form-urlencoded)
            - URL: 接口路径
            - method: HTTP方法
            - params: 请求参数
            - headers: 请求头 (可选)
            - file: 上传文件路径 (可选)
    
    Returns:
        接口响应或错误信息
    """
    # 参数校验与预处理
    config_env = get_config()['env']
    data_type = test_body.get('data_type', '').lower()
    required_keys = ['URL', 'method', 'params']
    
    if not all(key in test_body for key in required_keys):
        return {'code': '400', 'msg': '缺少必填参数'}
    
    if data_type not in ['params', 'json', 'form-data', 'form-urlencoded']:
        return {'code': '400', 'msg': f'不支持的参数类型: {data_type}'}
    
    # 构建完整请求参数
    request_spec = {
        'url': config_env + test_body['URL'],
        'method': test_body['method'].lower(),
        'data': test_body['params'],
        'headers': test_body.get('headers', {}),
        'file': test_body.get('file')
    }
    
    try:
        # 根据数据类型选择执行策略
        strategy = {
            'params': _execute_params,
            'json': _execute_json,
            'form-data': _execute_formdata,
            'form-urlencoded': _execute_urlencoded
        }
        return strategy[data_type](request_spec)
    except Exception as e:
        return {'code': '500', 'msg': f'测试执行异常: {str(e)}'}

def _execute_params(spec: Dict[str, Any]) -> Dict[str, Any]:
    try:
        response = session.get(
            spec['url'],
            params=spec['data'],
            headers=spec['headers']
        )
        response.raise_for_status()
        return _handle_response(response)
    except requests.RequestException as e:
        return {'code': str(e.response.status_code) if e.response else '500', 
                'msg': f'请求失败: {str(e)}'}

def _execute_json(spec: Dict[str, Any]) -> Dict[str, Any]:
    try:
        response = session.request(
            method=spec['method'],
            url=spec['url'],
            json=spec['data'],
            headers=spec['headers']
        )
        return _handle_response(response)
    except requests.RequestException as e:
        return {'code': str(e.response.status_code) if e.response else '500', 
                'msg': f'请求失败: {str(e)}'}

def _execute_urlencoded(spec: Dict[str, Any]) -> Dict[str, Any]:
    try:
        response = session.request(
            method=spec['method'],
            url=spec['url'],
            data=spec['data'],
            headers=spec['headers']
        )
        return _handle_response(response)
    except requests.RequestException as e:
        return {'code': str(e.response.status_code) if e.response else '500', 
                'msg': f'请求失败: {str(e)}'}

def _execute_formdata(spec: Dict[str, Any]) -> Dict[str, Any]:
    file_path = spec.get('file')
    if not file_path:
        return {'code': '400', 'msg': '文件上传模式需要指定file参数'}
    
    if not os.path.exists(file_path):
        return {'code': '404', 'msg': f'文件不存在: {file_path}'}
    
    try:
        with safe_file_opener(file_path) as f:
            files = {
                'file': (os.path.basename(file_path), 
                        f, 
                        get_mime_type(file_path))
            }
            response = session.request(
                method=spec['method'],
                url=spec['url'],
                data=spec['data'],
                files=files,
                headers=spec['headers']
            )
        return _handle_response(response)
    except requests.RequestException as e:
        return {'code': str(e.response.status_code) if e.response else '500', 
                'msg': f'上传失败: {str(e)}'}
    except IOError as e:
        return {'code': '500', 'msg': f'文件读取失败: {str(e)}'}

def _handle_response(response: requests.Response) -> Dict[str, Any]:
    """统一处理响应"""
    try:
        # 尝试解析JSON响应
        return response.json()
    except ValueError:
        # 非JSON响应处理
        return {
            'status_code': response.status_code,
            'text': response.text,
            'headers': dict(response.headers)
        }

在这个接口工具实现中，我们首先配置了会话池和重试策略，以提高请求的稳定性和可靠性。然后，我们定义了一系列辅助函数，如读取配置文件、获取文件MIME类型、安全打开文件等。api_test_tools函数是主要的接口测试执行器，它根据输入的测试请求规范，选择合适的执行策略（如发送GET请求、POST请求、上传文件等），并返回接口响应或错误信息。每个执行策略函数（如_execute_params、_execute_json等）都负责处理特定类型的请求，并调用_handle_response函数统一处理响应。

构建接口测试智能体

使用顺序智能体（SequentialAgent）构建接口测试智能体。智能体的主要工作流程为：

解析接口文档，将接口数据存到数据库或者缓存中
执行接口测试
总结测试报告，发送测试报告

[!NOTE]

提示词太长，这里就不写了。

下面我们分别定义三个子智能体，分别负责接口文档解析、接口测试执行和测试报告总结：

from google.adk.agents.llm_agent import LlmAgent  
from google.adk.agents.sequential_agent import SequentialAgent  
  
  
# 1. 接口文档解析Agent  
code_writer = LlmAgent(  
    name="ApiParse",  
    model=LiteLlm(
        model="qwen-plus-latest", 
        api_key="your api keys", 
        api_base="https://dashscope.aliyuncs.com/compatible-mode/v1"
    ),   
    instruction="根据用户的接口文档解析为结构化数据",  
    description="提示词",  
    output_key="api_data"  
)  
  
# 2. 接口测试Agent    
code_reviewer = LlmAgent(  
    name="ApiTest",   
    model=LiteLlm(
        model="qwen-plus-latest", 
        api_key="your api keys", 
        api_base="https://dashscope.aliyuncs.com/compatible-mode/v1"
    ),  
    instruction="根据解析的api_data，执行接口测试",  
    description="提示词",  
    output_key="test_result",
    tools=[api_test_tools]
)  
  
# 3. 测试总结Agent  
code_refactorer = LlmAgent(  
    name="ApiReport",  
    model=LiteLlm(
        model="qwen-plus-latest", 
        api_key="your api keys", 
        api_base="https://dashscope.aliyuncs.com/compatible-mode/v1"
    ),  
    instruction="根据审查意见重构代码",  
    description="负责代码重构优化",  
    output_key="refactored_code"  
)  
  
# --- 创建顺序执行的根Agent ---  
root_agent = SequentialAgent(  
    name="CodeProcessingPipeline",  
    sub_agents=[code_writer, code_reviewer, code_refactorer],  
    description="按顺序执行代码编写、审查和重构"  
)

在这个顺序智能体中，code_writer首先负责解析用户输入的接口文档，将其转换为结构化数据api_data。然后，code_reviewer根据解析得到的api_data执行接口测试，生成测试结果test_result。最后，code_refactorer根据测试结果总结测试报告，并提供优化建议。通过这种顺序执行的方式，我们实现了一个完整的接口测试流程。

通过以上步骤，我们成功构建了一个基于Google-ADK框架的接口自动化测试智能体。该智能体能够理解自然语言输入的接口文档，自动生成测试脚本并执行测试，最后提供详细的测试报告和优化建议。本文只提供一个创建接口测试智能体的思路。具体代码只作为演示示例。