测试全流程智能Agent体系（Python+LangChain）

基于 Python + LangChain 实现 5 大 Agent 协同的核心代码骨架，我会按照模块化、可运行、易扩展的原则编写，可以直接基于这个骨架对接实际的大模型、自动化工具和存储系统。

---

一、整体架构总览

目标：从PRD→测试点→用例→执行→分析→bug→回归，全链路自动化、自治闭环。

PRD/需求文档
   ↓
【需求分析Agent】→ 测试点清单 + 风险评估报告
   ↓
【用例生成Agent】→ 结构化测试用例（含RAG知识库增强）
   ↓
【执行Agent】→ 接口/UI/性能自动化执行
   ↓
【分析Agent】→ 失败用例根因分析 + 标准Bug报告
   ↓
【协调Agent】→ 决策：回归/新增用例/阻塞/结束

---

二、每个Agent详细设计

1. 需求分析Agent

核心定位：需求解析专家 + 测试点提取 + 风险评估
输入

• PRD、原型图、接口文档、业务流程图
  输出
• 标准化测试点清单（模块-功能-场景-优先级）
• 需求风险评估（逻辑矛盾、缺失、边界、性能、兼容性、安全）
• 可测性评估报告
  关键能力
• PRD文档解析（PDF/Word/URL/飞书文档）
• 歧义、缺失、矛盾点识别
• 自动拆分：主流程 / 异常 / 边界 / 性能 / 安全
• 输出结构化JSON，给下游用例Agent

---

2. 用例生成Agent

核心定位：用例工程师 + RAG增强生成
输入

• 需求Agent输出的测试点
• RAG知识库：历史用例、业务规则、接口规范、用例库
  输出
• 完整可执行测试用例
  • 标题、前置、步骤、预期、优先级、模块
  • 接口用例：URL、method、header、参数、断言
  • UI用例：页面、操作、校验点
    关键能力
• RAG检索相似用例 → 复用、改写、适配
• 自动生成：正向、逆向、边界、并发、异常用例
• 输出标准化格式，可直接导入测试平台

---

3. 执行Agent

核心定位：自动化测试执行器（接口/UI/性能）
输入

• 用例Agent产出的用例
• 环境配置、账号、数据构造
  输出
• 执行结果：成功/失败/阻塞
• 日志、截图、录像、接口返回、性能指标
  关键能力
• 接口自动化（HTTP/HTTPS/GRPC/Dubbo）
• UI自动化（Web/App）
• 性能压测（并发、RT、TPS、错误率）
• 测试数据自动准备/清理

---

4. 分析Agent

核心定位：测试分析师 + 缺陷自动提交
输入

• 执行失败日志、截图、返回包、服务端日志
  输出
• 失败根因分类
  • 用例问题 / 环境问题 / 代码Bug / 配置问题
• 自动生成Bug报告（标题、重现步骤、附件、优先级）
  关键能力
• 日志关键词匹配、异常栈解析
• 接口返回码、字段校验失败定位
• 自动推送到缺陷平台（Jira/Tapd等）

---

5. 协调Agent（大脑）

核心定位：流程控制器 + 决策中枢
输入

• 所有Agent的状态与结果
  输出
• 全局执行策略
  关键决策
• 失败用例是否重跑
• 是否需要新增补充用例
• 是否环境异常 → 暂停并告警
• 是否完成测试 → 输出测试报告
• 是否进入回归测试
  能力
• 多Agent调度、依赖管理、异常容错
• 最终产出：一站式智能测试报告

---

三、全流程自治闭环

1. 上传PRD
2. 需求分析Agent → 测试点+风险
3. 用例生成Agent → 生成用例
4. 协调Agent → 分发执行
5. 执行Agent → 执行
6. 分析Agent → 定位+提Bug
7. 协调Agent → 判断：
   - Bug多 → 继续回归
   - 用例不足 → 召回用例Agent补充
   - 环境异常 → 暂停并告警
   - 全部通过 → 结束并输出报告

---

四、极简技术栈

• 大模型：GPT-4o / 文心4.0 / 豆包API
• 框架：LangChain + 自定义Agent
• RAG：测试用例库 + 业务规则库 + 缺陷库
• 自动化：Playwright + Requests + JMeter
• 存储：MySQL/ES + Redis
• 输出：对接测试平台/Jira/Tapd

---

前置依赖

先安装核心依赖包：

pip install langchain langchain-openai python-dotenv requests playwright pyyaml
# 如需对接国内大模型，补充安装
pip install langchain-community zhipuai dashscope

核心代码骨架

import os
import json
import time
from typing import Dict, List, Any
from dotenv import load_dotenv
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langchain_openai import ChatOpenAI
# 如需对接国内大模型（如智谱/通义千问），替换上面的导入
# from langchain_community.chat_models import ChatZhipuAI, ChatTongyi

# 加载环境变量（存放API Key、环境配置等）
load_dotenv()

# ====================== 基础配置 ======================
class Config:
    """全局配置类"""
    # 大模型配置（替换为你的API Key和模型）
    LLM_API_KEY = os.getenv("OPENAI_API_KEY")
    LLM_MODEL = "gpt-4o-mini"  # 可替换为glm-4、qwen-plus等
    # 测试环境配置
    TEST_ENV_URL = os.getenv("TEST_ENV_URL", "https://test-api.example.com")
    # 输出路径
    OUTPUT_DIR = "./test_output"
    os.makedirs(OUTPUT_DIR, exist_ok=True)

# 初始化大模型
llm = ChatOpenAI(
    api_key=Config.LLM_API_KEY,
    model=Config.LLM_MODEL,
    temperature=0.1  # 低随机性保证输出稳定
)

# ====================== 1. 需求分析Agent ======================
class RequirementAnalysisAgent:
    """解析PRD、提取测试点、评估风险"""
    def __init__(self):
        # 测试点提取Prompt模板
        self.test_point_prompt = PromptTemplate(
            template="""
            你是资深测试分析师，请完成以下任务：
            1. 解析PRD内容，提取核心测试点（按模块-功能-场景分层）；
            2. 评估每个测试点的风险等级（高/中/低），并说明风险点；
            3. 输出格式必须为JSON，结构如下：
            {
                "module": "模块名",
                "test_points": [
                    {
                        "function": "功能名",
                        "scene": "测试场景",
                        "priority": "高/中/低",
                        "risk_level": "高/中/低",
                        "risk_desc": "风险描述"
                    }
                ],
                "overall_risk": "整体风险评估（高/中/低）",
                "ambiguity_points": ["PRD中模糊/缺失的点列表"]
            }

            PRD内容：{prd_content}
            """,
            input_variables=["prd_content"]
        )
        self.parser = JsonOutputParser()

    def analyze_prd(self, prd_content: str) -> Dict[str, Any]:
        """
        解析PRD生成测试点和风险评估
        :param prd_content: PRD文本内容
        :return: 结构化测试点+风险评估结果
        """
        print("=== 需求分析Agent：开始解析PRD ===")
        # 构建链并执行
        chain = self.test_point_prompt | llm | self.parser
        try:
            result = chain.invoke({"prd_content": prd_content})
            # 保存结果
            self._save_result("requirement_analysis_result.json", result)
            print("=== 需求分析Agent：PRD解析完成 ===")
            return result
        except Exception as e:
            print(f"需求分析Agent执行失败：{e}")
            raise

    def _save_result(self, filename: str, data: Dict):
        """保存分析结果到文件"""
        with open(os.path.join(Config.OUTPUT_DIR, filename), "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)

# ====================== 2. 用例生成Agent ======================
class TestCaseGenerationAgent:
    """基于测试点+RAG生成结构化测试用例"""
    def __init__(self):
        # 用例生成Prompt模板（区分接口/UI用例）
        self.case_prompt = PromptTemplate(
            template="""
            你是资深测试工程师，请基于测试点生成可执行的测试用例：
            1. 接口用例需包含：标题、前置条件、请求URL、请求方法、请求头、请求参数、预期结果、断言规则；
            2. UI用例需包含：标题、前置条件、操作步骤、预期结果、校验点；
            3. 覆盖正向、逆向、边界场景；
            4. 输出格式为JSON，结构如下：
            {
                "module": "模块名",
                "test_cases": [
                    {
                        "case_id": "用例ID（如TC_模块_功能_001）",
                        "title": "用例标题",
                        "type": "接口/UI/性能",
                        "priority": "高/中/低",
                        "pre_condition": "前置条件",
                        "steps": ["步骤1", "步骤2"],
                        "expected_result": "预期结果",
                        "assert_rule": "断言规则（如状态码200、字段xxx=yyy）"
                    }
                ]
            }

            测试点：{test_points}
            历史用例参考（RAG）：{rag_reference}
            """,
            input_variables=["test_points", "rag_reference"]
        )
        self.parser = JsonOutputParser()

    def generate_cases(self, test_points: Dict, rag_reference: str = "") -> Dict[str, Any]:
        """
        生成测试用例
        :param test_points: 需求分析Agent输出的测试点
        :param rag_reference: RAG检索的历史用例参考（默认空）
        :return: 结构化测试用例
        """
        print("=== 用例生成Agent：开始生成测试用例 ===")
        chain = self.case_prompt | llm | self.parser
        try:
            result = chain.invoke({
                "test_points": json.dumps(test_points, ensure_ascii=False),
                "rag_reference": rag_reference
            })
            self._save_result("test_cases_result.json", result)
            print("=== 用例生成Agent：用例生成完成 ===")
            return result
        except Exception as e:
            print(f"用例生成Agent执行失败：{e}")
            raise

    def _save_result(self, filename: str, data: Dict):
        """保存用例结果到文件"""
        with open(os.path.join(Config.OUTPUT_DIR, filename), "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)

# ====================== 3. 执行Agent ======================
class TestExecutionAgent:
    """自动化执行接口/UI/性能测试"""
    def __init__(self):
        self.test_env_url = Config.TEST_ENV_URL

    def execute_api_case(self, case: Dict) -> Dict[str, Any]:
        """执行接口测试用例"""
        print(f"=== 执行Agent：开始执行接口用例 {case['case_id']} ===")
        result = {
            "case_id": case["case_id"],
            "case_title": case["title"],
            "type": "接口",
            "status": "failed",  # 默认失败，执行成功后更新
            "actual_result": "",
            "error_msg": "",
            "execution_time": time.strftime("%Y-%m-%d %H:%M:%S")
        }
        try:
            # 模拟接口请求（替换为实际的requests/httpx调用）
            import requests
            response = requests.request(
                method=case.get("steps")[0].get("method", "GET"),
                url=f"{self.test_env_url}{case.get('steps')[0].get('url')}",
                headers=case.get("steps")[0].get("headers", {}),
                json=case.get("steps")[0].get("params", {})
            )
            # 断言校验
            assert response.status_code == 200, f"状态码不是200，实际为{response.status_code}"
            assert response.json().get("code") == 0, f"业务码不是0，实际为{response.json().get('code')}"
            
            result["status"] = "passed"
            result["actual_result"] = response.json()
        except Exception as e:
            result["error_msg"] = str(e)
            result["actual_result"] = str(response.text) if 'response' in locals() else ""
        print(f"=== 执行Agent：接口用例 {case['case_id']} 执行完成，状态：{result['status']} ===")
        return result

    def execute_ui_case(self, case: Dict) -> Dict[str, Any]:
        """执行UI测试用例（基于Playwright）"""
        # 此处仅为骨架，需结合Playwright实现实际UI操作
        print(f"=== 执行Agent：开始执行UI用例 {case['case_id']} ===")
        result = {
            "case_id": case["case_id"],
            "case_title": case["title"],
            "type": "UI",
            "status": "passed",  # 模拟成功
            "actual_result": "UI操作执行完成，校验点通过",
            "error_msg": "",
            "execution_time": time.strftime("%Y-%m-%d %H:%M:%S")
        }
        print(f"=== 执行Agent：UI用例 {case['case_id']} 执行完成 ===")
        return result

    def execute_all_cases(self, test_cases: Dict) -> List[Dict]:
        """执行所有测试用例"""
        all_results = []
        for case in test_cases["test_cases"]:
            if case["type"] == "接口":
                result = self.execute_api_case(case)
            elif case["type"] == "UI":
                result = self.execute_ui_case(case)
            else:
                result = {
                    "case_id": case["case_id"],
                    "status": "unsupported",
                    "error_msg": "不支持的用例类型"
                }
            all_results.append(result)
        # 保存执行结果
        with open(os.path.join(Config.OUTPUT_DIR, "execution_results.json"), "w", encoding="utf-8") as f:
            json.dump(all_results, f, ensure_ascii=False, indent=2)
        return all_results

# ====================== 4. 分析Agent ======================
class TestAnalysisAgent:
    """失败用例根因分析，生成Bug报告"""
    def __init__(self):
        self.root_cause_prompt = PromptTemplate(
            template="""
            你是测试分析专家，请分析以下失败用例的根因，并生成标准化Bug报告：
            1. 根因分类：用例问题/环境问题/代码Bug/配置问题；
            2. Bug报告包含：标题、重现步骤、实际结果、预期结果、根因分析、优先级、附件；
            3. 仅为失败用例生成Bug报告，成功用例忽略；
            4. 输出格式为JSON。

            执行结果：{execution_results}
            """,
            input_variables=["execution_results"]
        )
        self.parser = JsonOutputParser()

    def analyze_failures(self, execution_results: List[Dict]) -> Dict[str, Any]:
        """分析失败用例，生成Bug报告"""
        print("=== 分析Agent：开始根因分析 ===")
        # 过滤失败用例
        failed_cases = [r for r in execution_results if r["status"] == "failed"]
        if not failed_cases:
            print("=== 分析Agent：无失败用例，无需生成Bug报告 ===")
            return {"bug_reports": []}
        
        # 生成Bug报告
        chain = self.root_cause_prompt | llm | self.parser
        try:
            result = chain.invoke({
                "execution_results": json.dumps(failed_cases, ensure_ascii=False)
            })
            # 保存Bug报告
            with open(os.path.join(Config.OUTPUT_DIR, "bug_reports.json"), "w", encoding="utf-8") as f:
                json.dump(result, f, ensure_ascii=False, indent=2)
            print("=== 分析Agent：Bug报告生成完成 ===")
            return result
        except Exception as e:
            print(f"分析Agent执行失败：{e}")
            raise

# ====================== 5. 协调Agent ======================
class CoordinationAgent:
    """统筹整个流程，决策下一步"""
    def __init__(self):
        self.agents = {
            "requirement": RequirementAnalysisAgent(),
            "case_generation": TestCaseGenerationAgent(),
            "execution": TestExecutionAgent(),
            "analysis": TestAnalysisAgent()
        }

    def run_full_process(self, prd_content: str, rag_reference: str = "") -> Dict[str, Any]:
        """运行完整测试流程"""
        print("=== 协调Agent：启动测试全流程 ===")
        # 1. 需求分析
        requirement_result = self.agents["requirement"].analyze_prd(prd_content)
        # 2. 用例生成
        test_cases = self.agents["case_generation"].generate_cases(requirement_result, rag_reference)
        # 3. 用例执行
        execution_results = self.agents["execution"].execute_all_cases(test_cases)
        # 4. 失败分析
        bug_reports = self.agents["analysis"].analyze_failures(execution_results)
        # 5. 决策下一步
        decision = self._make_decision(execution_results, bug_reports)
        # 生成最终报告
        final_report = {
            "process_status": "completed",
            "requirement_analysis": requirement_result,
            "test_cases_count": len(test_cases["test_cases"]),
            "execution_summary": {
                "total": len(execution_results),
                "passed": len([r for r in execution_results if r["status"] == "passed"]),
                "failed": len([r for r in execution_results if r["status"] == "failed"]),
                "unsupported": len([r for r in execution_results if r["status"] == "unsupported"])
            },
            "bug_reports_count": len(bug_reports["bug_reports"]),
            "next_step_decision": decision
        }
        # 保存最终报告
        with open(os.path.join(Config.OUTPUT_DIR, "final_test_report.json"), "w", encoding="utf-8") as f:
            json.dump(final_report, f, ensure_ascii=False, indent=2)
        print("=== 协调Agent：测试全流程执行完成 ===")
        return final_report

    def _make_decision(self, execution_results: List[Dict], bug_reports: Dict) -> str:
        """决策下一步行动"""
        failed_count = len([r for r in execution_results if r["status"] == "failed"])
        bug_count = len(bug_reports["bug_reports"])
        if failed_count == 0:
            return "所有用例执行成功，测试流程结束，可输出测试报告"
        elif bug_count > 5:  # 自定义阈值
            return "失败用例较多，建议修复Bug后执行回归测试"
        elif any(["环境问题" in bug.get("根因分析", "") for bug in bug_reports["bug_reports"]]):
            return "存在环境问题，需先修复环境后重新执行测试"
        else:
            return "部分用例失败，已生成Bug报告，等待开发修复后执行回归测试"

# ====================== 运行示例 ======================
if __name__ == "__main__":
    # 1. 模拟PRD内容（替换为实际PRD文本）
    sample_prd = """
    需求名称：用户登录接口
    需求描述：
    1. 支持手机号+验证码登录；
    2. 支持账号密码登录；
    3. 登录成功返回token，失效时间2小时；
    4. 密码错误3次锁定账号10分钟；
    5. 接口地址：/api/v1/login，请求方法：POST。
    """
    # 2. 模拟RAG参考（历史用例/业务规则）
    sample_rag = """
    历史登录接口用例：
    1. 手机号为空返回错误码1001；
    2. 验证码错误返回错误码1002；
    3. 密码复杂度不足返回错误码1003。
    """
    # 3. 启动协调Agent执行全流程
    coordinator = CoordinationAgent()
    final_report = coordinator.run_full_process(sample_prd, sample_rag)
    # 打印最终结果
    print("\n=== 最终测试报告摘要 ===")
    print(f"总用例数：{final_report['test_cases_count']}")
    print(f"通过数：{final_report['execution_summary']['passed']}")
    print(f"失败数：{final_report['execution_summary']['failed']}")
    print(f"Bug数：{final_report['bug_reports_count']}")
    print(f"下一步决策：{final_report['next_step_decision']}")

代码使用说明

1. 环境配置

在项目根目录创建 .env 文件，配置以下内容：

# 大模型API Key（替换为你的）
OPENAI_API_KEY=sk-xxxxxx
# 测试环境地址
TEST_ENV_URL=https://test-api.example.com

2. 核心扩展点（你需要根据实际场景补充）

Agent	扩展点
需求分析Agent	对接实际PRD文档解析（PDF/Word/飞书文档）、完善风险评估规则
用例生成Agent	接入RAG知识库（如FAISS/ES）、适配公司用例模板
执行Agent	完善Playwright UI自动化、JMeter性能测试、测试数据构造/清理逻辑
分析Agent	对接Jira/Tapd等缺陷平台自动提Bug、接入服务端日志分析
协调Agent	增加定时执行、异常告警（邮件/钉钉）、回归测试触发逻辑

3. 运行方式

python test_agent_system.py

运行后会在 ./test_output 目录生成所有中间结果和最终报告：

• requirement_analysis_result.json：需求分析结果
• test_cases_result.json：生成的测试用例
• execution_results.json：用例执行结果
• bug_reports.json：Bug报告
• final_test_report.json：最终测试报告

总结

1. 代码采用模块化设计，每个Agent独立封装，便于单独调试和扩展；
2. 基于LangChain的Prompt+Parser模式，保证大模型输出结构化JSON，避免格式混乱；
3. 核心流程完整覆盖“需求分析→用例生成→执行→分析→协调”全链路，可直接落地；
4. 预留了RAG、自动化工具、缺陷平台的对接接口，只需补充具体实现即可适配实际业务。