合集：LangChain智能体开发（二）接口测试智能体示例

可遇_不可求

458人浏览 · 2025-12-16 09:44:26

可遇_不可求 · 2025-12-16 09:44:26 发布

大语言模型（LLM）的崛起让智能体（AI Agent）成为技术圈的热门话题。无论是自动化客服、个性化助手，还是复杂任务求解，智能体正逐步渗透到各个领域。而LangChain作为一款强大的LLM应用框架，凭借其模块化设计和丰富的工具链，成为开发者构建智能体的首选工具之一。

本文将带你从零开始，使用LangChain开发智能体，并分享代码和实现思路。

多智能体？

在上篇文章中，我们学习了怎么创建一个LangChain智能体。怎么让智能体使用工具。怎么流式输出智能体步骤结果。在本篇内容中，我们主要探索一下怎么使用多智能体协调工作。或组装智能体工作流。

开发环境准备

开发环境依旧是我们上一篇内容的开发环境。

多智能体设计思路

上篇文章中，我们知道了怎么创建一个带有工具的智能体。那么我们多智能体协调的思路是创建一个主智能体，其主要职责是进行意图识别、工作分工、任务调度等。然后我们封装几个工具。将各个子智能体封装成一个工具供主智能体使用。

主智能体实现

from langchain.agents import create_agent
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

# 加载环境变量
load_dotenv()

model = ChatOpenAI(
    model="deepseek-chat",
    base_url="https://api.deepseek.com",
    api_key=os.getenv("OPENAI_API_KEY", "")
)

agent = create_agent(
    model=model,
    system_prompt="你是一个资深的接口测试工程师，当用户发送给你接口文档是，首先使用工具解析接口信息。其次将解析的接口信息传给脚本生成工具生成测试脚本，包括正用例和反用例，并执行接口测试。最后将测试结果传给报告工具进行总结，生成测试报告。",
    tools=[analyze_api, generate_script, report_tool],
)

接口分析智能体

async def analyze_api(api_doc):
    """
    解析接口文档信息
    :return:
        dict: {
    	"domain": "https:www.baidu.com",
    	"url": "/login",
    	"method": "post",
    	"data_type": "json",
    	"headers": {
    		"Content-Type": "application/json"
    	},
    	"params": {
    		"user_name": "test",
    		"password": "123123123123"
    	}
    }
    """
    model_instance = ChatOpenAI(
        model="deepseek-chat",
        base_url="https://api.deepseek.com",
        api_key=os.getenv("OPENAI_API_KEY", "")
    )

    prompt = """
    你是接口文档分析工程师，你的职责是将用户的接口文档进行整理。返回固定的json格式数据，格式如下：
    {
    	"domain": "https:www.baidu.com",
    	"url": "/login",
    	"method": "post",
    	"data_type": "json",
    	"headers": {
    		"Content-Type": "application/json"
    	},
    	"params": {
    		"user_name": "test",
    		"password": "123123123123"
    	}
    }
    """

    analyze_api_agent = create_agent(
        model=model_instance,
        system_prompt=prompt,
    )

    # 使用异步流式调用
    async for chunk in analyze_api_agent.astream(
            {"messages": [{"role": "user", "content": api_doc}]}
    ):
        for step, data in chunk.items():
            print(f"step: {step}")
            print(f"content: {data['messages'][-1].content_blocks}")

脚本执行工具

import os
import mimetypes
from typing import Dict, Any
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))


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:
        接口响应或错误信息
    """
    # 参数校验与预处理

    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': test_body.get('domain', '')+ test_body.get('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)
        }

脚本生成智能体

async def generate_script(test_body):
    """
        根据接口信息生成测试脚本并执行

        Args:
            test_body: 接口测试相关信息

        Returns:
            list: 测试执行结果列表
        """
    model_instance = ChatOpenAI(
        model="deepseek-chat",
        base_url="https://api.deepseek.com",
        api_key=os.getenv("OPENAI_API_KEY", "")
    )

    prompt = """
    你是资深接口测试工程师，你的职责是将test_body的接口信息生成正反用例，并生成脚本，调用api_test_tools工具执行生成的脚本。将执行结果集合发送给生成报告智能体。
    保证test_body的格式为，没有的使用空字符串：
    {
    	"domain": "https:www.baidu.com",
    	"url": "/login",
    	"method": "post",
    	"data_type": "json",
    	"headers": {
    		"Content-Type": "application/json"
    	},
    	"params": {
    		"user_name": "test",
    		"password": "123123123123"
    	}
    }
    注意：接口失败不需要重试，只需要返回失败的结果给生成报告工具
    """

    analyze_api_agent = create_agent(
        model=model_instance,
        system_prompt=prompt,
        tools=[api_test_tools]
    )

    # 使用异步流式调用
    async for chunk in analyze_api_agent.astream(
            {"messages": [{"role": "user", "content": test_body}]}
    ):
        for step, data in chunk.items():
            print(f"step: {step}")
            print(f"content: {data['messages'][-1].content_blocks}")

生成报告智能体

async def report_tool(result_list):
    """
       生成测试报告

       Args:
           result_list: 测试结果列表

       Returns:
           str: 格式化的测试报告
       """
    model_instance = ChatOpenAI(
        model="deepseek-chat",
        base_url="https://api.deepseek.com",
        api_key=os.getenv("OPENAI_API_KEY", "")
    )

    prompt = """
    你的职责是总结测试结果。将测试结果以markdown的格式输出。
    """

    analyze_api_agent = create_agent(
        model=model_instance,
        system_prompt=prompt
    )

    # 使用异步流式调用
    async for chunk in analyze_api_agent.astream(
            {"messages": [{"role": "user", "content": result_list}]}
    ):
        for step, data in chunk.items():
            print(f"step: {step}")
            print(f"content: {data['messages'][-1].content_blocks}")

执行智能体

async def main():
    api_doc = """
    url
    http://www.baidu.com/s
    请求方式
    GET
    请求类型
    params
    wd=微信公众平台
    """

    # 开启异步流式响应
    async for chunk in agent.astream(
        {"messages": [{"role": "user", "content": api_doc}]}
    ):
        for step, data in chunk.items():
            print(f"step: {step}")
            print(f"content: {data['messages'][-1].content_blocks}")

# 运行异步函数
if __name__ == "__main__":
    asyncio.run(main())

执行结果

step: model
content: [{'type': 'text', 'text': '我来帮您进行接口测试。首先，让我解析您提供的接口文档信息。'}, {'type': 'tool_call', 'id': 'call_00_nTitIDfLnzDnfJvUdR6jtBAd', 'name': 'analyze_api', 'args': {'api_doc': '    url\n    http://www.baidu.com/s\n    请求方式\n    GET\n    请求类型\n    params\n    wd=微信公众平台'}}]
step: model
content: [{'type': 'text', 'text': '{\n    "domain": "http://www.baidu.com",\n    "url": "/s",\n    "method": "GET",\n    "data_type": "params",\n    "headers": {},\n    "params": {\n        "wd": "微信公众平台"\n    }\n}'}]
step: tools
content: [{'type': 'text', 'text': 'null'}]
step: model
content: [{'type': 'text', 'text': '让我重新整理一下接口文档信息，使用更清晰的格式：'}, {'type': 'tool_call', 'id': 'call_00_dAQJqF0dz4KLQPO3vc8fPfIz', 'name': 'analyze_api', 'args': {'api_doc': '接口文档：\nurl: http://www.baidu.com/s\n请求方式: GET\n请求类型: params\n参数: wd=微信公众平台'}}]
step: model
content: [{'type': 'text', 'text': '```json\n{\n    "domain": "http://www.baidu.com",\n    "url": "/s",\n    "method": "GET",\n    "data_type": "params",\n    "headers": {},\n    "params": {\n        "wd": "微信公众平台"\n    }\n}\n```'}]
step: tools
content: [{'type': 'text', 'text': 'null'}]
step: model
content: [{'type': 'text', 'text': '让我尝试使用更结构化的格式：'}, {'type': 'tool_call', 'id': 'call_00_Bty7Na715nZM4OAr7tLE4JMA', 'name': 'analyze_api', 'args': {'api_doc': '{\n  "url": "http://www.baidu.com/s",\n  "method": "GET",\n  "data_type": "params",\n  "params": {\n    "wd": "微信公众平台"\n  }\n}'}}]
step: model
content: [{'type': 'text', 'text': '```json\n{\n    "domain": "http://www.baidu.com",\n    "url": "/s",\n    "method": "GET",\n    "data_type": "params",\n    "headers": {\n        "Content-Type": "application/x-www-form-urlencoded"\n    },\n    "params": {\n        "wd": "微信公众平台"\n    }\n}\n```'}]
step: tools
content: [{'type': 'text', 'text': 'null'}]
step: model
content: [{'type': 'text', 'text': '看起来工具可能有一些问题。让我直接基于您提供的接口信息来生成测试脚本。根据您提供的信息，我了解到：\n\n1. 接口URL: http://www.baidu.com/s\n2. 请求方式: GET\n3. 请求类型: params (查询参数)\n4. 参数: wd=微信公众平台\n\n现在让我基于这些信息生成测试脚本：'}, {'type': 'tool_call', 'id': 'call_00_Ka00H027BPhVKgVS1fDL6iHW', 'name': 'generate_script', 'args': {'test_body': {'domain': 'http://www.baidu.com', 'url': '/s', 'method': 'GET', 'data_type': 'params', 'headers': {'Content-Type': 'application/x-www-form-urlencoded'}, 'params': {'wd': '微信公众平台'}}}}]
step: tools
content: [{'type': 'text', 'text': "Error invoking tool 'generate_script' with kwargs {'test_body': {'domain': 'http://www.baidu.com', 'url': '/s', 'method': 'GET', 'data_type': 'params', 'headers': {'Content-Type': 'application/x-www-form-urlencoded'}, 'params': {'wd': '微信公众平台'}}} with error:\n content.str: Input should be a valid string\ncontent.list[union[str,dict[any,any]]]: Input should be a valid list\n Please fix the error and try again."}]
step: model
content: [{'type': 'text', 'text': '看起来参数格式有问题。让我重新尝试：'}, {'type': 'tool_call', 'id': 'call_00_fXNpGHwNfsqYa7SSikT2sksn', 'name': 'generate_script', 'args': {'test_body': '{\n  "domain": "http://www.baidu.com",\n  "url": "/s",\n  "method": "GET",\n  "data_type": "params",\n  "headers": {\n    "Content-Type": "application/x-www-form-urlencoded"\n  },\n  "params": {\n    "wd": "微信公众平台"\n  }\n}'}}]
step: model
content: [{'type': 'text', 'text': '我将为您分析这个接口并生成正反用例，然后执行测试。首先让我分析一下接口信息：\n\n**接口分析：**\n- 域名：http://www.baidu.com\n- 路径：/s\n- 方法：GET\n- 参数类型：params\n- 参数：wd="微信公众平台"\n\n现在我将生成正反用例并执行测试：'}, {'type': 'tool_call', 'id': 'call_00_zmfBC7V0rf0f7Fw1Evfyoz8b', 'name': 'api_test_tools', 'args': {'test_body': {'data_type': 'params', 'URL': 'http://www.baidu.com/s', 'method': 'GET', 'params': {'wd': '微信公众平台'}, 'headers': {'Content-Type': 'application/x-www-form-urlencoded'}}}}]
step: tools
content: [{'type': 'text', 'text': '{"status_code": 200, "text": "<!DOCTYPE html>\\n<html la...}]
step: model
content: [{'type': 'text', 'text': '现在让我测试一些边界情况和异常情况：'}, {'type': 'tool_call', 'id': 'call_00_3DWzlVEpCyURwxtfyiUAZ0SZ', 'name': 'api_test_tools', 'args': {'test_body': {'data_type': 'params', 'URL': 'http://www.baidu.com/s', 'method': 'GET', 'params': {'wd': ''}, 'headers': {'Content-Type': 'application/x-www-form-urlencoded'}}}}]
step: tools
content: [{'type': 'text', 'text': '{"status_code": 200, "text": "<!DOCTYPE html>\\n<html>\\n...'}]
step: model
content: [{'type': 'text', 'text': '现在测试一个特殊字符的搜索：'}, {'type': 'tool_call', 'id': 'call_00_Q1guVxt8CM7l9zj0TzFOMmWv', 'name': 'api_test_tools', 'args': {'test_body': {'data_type': 'params', 'URL': 'http://www.baidu.com/s', 'method': 'GET', 'params': {'wd': 'test@#$%^&*()_+'}, 'headers': {'Content-Type': 'application/x-www-form-urlencoded'}}}}]
step: tools
content: [{'type': 'text', 'text': '{"status_code": 200, "text": "<!DOCTYPE html>\\n<htm...'}]
step: model
content: [{'type': 'text', 'text': '现在测试一个超长字符串：'}, {'type': 'tool_call', 'id': 'call_00_ZybY6PRr8tamLYsJfi2fbVcE', 'name': 'api_test_tools', 'args': {'test_body': {'data_type': 'params', 'URL': 'http://www.baidu.com/s', 'method': 'GET', 'params': {'wd': '这是一个非常长的搜索关键词，用于测试百度搜索接口对长字符串的处理能力，看看是否会截断或者返回错误信息，这个字符串应该超过100个字符以上，以确保测试的完整性。'}, 'headers': {'Content-Type': 'application/x-www-form-urlencoded'}}}}]
step: tools
content: [{'type': 'text', 'text': '{"status_code": 200, "text": "<!DOCTYPE h...'}]
step: model
content: [{'type': 'text', 'text': '现在测试一个不存在的域名，看看错误处理：'}, {'type': 'tool_call', 'id': 'call_00_kcEjNXLRGtzSoYQi1umbY1EC', 'name': 'api_test_tools', 'args': {'test_body': {'data_type': 'params', 'URL': 'http://www.nonexistentdomain12345.com/s', 'method': 'GET', 'params': {'wd': 'test'}, 'headers': {'Content-Type': 'application/x-www-form-urlencoded'}}}}]
step: tools
content: [{'type': 'text', 'text': '{"code": "500", "msg": "请求失败: HTTPConnectionPool(host=\'www.nonexistentdomain12345.com\', port=80): Max retries exceeded with url: /s?wd=test (Caused by NameResolutionError(\\"<urllib3.connection.HTTPConnection object at 0x7588fc35c950>: Failed to resolve \'www.nonexistentdomain12345.com\' ([Errno -2] Name or service not known)\\"))"}'}]
step: model
content: [{'type': 'text', 'text': '现在测试一个错误的URL路径：'}, {'type': 'tool_call', 'id': 'call_00_Sa7oiTDSYji7F2Kh6FCM5G4V', 'name': 'api_test_tools', 'args': {'test_body': {'data_type': 'params', 'URL': 'http://www.baidu.com/nonexistent', 'method': 'GET', 'params': {'wd': 'test'}, 'headers': {'Content-Type': 'application/x-www-form-urlencoded'}}}}]
step: tools
content: [{'type': 'text', 'text': '{"code": "500", "msg": "请求失败: 404 Client Error: Not Found for url: http://www.baidu.com/nonexistent?wd=test"}'}]
step: model
content: [{'type': 'text', 'text': '现在测试一个错误的HTTP方法：'}, {'type': 'tool_call', 'id': 'call_00_x3YJiHgc6QRzqP0otpwKcSzk', 'name': 'api_test_tools', 'args': {'test_body': {'data_type': 'params', 'URL': 'http://www.baidu.com/s', 'method': 'POST', 'params': {'wd': 'test'}, 'headers': {'Content-Type': 'application/x-www-form-urlencoded'}}}}]
step: tools
content: [{'type': 'text', 'text': '{"status_code": 200, "text": "<!DOCTYPE htm...'}]
step: model
content: [{'type': 'text', 'text': '现在测试一个正常的英文搜索：'}, {'type': 'tool_call', 'id': 'call_00_WrTjQDOFzHsijNxp0MfzPSe7', 'name': 'api_test_tools', 'args': {'test_body': {'data_type': 'params', 'URL': 'http://www.baidu.com/s', 'method': 'GET', 'params': {'wd': 'python programming'}, 'headers': {'Content-Type': 'application/x-www-form-urlencoded'}}}}]
step: tools
content: [{'type': 'text', 'text': '{"status_code": 200, "text": "<!DOCTYPE html...'}]
step: model
content: [{'type': 'text', 'text': '## 测试执行完成\n\n我已经为百度搜索接口 `/s` 执行了全面的正反用例测试。以下是测试结果的总结：\n\n### **测试用例设计：**\n\n**正向用例：**\n1. **正常中文搜索** - 搜索"微信公众平台"（状态码：200，返回安全验证页面）\n2. **空搜索词** - 搜索空字符串（状态码：200，返回百度首页）\n3. **英文搜索** - 搜索"python programming"（状态码：200，返回安全验证页面）\n4. **特殊字符搜索** - 搜索"test@#$%^&*()_+"（状态码：200，返回安全验证页面）\n5. **超长字符串搜索** - 搜索超过100字符的长文本（状态码：200，返回安全验证页面）\n\n**反向用例：**\n1. **错误HTTP方法** - 使用POST方法（状态码：200，返回安全验证页面）\n2. **错误URL路径** - 访问不存在的路径（状态码：404，返回404错误）\n3. **不存在的域名** - 访问不存在的域名（状态码：500，域名解析失败）\n\n### **测试结果分析：**\n\n**主要发现：**\n1. **安全验证机制**：百度搜索接口对大多数请求都返回了安全验证页面，而不是实际的搜索结果页面\n2. **接口稳定性**：所有有效的请求都返回了200状态码，说明接口本身是稳定的\n3. **错误处理**：\n   - 不存在的域名：返回500错误，域名解析失败\n   - 不存在的路径：返回404错误\n   - 其他异常情况：都返回了安全验证页面\n\n**接口行为特点：**\n- 百度搜索接口似乎有安全验证机制，对于非正常访问会返回安全验证页面\n- 即使使用错误的HTTP方法（如POST），接口也返回200状态码\n- 接口对参数长度和特殊字符都有较好的容错性\n\n**建议：**\n1. 对于生产环境测试，可能需要处理安全验证机制\n2. 接口的异常处理机制比较完善\n3. 可以考虑测试更多边界情况，如超长参数、特殊编码等\n\n所有测试结果已收集完成，可以发送给生成报告智能体进行详细分析和报告生成。'}]