告别Demo｜手把手教你构建可用的LangChain测试智能体

1. 任务理解与规划模块（Task Planner）
2. 知识与应用工具库（Tools & Knowledge）
3. 测试执行引擎（Test Executor）
4. 记忆与反馈系统（Memory & Feedback）
5. 结果分析与报告模块（Analyzer & Reporter）

# 核心库
pip install langchain langchain-openai
pip install playwright  # 用于 UI 自动化
pip install requests pytest  # 用于 API 测试
pip install sqlalchemy  # 用于数据库验证

# 如果你需要更强大的规划能力，可以考虑使用 LangGraph
pip install langgraph

from langchain.tools import BaseTool
from typing import Type, Optional
from pydantic import BaseModel, Field
import requests
import json

class APITestInput(BaseModel):
    endpoint: str = Field(description="API 端点地址")
    method: str = Field(description="HTTP 方法，如 GET、POST")
    payload: Optional[dict] = Field(None, description="请求体")
    expected_status: int = Field(200, description="期望的 HTTP 状态码")

class APITestTool(BaseTool):
    name = "api_test_tool"
    description = "执行 API 测试并验证响应"
    args_schema: Type[BaseModel] = APITestInput
    
    def _run(self, endpoint: str, method: str, payload: dict = None, expected_status: int = 200):
        """实际执行测试"""
        try:
            if method.upper() == "GET":
                response = requests.get(endpoint, params=payload)
            elif method.upper() == "POST":
                response = requests.post(endpoint, json=payload)
            else:
                return {"error": f"不支持的 HTTP 方法: {method}"}
            
            # 基础断言
            success = response.status_code == expected_status
            result = {
                "success": success,
                "status_code": response.status_code,
                "response_body": response.json() if response.content elseNone,
                "message": f"状态码验证{'通过' if success else '失败'}"
            }
            
            # 记录到测试记忆
            self.memory_store.append(result)
            return result
            
        except Exception as e:
            return {"error": f"API 测试执行异常: {str(e)}"}
    
    memory_store = []  # 简单的记忆存储

from langchain.agents import AgentExecutor, create_react_agent
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain.memory import ConversationBufferMemory

# 测试专属的 Prompt 模板
TEST_AGENT_PROMPT = PromptTemplate.from_template(
    """你是一个资深的自动化测试工程师。请遵循以下步骤执行测试任务：
    
1. **需求分析**：首先理解测试目标，识别测试类型（功能、性能、安全等）
2. **环境检查**：确认测试环境可用性
3. **测试设计**：设计测试场景和用例，考虑边界条件和异常场景
4. **工具选择**：选择合适的测试工具
5. **执行验证**：执行测试并验证结果
6. **结果分析**：分析测试结果，给出结论和建议

当前任务：{input}

你拥有以下工具：
{tools}

{agent_scratchpad}"""
)

# 初始化 LLM
llm = ChatOpenAI(
    model="gpt-4-turbo",
    temperature=0.1,  # 测试需要确定性，温度值设低
    api_key=os.getenv("OPENAI_API_KEY")
)

# 创建 Agent
tools = [APITestTool(), UITestTool(), DBValidationTool()]
agent = create_react_agent(
    llm=llm,
    tools=tools,
    prompt=TEST_AGENT_PROMPT
)

# 创建执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=ConversationBufferMemory(
        memory_key="chat_history",
        return_messages=True
    ),
    verbose=True,
    handle_parsing_errors=True
)

class TestContextMemory:
    def __init__(self):
        self.test_cases = []
        self.environment_info = {}
        self.execution_history = []
        self.known_issues = []
    
    def add_test_case(self, test_case):
        """记录测试用例"""
        self.test_cases.append({
            "id": len(self.test_cases) + 1,
            "description": test_case.description,
            "steps": test_case.steps,
            "expected": test_case.expected,
            "actual": None,
            "status": "pending"
        })
    
    def record_execution(self, test_id, result):
        """记录执行结果"""
        for tc in self.test_cases:
            if tc["id"] == test_id:
                tc["actual"] = result.get("response_body") or result.get("output")
                tc["status"] = "passed"if result.get("success") else"failed"
                break
        
        self.execution_history.append({
            "test_id": test_id,
            "timestamp": datetime.now(),
            "result": result
        })

class TestDataManager:
    def __init__(self):
        self.data_pool = {}
        
    def generate_test_data(self, schema: dict, constraints: list = None):
        """根据数据模式生成测试数据"""
        # 集成第三方测试数据生成库
        # 支持边界值、异常值生成
        pass
    
    def cleanup(self):
        """测试数据清理"""
        # 自动清理测试产生的数据
        pass

from concurrent.futures import ThreadPoolExecutor, as_completed

class ConcurrentTestRunner:
    def __init__(self, max_workers=5):
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
    
    def run_concurrent_tests(self, test_cases):
        """并发执行测试用例"""
        futures = {}
        for test_case in test_cases:
            future = self.executor.submit(
                agent_executor.invoke,
                {"input": f"执行测试：{test_case['description']}"}
            )
            futures[future] = test_case["id"]
        
        results = []
        for future in as_completed(futures):
            test_id = futures[future]
            try:
                result = future.result()
                results.append({"test_id": test_id, "result": result})
            except Exception as e:
                results.append({"test_id": test_id, "error": str(e)})
        
        return results

# 定义测试任务
test_mission = """
请测试用户下单完整流程：
1. 用户登录（使用测试账号 test_user@example.com）
2. 浏览商品列表，选择第一个商品
3. 加入购物车
4. 进入结算页面，填写收货地址
5. 选择支付方式（模拟支付）
6. 确认下单
7. 验证订单状态变为'已支付'
8. 验证库存相应减少
注意：请检查每个步骤的异常情况，如库存不足、地址无效等。
"""

# 执行测试
result = agent_executor.invoke({
    "input": test_mission,
    "chat_history": []
})

# 分析结果
print(f"测试完成，耗时: {result['execution_time']}")
print(f"测试用例数: {len(memory.test_cases)}")
print(f"通过率: {sum(1 for tc in memory.test_cases if tc['status'] == 'passed') / len(memory.test_cases) * 100:.1f}%")

# GitLab CI 示例
stages:
-test

ai_automated_test:
stage:test
image:python:3.11
script:
    -pipinstall-rrequirements.txt
    -pythontest_agent_runner.py--env$ENVIRONMENT--suite$TEST_SUITE
artifacts:
    reports:
      junit:test-results/report.xml