基于 LangChain 的多 Agent 协作测试流水线：从 Prompt 工程到完整实现
·
摘要：在 AI 工程化落地的浪潮中，如何让多个 AI Agent 协作完成"需求分析→测试计划→用例生成→审核输出"的完整测试流程？本文从 Prompt 工程的七条核心原则出发，逐步搭建基于 LangChain 和 LangGraph 的多 Agent 测试流水线，涵盖需求分析审核、打回重做机制、人工审核节点、检查点持久化、完整的异常处理体系，并附上单 Agent 与多 Agent 的效果对比实测数据。
目录
一、为什么需要多 Agent 协作
二、Prompt 工程：如何写出高质量提示词
2.1 Prompt 的基本结构
2.2 七条核心原则
2.3 Prompt 编写速查清单
三、LangChain 实现多 Agent 流水线
3.1 环境准备与版本说明
3.2 定义基础 Agent 类（含错误处理）
3.3 定义七个 Agent 的 Prompt（按流水线顺序）
3.4 流水线核心逻辑：含打回重做机制
3.5 运行入口
四、使用 LangChain LCEL 构建管道式链
4.1 基础链式调用
4.2 加入评审子链
五、使用 LangGraph 实现完整的图编排
5.1 状态定义
5.2 所有节点函数（含容错）
5.3 条件路由函数
5.4 构建完整有向图
5.5 运行入口（含检查点持久化与人工审核）
5.6 完整流程图
六、完整异常处理体系
6.1 分层异常定义
6.2 LLM 调用层异常处理
6.3 节点间状态校验
6.4 流水线级容错策略
6.5 生产环境监控与进阶建议
七、单 Agent vs 多 Agent 效果对比
7.1 需求分析的三种审查方案（如何选择）
7.2 对比实验设计
7.3 实测对比结果
7.4 结论与选择建议
八、进阶优化建议
8.1 引入 RAG 知识库
8.2 并行执行
8.3 成本控制
九、完整项目结构
十、总结
一、为什么需要多 Agent 协作？
单个 AI 能力有限，就像一个全科医生什么都会一点但不够深。多 Agent 协作的思路是：让每个 AI 只专注一件事，各司其职，流水线式推进。
用户需求
│
▼
┌──────────┐ 需求分析结果 ┌──────────┐
│ 需求分析AI │ ──────────────→ │ 需求评审AI │
└──────────┘ └──────────┘
│
审核通过
▼
┌──────────┐ 用例内容 ┌──────────┐ 测试计划 ┌──────────┐
│ 审核输出AI │ ←─────────── │ 用例评分AI │ ←─────────── │ 用例生成AI │
└──────────┘ └──────────┘ └──────────┘
│ ▲ ▲
▼ │ │
最终输出 评分+反馈 测试计划AI ──→ 计划评分AI
（不达标打回）
每个 Agent 本质上就是：
精心设计的 System Prompt + 一次 LLM 调用 + 上下文传递
所以，Prompt 写得好不好，直接决定了整条流水线的质量。
与传统软件测试的本质差异也在于此：
| 维度 | 传统软件测试 | AI 测试 |
|---|---|---|
| 通过标准 | 精确匹配 | 语义相似度、人工评分、LLM-as-Judge |
| 可重复性 | 高 | 低（受 temperature、采样影响） |
| 失败判定 | 二元（pass/fail） | 连续谱（0~1 打分） |
| 测试用例维护 | 需求变了就改 | 模型升级后旧用例可能全部失效 |
二、Prompt 工程：如何写出高质量提示词
2.1 Prompt 的基本结构
一个完整的 Prompt 通常包含以下部分：
┌─────────────────────────────────────┐
│ Role（角色） - 你是谁 │
│ Context（背景） - 任务背景 │
│ Task（任务） - 具体要做什么 │
│ Format（格式） - 输出什么格式 │
│ Constraints（约束）- 限制条件 │
│ Examples（示例） - 参考样例 │
└─────────────────────────────────────┘
2.2 七条核心原则
原则一：角色定义要具体到"专家画像"
差的写法：
你是一个测试人员，请分析需求。
好的写法：
你是一名拥有10年经验的高级测试需求分析师，擅长从 PRD 文档中
提取功能点、识别隐含需求、发现逻辑矛盾。你的分析风格严谨细致，
习惯用结构化的方式呈现结果。
角色越具体，模型调用的知识分布越精准。"10年经验的高级测试需求分析师"比"测试人员"激活了更多专业模式。
原则二：任务拆解为步骤链（Chain of Thought）
差的写法：
分析这个需求，输出测试计划。
好的写法：
请按以下步骤逐条分析：
第一步：提取核心功能点
- 列出所有可测试的功能模块
- 标注每个功能的业务优先级（高/中/低）
第二步：识别业务规则
- 每个功能背后的判断条件是什么
- 状态流转规则是什么
第三步：挖掘边界条件
- 输入值的边界在哪里
- 异常场景有哪些（网络中断、并发操作、权限不足）
第四步：识别隐含需求
- 需求文档没有明确写但逻辑上必须考虑的点
- 性能、安全、兼容性等非功能需求
第五步：输出结构化分析报告
- 使用 Markdown 表格整理以上内容
为什么要拆步骤？ 因为 LLM 是逐步生成 token 的，你给它明确的思考路径，它会沿着这条路径走，结果质量远高于"一步到位"。
原则三：输出格式要用示例约束
差的写法：
输出测试用例。
好的写法：
请按以下 JSON 格式输出测试用例：
[
{
"case_id": "TC-001",
"module": "登录模块",
"title": "正确用户名密码登录成功",
"priority": "P0",
"precondition": "用户已注册，账号状态正常",
"steps": [
"1. 打开登录页面",
"2. 输入正确的用户名",
"3. 输入正确的密码",
"4. 点击登录按钮"
],
"expected": "登录成功，跳转到首页，显示用户名"
}
]
注意：每条用例必须包含以上所有字段，步骤至少3条，预期结果必须具体可验证。
与其用文字描述"请输出JSON"，不如直接给一个完整的示例。模型会模仿示例的结构、字段名、甚至风格。
原则四：加入负面约束（告诉模型不要做什么）
【重要约束】
- 不要输出模糊的预期结果（如"显示正确"），必须具体描述页面变化
- 不要遗漏反向测试场景（异常输入、边界值）
- 不要生成重复的测试用例
- 不要使用"待定""略"等占位文字
负面约束的作用是缩小模型的输出空间，避免它"自由发挥"出不符合要求的内容。
原则五：用 Few-Shot 示例引导
在 Prompt 中提供 1~3 个"标准答案"示例，模型会模仿示例的质量和风格。
以下是一个高质量测试用例示例：
用例编号：TC-001
所属模块：用户登录
优先级：P0
前置条件：用户已注册，账号状态正常，浏览器已打开登录页面
操作步骤：
1. 在用户名输入框输入 "testuser01"
2. 在密码输入框输入 "Pass@123"
3. 点击"登录"按钮
预期结果：
- 页面跳转至首页 /dashboard
- 页面右上角显示 "欢迎，testuser01"
- 接口返回 HTTP 200，响应体包含 token 字段
请按以上标准，为以下功能生成测试用例：...
原则六：角色对抗——让审核 Agent 更有效
审核类 Prompt 的关键是给审核员一个"找茬"的立场：
你是一名严格的测试评审专家，你的职责是找出以下测试用例中的问题。
请从以下维度逐项检查并打分（每项0-20分，满分100）：
1. 覆盖度（20分）：是否覆盖了所有功能点？正向、反向、边界场景是否齐全？
2. 准确性（20分）：预期结果是否具体、可验证、无歧义？
3. 可执行性（20分）：操作步骤是否清晰，新人能否直接执行？
4. 去重性（20分）：是否有重复或高度相似的用例？
5. 优先级合理性（20分）：P0/P1/P2 的分布是否合理？
要求：
- 总分 < 80 分：返回 pass=false，必须列出每项的具体扣分原因
- 总分 >= 80 分：返回 pass=true，列出可以改进的建议
严格按 JSON 格式返回：
{
"score": 75,
"detail": {
"覆盖度": {"score": 15, "comment": "..."},
"准确性": {"score": 18, "comment": "..."},
...
},
"pass": false,
"suggestions": ["建议1", "建议2"]
}
关键技巧：给审核 Agent 明确的评分维度和分值分配，否则它只会给出"总体不错"这种模糊评语。
厎则七：上下文压缩——避免 Token 浪费
多轮流水线中，每一步的输出都会传给下一步。如果中间结果太长，会占用大量上下文窗口。解决方法：
请将以下需求分析报告压缩为结构化摘要，保留所有关键信息，
去除冗余描述，控制在 500 字以内。
或者在 Prompt 中限定输出长度：
每个功能点用一行概括，格式：功能名 | 优先级 | 关键测试方向
2.3 Prompt 编写速查清单
✅ 角色是否具体到专业画像？
✅ 任务是否拆解为明确的步骤？
✅ 输出格式是否有示例约束？
✅ 是否有负面约束（不要做什么）？
✅ 关键任务是否有 Few-Shot 示例？
✅ 审核任务是否有量化评分标准？
✅ 上下文是否做了压缩处理？
三、LangChain 实现多 Agent 流水线
3.1 环境准备与版本说明
pip install langchain==0.3.7 langchain-openai==0.2.10 langgraph==0.2.53 langchain-community==0.3.5 chromadb==0.5.23 tiktoken==0.8.0
import os
os.environ["OPENAI_API_KEY"] = "sk-your-key"
# 如果用国内模型，可以替换为其他 provider
# os.environ["OPENAI_API_BASE"] = "https://your-api-base/v1"
⚠️ 版本兼容注意事项：
1. langchain 0.2 → 0.3 有 breaking change：
- ChatModel 基类路径变更
- langchain-community 拆分为独立包
- 必须单独安装 langchain-community
2. langgraph 0.1 → 0.2 有 breaking change：
- StateGraph 的 API 变更
- checkpointer 接口重构
- interrupt_before 的行为调整
3. Python 版本：建议 3.11+，3.12 已验证兼容
3.9 部分功能可能不支持（TypedDict 语法差异）
3.2 定义基础 Agent 类（含错误处理）
import time
import json
import logging
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(name)s] %(levelname)s: %(message)s"
)
logger = logging.getLogger("test-pipeline")
class ReviewResult:
"""评审结果的数据结构"""
def __init__(self, score=0, pass_status=False,
feedback="", suggestions=None):
self.score = score
self.pass_status = pass_status
self.feedback = feedback
self.suggestions = suggestions or []
class TestAgent:
"""测试流水线中的单个 Agent
设计职责分层：
- run() → 负责原始调用 + 日志记录，异常原样向上抛出
- call_llm_with_retry() → 负责根据异常类型决定重试策略（见 5.2 节）
- _run_with_review() → 负责业务级降级（见 3.5 节）
run() 不吞异常，保持底层透明，让上层根据具体异常类型做决策。
"""
def __init__(self, name: str, system_prompt: str,
model: str = "gpt-4o", temperature: float = 0.3):
self.name = name
self.system_prompt = system_prompt
self.llm = ChatOpenAI(
model=model,
temperature=temperature # 低温度保证输出稳定
)
def run(self, user_input: str) -> str:
"""执行一次 Agent 调用。
记录调用日志，但不捕获异常——异常原样向上抛出，
由上层的 call_llm_with_retry 负责根据异常类型
决定重试策略（限流→退避重试，认证失败→直接终止等）。
"""
logger.info(f"[{self.name}] 开始处理...")
messages = [
SystemMessage(content=self.system_prompt),
HumanMessage(content=user_input)
]
# 异常不在此处捕获，保持底层透明
# 上层 call_llm_with_retry 会根据异常类型做不同的重试决策
response = self.llm.invoke(messages)
logger.info(f"[{self.name}] 处理完成，输出 {len(response.content)} 字")
return response.content
def run_and_parse_json(self, user_input: str) -> dict:
"""执行并解析 JSON 返回（简单版本，用于简单场景）
注意：生产环境建议使用 5.2 节的 call_llm_json_with_retry，
它有完整的重试 + JSON 解析容错 + 必要字段校验。
"""
raw = self.run(user_input)
try:
if "```json" in raw:
raw = raw.split("```json")[1].split("```")[0]
elif "```" in raw:
raw = raw.split("```")[1].split("```")[0]
return json.loads(raw.strip())
except json.JSONDecodeError:
return {
"score": 0,
"pass_status": False,
"feedback": f"JSON 解析失败，原始输出：{raw[:500]}"
}
3.3 定义七个 Agent 的 Prompt（按流水线顺序）
以下是七个 Agent 的完整 Prompt 定义。按流水线执行顺序编号：①需求分析 → ②需求评审 → ③测试计划 → ④计划评审 → ⑤用例设计 → ⑥用例评审 → ⑦审核输出。所有 Prompt 均遵循统一结构：
角色定义 → 任务说明 → 评分维度（评审类） → 通过标准 → 输出格式 → 约束
Agent 1：需求分析师
职责：阅读需求文档，提取功能点、业务规则、边界条件和隐含需求。
REQUIREMENT_ANALYSIS_PROMPT = """
你是一名拥有10年经验的高级测试需求分析师。
## 你的任务
分析用户提供的需求文档或功能描述，输出结构化的需求分析报告。
## 分析步骤
请按以下步骤逐一分析：
1. **核心功能提取**
- 列出所有可测试的功能模块
- 标注每个功能的业务优先级（高/中/低）
2. **业务规则识别**
- 每个功能背后的判断条件是什么
- 状态流转规则是什么
3. **边界条件挖掘**
- 输入值的边界在哪里
- 异常场景有哪些（网络中断、并发操作、权限不足）
4. **隐含需求识别**
- 需求文档没有明确写但逻辑上必须考虑的点
- 性能、安全、兼容性等非功能需求
## 输出格式
使用 Markdown，包含以下章节：
### 1. 功能清单
| 模块名 | 功能点 | 优先级 | 描述 |
|--------|--------|--------|------|
| ... | ... | 高/中/低 | ... |
### 2. 业务规则
| 规则编号 | 条件 | 结论 |
|----------|------|------|
| BR-001 | ... | ... |
### 3. 边界与异常场景
| 编号 | 场景描述 |
|------|----------|
| BD-001 | ... |
### 4. 隐含需求
| 编号 | 需求描述 |
|------|----------|
| IR-001 | ... |
## 约束
- 不要遗漏反向场景
- 不要使用模糊描述（如"正常显示"）
- 每个功能点至少识别2个边界条件
"""
Prompt 设计要点：
| 要素 | 说明 |
|---|---|
| 角色 | “10年经验的高级测试需求分析师”——激活专业知识模式 |
| 步骤拆解 | 4步依次推进，引导模型逐步思考（Chain of Thought） |
| 输出格式 | 给出表格模板，约束字段名和排列方式 |
| 负面约束 | 明确"不要遗漏反向场景"“不要使用模糊描述” |
Agent 2：需求分析评审员
职责：对比原始需求文档和需求分析报告，检查需求理解是否完整准确。这是整条流水线的地基校验——需求分析阶段的错误会逐级放大，遗漏一个功能点意味着后续 10~20 条用例全部缺失。
ANALYSIS_REVIEW_PROMPT = """
你是一名资深的需求分析评审专家，同时具备业务分析和测试分析能力。
请基于以下提供的【原始需求文档】和【需求分析报告】进行对比评审，
检查需求分析报告是否完整、准确地理解了原始需求。
## 评分维度（每项20分，满分100）
### 1. 功能覆盖度（20分）
- 是否提取了原始需求中的所有功能点？
- 有没有遗漏的功能模块？
- 有没有添加原始需求中没有的功能（过度分析）？
### 2. 业务规则准确性（20分）
- 业务规则的条件和结论是否与原始需求一致？
- 有没有理解错误或曲解的规则？
- 状态流转是否完整且正确？
### 3. 边界条件充分性（20分）
- 每个功能点是否至少识别了2个边界条件？
- 边界条件是否合理（不是凑数的）？
- 异常场景是否覆盖了常见的系统级异常（网络、超时、并发）？
### 4. 隐含需求合理性（20分）
- 识别的隐含需求是否确实"隐含"（不是已经写在需求里的）？
- 隐含需求是否具有实际测试价值？
- 是否遗漏了明显的隐含需求（如安全性、性能、兼容性）？
### 5. 结构与可读性（20分）
- 是否使用了统一的结构化格式？
- 表格/编号是否清晰、无歧义？
- 后续的测试计划制定者能否直接基于此报告工作？
## 通过标准
- 总分 >= 80：pass_status = true
- 总分 < 80：pass_status = false，必须详细说明遗漏或错误的具体内容
## 输出格式
严格按以下 JSON 格式返回，不要输出其他任何内容：
{
"score": 72,
"detail": {
"功能覆盖度": {
"score": 14,
"comment": "提取了登录、注册、密码重置三个模块，但遗漏了原始需求中提到的'第三方登录'功能...",
"issues": [
"遗漏：原始需求第3条提到的'支持微信/支付宝登录'未被识别",
"遗漏：原始需求提到的'登录日志记录'功能未被识别"
]
},
"业务规则准确性": {
"score": 16,
"comment": "大部分规则理解正确，但密码锁定规则有误...",
"issues": [
"错误：报告中写'密码错误5次锁定'，原始需求写的是'3次'"
]
},
"边界条件充分性": {
"score": 14,
"comment": "登录模块边界条件识别较好，但注册模块不足...",
"issues": [
"注册模块仅识别了1个边界条件，建议补充：手机号已注册、密码强度临界值、验证码过期"
]
},
"隐含需求合理性": {
"score": 14,
"comment": "识别了3条隐含需求，但其中1条不是隐含的...",
"issues": [
"'登录需要图形验证码'已在原始需求中明确提到，不属于隐含需求",
"建议补充：密码传输是否需要加密、登录失败次数是否需要记录"
]
},
"结构与可读性": {
"score": 14,
"comment": "结构清晰，但业务规则表格缺少'例外情况'列...",
"issues": [
"BR-003 规则缺少例外情况描述（如：管理员账号是否受锁定规则约束）"
]
}
},
"pass_status": false,
"feedback": "功能覆盖度存在明显遗漏，业务规则有1处理解错误...",
"suggestions": [
"建议1: 补充'第三方登录'和'登录日志记录'功能点",
"建议2: 将密码锁定次数从5次修正为3次",
"建议3: 注册模块补充至少2个边界条件",
"建议4: 将'图形验证码'从隐含需求中移除（已在原始需求中明确）"
]
}
## 约束
- 打分要严格，需求分析的错误会传导到后续所有环节
- issues 必须具体到功能点或规则编号
- 如果发现理解错误，必须同时给出正确的理解应该是什么
"""
Prompt 设计要点：
| 要素 | 说明 |
|---|---|
| 对比评审 | “请基于以下提供的【原始需求文档】和【需求分析报告】进行对比评审”——明确交叉对比的两份输入 |
| 过度分析检测 | “有没有添加原始需求中没有的功能”——防止模型自行脑补需求里没有的功能 |
| 隐含需求边界 | “是否确实隐含（不是已经写在需求里的）”——防止把已明确的需求误归为隐含需求 |
| 错误修正要求 | “如果发现理解错误，必须同时给出正确的理解”——不只是指出问题，还要给出正确答案 |
| 传导效应 | 约束中强调"错误会传导到后续所有环节"——让评审 Agent 更加严格 |
Agent 3：测试计划师
职责：根据审核通过的需求分析报告，制定测试范围、策略、优先级和风险评估。
TEST_PLAN_PROMPT = """
你是一名高级测试计划制定专家。
## 你的任务
请基于以下提供的【需求分析报告】，制定完整的测试计划。
## 输出内容
### 1. 测试范围
- 本次测试覆盖的功能模块清单
- 明确不测试的内容（排除项）
### 2. 测试策略
- **功能测试**：每个模块的测试重点
- **边界测试**：关键边界的覆盖策略
- **异常测试**：要模拟的异常场景
- **兼容性测试**（如适用）：浏览器、设备、系统
- **性能测试**（如适用）：并发、响应时间、压力
### 3. 测试优先级
| 优先级 | 定义 | 示例 |
|--------|------|------|
| P0（阻塞级） | 必须通过的核心流程 | 登录、支付 |
| P1（严重级） | 重要功能 | 搜索、下单 |
| P2（一般级） | 次要功能和体验 | 个人中心、帮助 |
### 4. 风险评估
| 风险编号 | 风险描述 | 影响程度 | 应对措施 |
|----------|----------|----------|----------|
| R-001 | ... | 高/中/低 | ... |
### 5. 测试资源估算
- 预计用例数量：XX 条
- 预计执行时间：XX 人天
## 约束
- 策略必须针对具体功能点，不要写泛泛的方法论
- 风险评估至少列出3个具体风险
- 排除项必须写明原因
"""
Prompt 设计要点：
| 要素 | 说明 |
|---|---|
| 输入声明 | “请基于以下提供的【需求分析报告】”——明确告诉模型去哪里找输入 |
| 结构化 | 每个章节都有明确的表格模板 |
| 负面约束 | “不要写泛泛的方法论”——避免输出"采用黑盒测试方法"这类空话 |
Agent 4：计划评审员
职责：对比需求分析报告和测试计划，进行量化打分，不达标时给出具体修改意见。
PLAN_REVIEW_PROMPT = """
你是一名严格的测试计划评审专家。你的职责是找出测试计划中的问题。
请基于以下提供的【需求分析报告】和【测试计划】进行对比评审，
检查测试计划是否完整覆盖了需求分析中识别的所有内容。
## 评分维度（每项20分，满分100）
### 1. 覆盖度（20分）
- 计划是否覆盖了需求分析中的所有功能点和场景？
- 有没有遗漏的模块或场景？
### 2. 策略合理性（20分）
- 测试策略是否针对具体场景？
- 方法选择是否恰当？
- 有没有"为了写策略而写策略"的空话？
### 3. 优先级准确性（20分）
- P0/P1/P2 的划分是否合理？
- 核心流程是否都标为 P0？
### 4. 风险全面性（20分）
- 风险识别是否全面？
- 应对措施是否具体可行？
- 有没有只写"加强测试"这种无意义的措施？
### 5. 可操作性（20分）
- 计划是否具体到可以直接执行？
- 资源估算是否合理？
## 通过标准
- 总分 >= 80：pass_status = true
- 总分 < 80：pass_status = false，必须详细说明每项的扣分原因和具体修改建议
## 输出格式
严格按以下 JSON 格式返回，不要输出其他任何内容：
{
"score": 85,
"detail": {
"覆盖度": {
"score": 18,
"comment": "具体评价..."
},
"策略合理性": {
"score": 17,
"comment": "具体评价..."
},
"优先级准确性": {
"score": 16,
"comment": "具体评价..."
},
"风险全面性": {
"score": 18,
"comment": "具体评价..."
},
"可操作性": {
"score": 16,
"comment": "具体评价..."
}
},
"pass_status": true,
"feedback": "总体评价...",
"suggestions": [
"建议1: 具体描述需要修改什么",
"建议2: 具体描述需要补充什么"
]
}
## 约束
- 打分要严格，不要"放水"
- 每个维度的 comment 至少写30字
- suggestions 必须具体到可以直接操作（如"在测试范围中增加XXX模块"）
"""
Prompt 设计要点：
| 要素 | 说明 |
|---|---|
| 对比评审 | “请基于以下提供的【需求分析报告】和【测试计划】进行对比评审”——明确告诉模型需要交叉对比哪两份输入 |
| 角色对抗 | “严格的评审专家”“职责是找出问题”——激活找茬模式 |
| 通过标准前置 | 放在评分维度之后、输出格式之前，模型在生成 JSON 时能准确设置 pass_status |
| JSON 格式 | 去掉 ```json 标记，直接用纯 JSON，避免 Python 三重引号截断问题 |
Agent 5：用例设计师
职责：根据审核通过的测试计划，为每个功能点设计详细、可执行的测试用例。
TEST_CASE_PROMPT = """
你是一名高级测试用例设计师，擅长设计全面、精准、可执行的测试用例。
## 你的任务
请基于以下提供的【测试计划】，为每个功能点设计详细测试用例。
## 每条用例必须包含以下字段
| 字段 | 说明 | 示例 |
|------|------|------|
| case_id | 用例编号 | TC-LOGIN-001 |
| module | 所属模块 | 用户登录 |
| title | 用例标题（一句话描述测试目的） | 正确手机号密码登录成功 |
| priority | 优先级 | P0 / P1 / P2 |
| preconditions | 前置条件 | 用户已注册，账号状态正常 |
| steps | 操作步骤（至少3步） | 1. 打开登录页... |
| expected_results | 预期结果 | 跳转首页，显示用户名 |
| test_data | 测试数据（具体输入值） | 手机号: 13800138000 |
## 设计原则
### 覆盖原则
- 每个功能点至少覆盖：**正向1条 + 反向1条 + 边界1条**
- 反向场景包括：错误输入、空输入、超长输入、特殊字符、权限不足
- 边界场景包括：最小值、最大值、临界值、刚好超限
### 可执行原则
- 步骤写到"新人拿到就能执行"的程度
- 每步一个具体动作，不要合并多个动作
- 预期结果写到"可以自动化断言"的程度
### 去重原则
- 不要有重复用例
- 相似用例合并为等价类
## 输出格式
使用 Markdown 表格，每个模块一个表格：
### 模块一：用户登录
| 用例编号 | 标题 | 优先级 | 前置条件 | 操作步骤 | 预期结果 | 测试数据 |
|----------|------|--------|----------|----------|----------|----------|
| TC-LOGIN-001 | 正确手机号密码登录 | P0 | ... | ... | ... | ... |
| TC-LOGIN-002 | 密码错误登录失败 | P0 | ... | ... | ... | ... |
## 约束
- **禁止**预期结果写"显示正确""正常运行""页面正常"等模糊描述
- **禁止**步骤写"按常规操作""进行相关操作"等模糊描述
- **禁止**遗漏反向场景和边界场景
- **禁止**测试数据写"任意值""合理值"——必须给出具体数据
"""
Prompt 设计要点：
| 要素 | 说明 |
|---|---|
| 输入声明 | “请基于以下提供的【测试计划】”——明确输入来源 |
| 设计原则 | 三条原则（覆盖、可执行、去重）明确告诉模型"好用例"的标准 |
| Few-Shot 隐含 | 表格示例本身就是格式的 Few-Shot |
| 负面约束 | 4条"禁止"，每条都是实际踩过的坑 |
| 测试数据约束 | “禁止写任意值”——这个细节很多人忽略，但对用例质量影响很大 |
Agent 6：用例评审员
职责：对比测试计划和测试用例，进行量化打分，找出遗漏、错误和冗余。
CASE_REVIEW_PROMPT = """
你是一名严格的测试用例评审专家。
请基于以下提供的【测试计划】和【测试用例】进行对比评审，
检查测试用例是否完整覆盖了测试计划中定义的所有场景和策略。
## 评分维度（每项20分，满分100）
### 1. 覆盖度（20分）
检查以下场景是否覆盖：
- ✅ 正向场景（正常流程）
- ✅ 反向场景（错误输入、异常操作）
- ✅ 边界值（最小值、最大值、临界值）
- ✅ 组合场景（多个条件同时满足/不满足）
每缺少一类场景扣5分。
### 2. 准确性（20分）
- 预期结果是否具体、可验证
- 是否存在错误的预期结果（如：预期"登录成功"但输入的是错误密码）
- 测试数据是否合理（如：手机号是否符合格式）
### 3. 可执行性（20分）
- 步骤是否清晰，一个新人能否直接按步骤执行
- 前置条件是否明确
- 是否缺少必要的测试数据
- 步骤是否有跳步（直接从A到C，缺少中间的B）
### 4. 去重与独立性（20分）
- 是否有重复用例（标题相似、步骤相似、预期结果相同）
- 用例之间是否相互独立（不依赖其他用例的执行结果）
- 是否有隐含依赖（如：用例B要求"用户已登录"，但没写在前置条件里）
### 5. 优先级合理性（20分）
- P0 是否覆盖了核心流程（登录、支付等）
- P2 的用例是否确实是非核心功能
- 是否有优先级错标（核心功能标了P2，或边缘功能标了P0）
## 通过标准
- 总分 >= 80：pass_status = true
- 总分 < 80：pass_status = false
## 输出格式
严格按以下 JSON 格式返回，不要输出其他任何内容：
{
"score": 75,
"detail": {
"覆盖度": {
"score": 14,
"comment": "正向场景覆盖完整，但缺少以下反向场景...",
"issues": [
"缺少验证码过期场景",
"缺少账号锁定后尝试登录场景"
]
},
"准确性": {
"score": 17,
"comment": "大部分预期结果具体可验证...",
"issues": [
"TC-LOGIN-003 的预期结果'显示错误提示'不够具体，应改为'页面输入框下方显示红色文字：手机号格式不正确'"
]
},
"可执行性": {
"score": 16,
"comment": "步骤基本清晰...",
"issues": []
},
"去重与独立性": {
"score": 14,
"comment": "发现以下重复用例...",
"issues": [
"TC-LOGIN-001 和 TC-LOGIN-003 步骤高度相似，建议合并"
]
},
"优先级合理性": {
"score": 14,
"comment": "P0 覆盖了核心流程...",
"issues": [
"TC-LOGIN-008（帮助页面跳转）标为P0不合理，建议改为P2"
]
}
},
"pass_status": false,
"feedback": "总体用例质量中等，主要问题在于反向场景覆盖不足和个别优先级错标...",
"suggestions": [
"建议1: 补充验证码过期、验证码错误、账号锁定后登录等反向场景",
"建议2: TC-LOGIN-003 的预期结果改为具体描述",
"建议3: 合并 TC-LOGIN-001 和 TC-LOGIN-003",
"建议4: TC-LOGIN-008 优先级从 P0 改为 P2"
]
}
## 约束
- 打分要严格，宁可打低不要放水
- issues 列表要具体到用例编号（如 TC-LOGIN-003），不要泛泛而谈
- suggestions 必须可直接操作（如"补充XXX场景"、"将XXX改为XXX"）
- 每个维度的 comment 至少写30字
"""
Prompt 设计要点：
| 要素 | 说明 |
|---|---|
| 对比评审 | “请基于以下提供的【测试计划】和【测试用例】进行对比评审”——明确交叉对比的两份输入 |
| 评分细化 | 每个维度不只有分数，还要求 issues 列表——把问题定位到具体用例 |
| 通过标准前置 | 放在评分维度之后、输出格式之前，逻辑顺序清晰 |
| JSON 格式 | 3层嵌套 JSON（score → detail → issues），去掉 ```json 标记避免字符串截断 |
| 判断标准 | “每缺少一类场景扣5分”——给模型一个可量化的扣分规则 |
Agent 7：审核输出员
职责：对最终用例做格式校验、合规检查、质量抽检，整理为标准化交付文档。
FINAL_OUTPUT_PROMPT = """
你是一名测试输出审核员，负责对测试用例做最终的格式校验和质量把关。
## 你的任务
### 1. 格式校验
- 用例编号是否连续（如 TC-LOGIN-001, TC-LOGIN-002, TC-LOGIN-003）
- 有无编号跳号或重复
- 字段名称是否统一
- 表格格式是否对齐
### 2. 合规检查
- 每条用例的必填字段是否都有值（不允许空值或"待定"）
- 优先级是否为 P0/P1/P2 之一
- 步骤数量是否 >= 3
- 预期结果是否具体（不含"正常""正确"等模糊词）
### 3. 质量抽检
- 随机抽取 3 条用例，逐字段检查是否符合执行标准
- 如果抽检发现问题，标注具体问题
### 4. 整理输出
将用例整理为一份标准化的测试文档，格式如下：
# 测试用例文档
## 文档信息
- **项目名称**：XXX
- **版本号**：V1.0
- **编写日期**：YYYY-MM-DD
- **编写人**：AI 测试助手
## 目录
1. 测试范围概述
2. 用例统计
3. 测试用例详情
- 3.1 模块一
- 3.2 模块二
- ...
## 1. 测试范围概述
（从测试计划中提取摘要）
## 2. 用例统计
| 优先级 | 数量 | 占比 |
|--------|------|------|
| P0 | XX | XX% |
| P1 | XX | XX% |
| P2 | XX | XX% |
| **合计** | **XX** | **100%** |
## 3. 测试用例详情
### 3.1 模块一：XXX
| 用例编号 | 标题 | 优先级 | 前置条件 | 操作步骤 | 预期结果 | 测试数据 |
|----------|------|--------|----------|----------|----------|----------|
| TC-XXX-001 | ... | P0 | ... | ... | ... | ... |
（后续模块同上格式）
## 约束
- 格式必须整齐统一，可直接作为交付物
- 如有格式问题，在输出前自行修正（如重排编号、统一字段名）
- 统计数据要准确（数量和占比要对得上）
"""
Prompt 设计要点：
| 要素 | 说明 |
|---|---|
| 四步任务 | 格式校验 → 合规检查 → 质量抽检 → 整理输出，层层递进 |
| 文档模板 | 给出完整的文档结构模板，模型直接填充内容 |
| 自修正 | “如有格式问题，在输出前自行修正”——让审核员同时也是修复员 |
| 统计校验 | “数量和占比要对得上”——防止模型算错数 |
七个 Agent 的协作关系总览
┌──────────────────────────────────────────────────────────────────────┐
│ │
│ Agent 1 Agent 2 Agent 3 │
│ 需求分析师 ──输出──→ 需求评审员 ──通过──→ 测试计划师 │
│ (分析需求) (审核需求) (制定计划) │
│ ▲ │ │
│ 不通过+反馈 │ │
│ │
│ Agent 4 Agent 5 Agent 6 │
│ 计划评审员 用例设计师 ──输出──→ 用例评审员 │
│ (打分+反馈) (生成用例) (打分+反馈) │
│ ▲ │ │ │
│ 不通过+反馈 │ │
│ 打回Agent3 不通过时 │ │
│ └────┘ │
│ │
│ Agent 7 │
│ 审核输出员 │
│ (格式校验+输出) │
│ │
└──────────────────────────────────────────────────────────────────────┘
| Agent | 流水线顺序 | 输入来源声明 | 输出 | 关键 Prompt 技巧 |
|---|---|---|---|---|
| ① 需求分析师 | 第1步 | 用户需求文本 | 结构化分析报告 | 步骤拆解 + 表格模板 |
| ② 需求评审员 | 第2步 | “基于以下提供的【原始需求文档】和【需求分析报告】” | JSON 评分 + 反馈 | 对比评审 + 过度分析检测 + 错误修正要求 |
| ③ 测试计划师 | 第3步 | “基于以下提供的【需求分析报告】” | 测试计划 | 针对性约束 + 排除项要求 |
| ④ 计划评审员 | 第4步 | “基于以下提供的【需求分析报告】和【测试计划】” | JSON 评分 + 反馈 | 量化打分 + 角色对抗 + 通过标准前置 |
| ⑤ 用例设计师 | 第5步 | “基于以下提供的【测试计划】” | 测试用例集 | 覆盖原则 + 负面约束 |
| ⑥ 用例评审员 | 第6步 | “基于以下提供的【测试计划】和【测试用例】” | JSON 评分 + 反馈 | 3层嵌套 JSON + issues 定位 |
| ⑦ 审核输出员 | 第7步 | 测试用例集 | 标准化文档 | 文档模板 + 自修正指令 |
3.4 流水线核心逻辑：含打回重做机制
class TestPipeline:
"""多 Agent 测试流水线（含需求分析审核）
Agent 按流水线顺序初始化：
①需求分析师 → ②需求评审员 → ③测试计划师 → ④计划评审员
→ ⑤用例设计师 → ⑥用例评审员 → ⑦审核输出员
"""
def __init__(self, max_retries: int = 3,
enable_analysis_review: bool = True):
self.max_retries = max_retries
self.enable_analysis_review = enable_analysis_review
# 初始化 7 个 Agent（按流水线顺序）
self.requirement_agent = TestAgent(
"①需求分析师", REQUIREMENT_ANALYSIS_PROMPT
)
self.analysis_review_agent = TestAgent(
"②需求评审员", ANALYSIS_REVIEW_PROMPT
)
self.plan_agent = TestAgent(
"③测试计划师", TEST_PLAN_PROMPT
)
self.plan_review_agent = TestAgent(
"④计划评审员", PLAN_REVIEW_PROMPT
)
self.case_agent = TestAgent(
"⑤用例设计师", TEST_CASE_PROMPT
)
self.case_review_agent = TestAgent(
"⑥用例评审员", CASE_REVIEW_PROMPT
)
self.final_agent = TestAgent(
"⑦审核输出员", FINAL_OUTPUT_PROMPT
)
def _run_with_review(self, generator: TestAgent,
reviewer: TestAgent,
input_data: str,
stage_name: str,
review_with_original: bool = False,
original_input: str = None) -> str:
"""
通用的 生成-评审 循环
生成 → 评审 → 不通过就带着反馈重新生成
异常处理策略：
- LLM 调用异常（限流、超时等）→ 由 call_llm_with_retry 处理重试
- JSON 解析异常 → 由 call_llm_json_with_retry 处理重试
- 以上全部失败 → 返回兜底结果，不中断流水线
- 业务级异常（评审超限）→ 返回最后版本 + 警告日志
Args:
review_with_original: 是否需要把原始输入也传给评审 Agent
original_input: 评审时需要对照的原始输入
（需求分析审核时为原始需求文档）
"""
current_input = input_data
for attempt in range(1, self.max_retries + 1):
logger.info(f"\n--- {stage_name} 第 {attempt} 轮 ---")
# 生成
try:
output = generator.run(current_input)
except Exception as e:
logger.error(f"[{stage_name}] 生成环节异常: {type(e).__name__}: {e}")
if attempt >= self.max_retries:
raise
continue
# 评审：如果有原始输入，拼接后一起传给评审 Agent
if review_with_original and original_input:
review_input = (
f"--- 原始需求文档 ---\n{original_input}\n\n"
f"--- 需求分析报告 ---\n{output}"
)
else:
review_input = output
review = reviewer.run_and_parse_json(review_input)
score = review.get("score", 0)
passed = review.get("pass_status", False)
feedback = review.get("feedback", "")
suggestions = review.get("suggestions", [])
logger.info(
f"[评审结果] 分数: {score} | 通过: {passed}"
)
logger.info(f"[反馈] {feedback}")
if passed:
logger.info(f"✅ {stage_name} 评审通过！")
return output
# 未通过：把评审反馈拼接到输入中，重新生成
logger.info(f"❌ {stage_name} 评审未通过，准备重做...")
current_input = (
f"{input_data}\n\n"
f"--- 上一版本评审反馈（请据此修改）---\n"
f"分数：{score}/100\n"
f"反馈：{feedback}\n"
f"改进建议：\n"
)
for i, s in enumerate(suggestions, 1):
current_input += f" {i}. {s}\n"
# 超过最大重试次数
logger.warning(
f"⚠️ {stage_name} 经过 {self.max_retries} 轮仍未通过，使用最后版本"
)
return output
def run(self, requirement: str) -> str:
"""执行完整流水线"""
logger.info("=" * 60)
logger.info("🚀 测试流水线启动")
logger.info("=" * 60)
# ============ Step 1: 需求分析 ============
logger.info("📋 Step 1: 需求分析")
analysis = self.requirement_agent.run(requirement)
# ============ Step 2: 需求分析审核（可选） ============
if self.enable_analysis_review:
logger.info("🔍 Step 2: 需求分析审核")
analysis = self._run_with_review(
generator=self.requirement_agent,
reviewer=self.analysis_review_agent,
input_data=requirement,
stage_name="需求分析",
review_with_original=True,
original_input=requirement
)
# ============ Step 3 + 4: 测试计划 + 评审 ============
logger.info("📝 Step 3-4: 测试计划制定 + 评审")
plan = self._run_with_review(
generator=self.plan_agent,
reviewer=self.plan_review_agent,
input_data=analysis,
stage_name="测试计划"
)
# ============ Step 5 + 6: 用例生成 + 评审 ============
logger.info("🧪 Step 5-6: 测试用例生成 + 评审")
cases = self._run_with_review(
generator=self.case_agent,
reviewer=self.case_review_agent,
input_data=plan,
stage_name="测试用例"
)
# ============ Step 7: 审核输出 ============
logger.info("✅ Step 7: 最终审核输出")
final_output = self.final_agent.run(cases)
logger.info("=" * 60)
logger.info("🎉 流水线执行完成！")
logger.info("=" * 60)
return final_output
3.5 运行入口
if __name__ == "__main__":
user_requirement = """
【需求描述】电商系统 - 用户登录功能
1. 用户通过手机号 + 密码登录
2. 密码错误3次后账号锁定30分钟
3. 支持"记住密码"功能（7天免登录）
4. 登录成功后跳转到首页
5. 需要图形验证码
"""
# 方案一：不加需求审核（快速模式）
# pipeline = TestPipeline(max_retries=3, enable_analysis_review=False)
# 方案二：加需求审核（推荐）
pipeline = TestPipeline(max_retries=3, enable_analysis_review=True)
result = pipeline.run(user_requirement)
# 保存结果
with open("test_output.md", "w", encoding="utf-8") as f:
f.write(result)
print("结果已保存到 test_output.md")
四、使用 LangChain LCEL 构建管道式链
上面是手写循环，LangChain 提供了 LCEL（LangChain Expression Language） 可以用 | 管道符串联。
4.1 基础链式调用
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
# 定义提示词模板
requirement_prompt = ChatPromptTemplate.from_messages([
("system", REQUIREMENT_ANALYSIS_PROMPT),
("human", "{input}")
])
plan_prompt = ChatPromptTemplate.from_messages([
("system", TEST_PLAN_PROMPT),
("human", "{input}")
])
case_prompt = ChatPromptTemplate.from_messages([
("system", TEST_CASE_PROMPT),
("human", "{input}")
])
# 创建模型
llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
parser = StrOutputParser()
# 用 | 串联成链
requirement_chain = requirement_prompt | llm | parser
plan_chain = plan_prompt | llm | parser
case_chain = case_prompt | llm | parser
# 顺序执行
analysis = requirement_chain.invoke({"input": user_requirement})
plan = plan_chain.invoke({"input": analysis})
cases = case_chain.invoke({"input": plan})
4.2 加入评审子链
from langchain_core.runnables import RunnableLambda
def parse_json_safe_with_retry(raw: str, agent_name: str = "LCEL",
max_retries: int = 2,
re_call_fn=None,
original_input: str = None) -> dict:
"""带重试 + 必要字段校验的 JSON 解析。
与 5.2 节 call_llm_json_with_retry 功能对齐：
1. 先尝试解析 JSON
2. 解析失败则通过 re_call_fn 重新调用 LLM（追加提示词）
3. 校验必要字段（score、pass_status）
4. 全部失败返回兜底结构
Args:
raw: LLM 原始输出
agent_name: Agent 名称（用于日志）
max_retries: JSON 解析最大重试次数
re_call_fn: 重新调用 LLM 的函数（可选，传入则启用重调）
original_input: 重调时的原始输入（可选）
"""
last_raw = raw
for attempt in range(max_retries):
try:
# 提取 JSON 部分
cleaned = last_raw
if "```json" in cleaned:
cleaned = cleaned.split("```json")[1].split("```")[0]
elif "```" in cleaned:
cleaned = cleaned.split("```")[1].split("```")[0]
parsed = json.loads(cleaned.strip())
# 校验必要字段
required_fields = ["score", "pass_status"]
missing = [f for f in required_fields if f not in parsed]
if missing:
logger.warning(
f"[{agent_name}] JSON 缺少字段 {missing}，"
f"重试 ({attempt+1}/{max_retries})"
)
if re_call_fn and original_input:
last_raw = re_call_fn(
original_input
+ f"\n\n【重要提醒】上次输出缺少字段 {missing}，"
f"请确保输出包含 score、pass_status、feedback、suggestions。"
)
continue
else:
# 没有重调函数，返回兜底
break
return parsed
except json.JSONDecodeError:
logger.warning(
f"[{agent_name}] JSON 解析失败，"
f"重试 ({attempt+1}/{max_retries})"
)
if re_call_fn and original_input:
last_raw = re_call_fn(
original_input
+ "\n\n【重要提醒】上次输出无法解析为 JSON，"
"请务必只输出合法的 JSON，不要包含任何额外文字。"
)
else:
break
# 全部失败，返回兜底结构
return {
"score": 0,
"pass_status": False,
"feedback": f"JSON 解析失败，原始输出前500字：{last_raw[:500]}",
"suggestions": ["请检查 Prompt 是否要求了严格的 JSON 格式"]
}
def generate_and_review(generate_chain, review_prompt_template,
input_data, threshold=80, max_retry=3):
"""带评审循环的执行逻辑（与 3.5 节 _run_with_review 对齐）"""
review_chain = review_prompt_template | llm | parser
current_input = input_data
for i in range(max_retry):
output = generate_chain.invoke({"input": current_input})
# 带重试的 JSON 解析
def re_call_llm(prompt_text):
return review_chain.invoke({"input": prompt_text})
review = parse_json_safe_with_retry(
raw=review_chain.invoke({"input": output}),
agent_name=f"评审(第{i+1}轮)",
max_retries=2,
re_call_fn=re_call_llm,
original_input=output
)
if review.get("pass_status") or review.get("score", 0) >= threshold:
return output
# 拼接反馈重试
current_input = (
f"{input_data}\n\n"
f"--- 评审反馈（第{i+1}轮）---\n"
f"分数：{review.get('score', 0)}/100\n"
f"反馈：{review.get('feedback', '')}\n"
f"改进建议：{review.get('suggestions', [])}"
)
return output # 返回最后一版
# 使用 RunnableLambda 包装成 LangChain 链节点
analysis_review_prompt = ChatPromptTemplate.from_messages([
("system", ANALYSIS_REVIEW_PROMPT),
("human", "{input}")
])
plan_review_prompt = ChatPromptTemplate.from_messages([
("system", PLAN_REVIEW_PROMPT),
("human", "{input}")
])
case_review_prompt = ChatPromptTemplate.from_messages([
("system", CASE_REVIEW_PROMPT),
("human", "{input}")
])
final_prompt = ChatPromptTemplate.from_messages([
("system", FINAL_OUTPUT_PROMPT),
("human", "{input}")
])
final_chain = final_prompt | llm | parser
# Step 1-2: 需求分析 + 审核
analysis_with_review = RunnableLambda(
lambda x: generate_and_review(
requirement_chain, analysis_review_prompt, x
)
)
# Step 3-4: 测试计划 + 审核
plan_with_review = RunnableLambda(
lambda x: generate_and_review(plan_chain, plan_review_prompt, x)
)
# Step 5-6: 测试用例 + 审核
case_with_review = RunnableLambda(
lambda x: generate_and_review(case_chain, case_review_prompt, x)
)
# 完整流水线
full_pipeline = (
analysis_with_review
| plan_with_review
| case_with_review
| final_chain
)
result = full_pipeline.invoke({"input": user_requirement})
五、使用 LangGraph 实现完整的图编排
LangChain 官方推荐用 LangGraph 处理多 Agent 有向图流程，支持条件分支、循环、并行、检查点持久化、人工审核节点。这是整个流水线最完整的实现方式。
5.1 状态定义
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
import operator
class PipelineState(TypedDict):
"""流水线全局状态
字段按流水线顺序组织：
Step 1-2: 需求分析 + 审核
Step 3-4: 测试计划 + 审核
Step 5-6: 测试用例 + 审核
Step 7: 人工审核
Step 8: 最终输出
"""
# 输入
requirement: str
# Step 1-2: 需求分析 + 审核
analysis: str
analysis_review: dict
analysis_retry_count: int
# Step 3-4: 测试计划 + 审核
plan: str
plan_review: dict
plan_retry_count: int
# Step 5-6: 测试用例 + 审核
cases: str
case_review: dict
case_retry_count: int
# Step 7: 人工审核相关
human_approved: bool
human_feedback: str
# Step 8: 最终输出
final_output: str
# 日志追踪
logs: Annotated[list[str], operator.add] # 累积追加，不会覆盖
# 异常追踪
error: str # 记录节点级异常，供路由判断
Annotated[list[str], operator.add]让logs字段在每次节点更新时追加而非覆盖，适合做全链路追踪。
5.2 所有节点函数（含容错）
from openai import (
RateLimitError,
APITimeoutError,
BadRequestError,
APIConnectionError,
AuthenticationError
)
# ============================================================
# 常量定义
# ============================================================
MAX_ANALYSIS_RETRY = 3
MAX_PLAN_RETRY = 3
MAX_CASE_RETRY = 3
LLM_MAX_RETRIES = 3
LLM_TIMEOUT = 120 # 单次调用超时（秒）
PIPELINE_TIMEOUT = 600 # 整条流水线超时（秒）
# ============================================================
# 生产级 LLM 调用（带完整异常处理）
# ============================================================
def call_llm_with_retry(agent: TestAgent, input_text: str,
max_retries: int = LLM_MAX_RETRIES) -> str:
"""带指数退避的 LLM 调用，覆盖所有常见异常。
异常处理分层：
- 可重试（限流/超时/网络）→ 指数退避后重试
- 可恢复（上下文溢出）→ 截断输入后重试
- 不可重试（认证失败/内容过滤）→ 直接抛出
"""
last_error = None
for attempt in range(1, max_retries + 1):
try:
return agent.run(input_text)
except RateLimitError as e:
wait = 2 ** attempt
logger.warning(
f"[{agent.name}] API 限流(429)，{wait}s 后重试 "
f"({attempt}/{max_retries})"
)
time.sleep(wait)
last_error = e
except APITimeoutError as e:
wait = 2 ** attempt
logger.warning(
f"[{agent.name}] 请求超时({LLM_TIMEOUT}s)，{wait}s 后重试 "
f"({attempt}/{max_retries})"
)
time.sleep(wait)
last_error = e
except APIConnectionError as e:
wait = 2 ** attempt
logger.warning(
f"[{agent.name}] 网络连接失败，{wait}s 后重试 "
f"({attempt}/{max_retries})"
)
time.sleep(wait)
last_error = e
except BadRequestError as e:
error_msg = str(e)
# 上下文窗口溢出：输入太长
if "context length" in error_msg.lower() or "maximum" in error_msg.lower():
logger.error(
f"[{agent.name}] 输入超过上下文窗口限制({len(input_text)}字)。"
f"请缩短输入或启用上下文压缩。"
)
# 尝试截断输入后重试
if attempt < max_retries:
truncation_point = int(len(input_text) * 0.7)
input_text = input_text[:truncation_point] + (
"\n\n... [输入过长，已截断至70%] ..."
)
logger.warning(f"[{agent.name}] 已截断输入至 {len(input_text)} 字，重试")
continue
# 内容被安全过滤
elif "content_policy" in error_msg.lower() or "moderation" in error_msg.lower():
logger.error(
f"[{agent.name}] 内容被安全过滤器拦截，请检查输入内容。"
)
raise # 不可重试，直接抛出
else:
logger.error(f"[{agent.name}] 请求参数错误(400): {e}")
raise # 不可重试
last_error = e
except AuthenticationError as e:
logger.error(
f"[{agent.name}] API Key 认证失败，请检查密钥配置。"
)
raise # 不可重试，直接终止
except Exception as e:
logger.error(
f"[{agent.name}] 未知错误: {type(e).__name__}: {e}"
)
last_error = e
if attempt >= max_retries:
break
wait = 2 ** attempt
time.sleep(wait)
raise RuntimeError(
f"[{agent.name}] 调用失败，已重试 {max_retries} 次。"
f"最后错误: {type(last_error).__name__}: {last_error}"
)
def call_llm_json_with_retry(agent: TestAgent, input_text: str,
max_retries: int = LLM_MAX_RETRIES) -> dict:
"""带重试 + JSON 解析容错的调用"""
last_raw = ""
for attempt in range(max_retries):
raw = call_llm_with_retry(agent, input_text, max_retries=2)
last_raw = raw
try:
if "```json" in raw:
raw = raw.split("```json")[1].split("```")[0]
elif "```" in raw:
raw = raw.split("```")[1].split("```")[0]
parsed = json.loads(raw.strip())
# 校验必要字段是否存在
required_fields = ["score", "pass_status"]
missing = [f for f in required_fields if f not in parsed]
if missing:
logger.warning(
f"[{agent.name}] JSON 缺少必要字段: {missing}，重试"
)
input_text += (
f"\n\n【重要提醒】上次输出缺少字段 {missing}，"
f"请确保输出包含 score、pass_status、feedback、suggestions 字段。"
)
continue
return parsed
except json.JSONDecodeError:
logger.warning(
f"[{agent.name}] JSON 解析失败，重试 ({attempt+1}/{max_retries})"
)
input_text = input_text + (
"\n\n【重要提醒】上次输出无法解析为 JSON，"
"请务必只输出合法的 JSON，不要包含任何额外文字。"
)
# 全部失败，返回兜底结构
return {
"score": 0,
"pass_status": False,
"feedback": f"JSON 解析 {max_retries} 次均失败，原始输出前500字：{last_raw[:500]}",
"suggestions": [
"请检查 Prompt 是否要求了严格的 JSON 格式",
"考虑在 system prompt 中强化 JSON-only 输出要求"
]
}
# ============================================================
# 节点级状态校验
# ============================================================
def validate_state_field(state: dict, field: str,
field_type=str, node_name: str = "") -> bool:
"""校验状态字段是否存在且类型正确"""
value = state.get(field)
if value is None:
logger.error(f"[{node_name}] 状态字段 '{field}' 为 None")
return False
if not isinstance(value, field_type):
logger.error(
f"[{node_name}] 状态字段 '{field}' 类型错误: "
f"期望 {field_type.__name__}，实际 {type(value).__name__}"
)
return False
if field_type == str and len(value.strip()) == 0:
logger.error(f"[{node_name}] 状态字段 '{field}' 为空字符串")
return False
return True
# ============================================================
# Step 1: 需求分析
# ============================================================
def node_analyze(state: PipelineState) -> dict:
if not validate_state_field(state, "requirement", str, "Step1-需求分析"):
return {
"analysis": "",
"error": "输入的需求文档为空",
"logs": ["[Step1] ❌ 需求分析失败：输入为空"]
}
result = call_llm_with_retry(requirement_agent, state["requirement"])
return {
"analysis": result,
"error": "",
"logs": ["[Step1] 需求分析完成"]
}
# ============================================================
# Step 2: 需求分析评审
# ============================================================
def node_review_analysis(state: PipelineState) -> dict:
if not validate_state_field(state, "analysis", str, "Step2-需求评审"):
return {
"analysis_review": {
"score": 0, "pass_status": False,
"feedback": "需求分析报告为空，无法评审"
},
"error": "需求分析报告为空",
"logs": ["[Step2] ❌ 需求评审跳过：分析报告为空"]
}
review_input = (
f"--- 原始需求文档 ---\n{state['requirement']}\n\n"
f"--- 需求分析报告 ---\n{state['analysis']}"
)
result = call_llm_json_with_retry(analysis_review_agent, review_input)
score = result.get("score", 0)
passed = result.get("pass_status", False)
return {
"analysis_review": result,
"error": "",
"logs": [f"[Step2] 需求分析评审完成 | 分数:{score} | 通过:{passed}"]
}
# ============================================================
# Step 3: 制定测试计划
# ============================================================
def node_plan(state: PipelineState) -> dict:
if not validate_state_field(state, "analysis", str, "Step3-测试计划"):
return {
"plan": "",
"error": "需求分析报告为空，无法制定计划",
"logs": ["[Step3] ❌ 测试计划失败：分析报告为空"]
}
retry_count = state.get("plan_retry_count", 0) + 1
input_text = state["analysis"]
if state.get("plan_review") and not state["plan_review"].get("pass_status"):
feedback = state["plan_review"].get("feedback", "")
suggestions = state["plan_review"].get("suggestions", [])
input_text += (
f"\n\n--- 评审反馈（第{retry_count-1}轮，请据此修改）---\n"
f"分数：{state['plan_review'].get('score', 0)}/100\n"
f"反馈：{feedback}\n"
f"改进建议：\n"
)
for i, s in enumerate(suggestions, 1):
input_text += f" {i}. {s}\n"
result = call_llm_with_retry(plan_agent, input_text)
return {
"plan": result,
"plan_retry_count": retry_count,
"error": "",
"logs": [f"[Step3] 测试计划制定完成（第{retry_count}轮）"]
}
# ============================================================
# Step 4: 评审测试计划
# ============================================================
def node_review_plan(state: PipelineState) -> dict:
if not validate_state_field(state, "plan", str, "Step4-计划评审"):
return {
"plan_review": {
"score": 0, "pass_status": False,
"feedback": "测试计划为空，无法评审"
},
"error": "测试计划为空",
"logs": ["[Step4] ❌ 计划评审跳过：计划为空"]
}
result = call_llm_json_with_retry(plan_review_agent, state["plan"])
score = result.get("score", 0)
passed = result.get("pass_status", False)
return {
"plan_review": result,
"error": "",
"logs": [f"[Step4] 计划评审完成 | 分数:{score} | 通过:{passed}"]
}
# ============================================================
# Step 5: 生成测试用例
# ============================================================
def node_generate_cases(state: PipelineState) -> dict:
if not validate_state_field(state, "plan", str, "Step5-用例生成"):
return {
"cases": "",
"error": "测试计划为空，无法生成用例",
"logs": ["[Step5] ❌ 用例生成失败：计划为空"]
}
retry_count = state.get("case_retry_count", 0) + 1
input_text = state["plan"]
if state.get("case_review") and not state["case_review"].get("pass_status"):
feedback = state["case_review"].get("feedback", "")
suggestions = state["case_review"].get("suggestions", [])
input_text += (
f"\n\n--- 评审反馈（第{retry_count-1}轮，请据此修改）---\n"
f"分数：{state['case_review'].get('score', 0)}/100\n"
f"反馈：{feedback}\n"
f"改进建议：\n"
)
for i, s in enumerate(suggestions, 1):
input_text += f" {i}. {s}\n"
result = call_llm_with_retry(case_agent, input_text)
return {
"cases": result,
"case_retry_count": retry_count,
"error": "",
"logs": [f"[Step5] 测试用例生成完成（第{retry_count}轮）"]
}
# ============================================================
# Step 6: 评审测试用例
# ============================================================
def node_review_cases(state: PipelineState) -> dict:
if not validate_state_field(state, "cases", str, "Step6-用例评审"):
return {
"case_review": {
"score": 0, "pass_status": False,
"feedback": "测试用例为空，无法评审"
},
"error": "测试用例为空",
"logs": ["[Step6] ❌ 用例评审跳过：用例为空"]
}
result = call_llm_json_with_retry(case_review_agent, state["cases"])
score = result.get("score", 0)
passed = result.get("pass_status", False)
return {
"case_review": result,
"error": "",
"logs": [f"[Step6] 用例评审完成 | 分数:{score} | 通过:{passed}"]
}
# ============================================================
# Step 7: 人工审核（Human-in-the-Loop）
# ============================================================
def node_human_review(state: PipelineState) -> dict:
"""
LangGraph 的 interrupt 机制会在这里暂停，
把当前状态输出等待人工输入，然后恢复执行。
这里用 input() 模拟，生产中可对接 Slack / 钉钉 / 邮件。
"""
print("\n" + "=" * 60)
print("⏸️ [人工审核节点] 流水线暂停，等待人工确认")
print("=" * 60)
# 展示预览内容（可能来自不同阶段）
preview = state.get("cases", "") or state.get("analysis", "")
print("\n--- 内容预览（前3000字）---\n")
print(preview[:3000])
while True:
try:
choice = input(
"\n请审核：[y] 通过 / [f] 打回并输入反馈 / [q] 终止: "
).strip().lower()
except EOFError:
# 非交互环境（如 CI/CD），默认通过
logger.warning("[人工审核] 非交互环境，自动通过")
return {
"human_approved": True,
"human_feedback": "",
"logs": ["[Step7] 人工审核（非交互环境）自动通过"]
}
if choice == "y":
return {
"human_approved": True,
"human_feedback": "",
"logs": ["[Step7] 人工审核通过"]
}
elif choice == "f":
try:
feedback = input("请输入修改意见：").strip()
except EOFError:
feedback = "人工审核打回（无具体反馈）"
return {
"human_approved": False,
"human_feedback": feedback,
"logs": [f"[Step7] 人工打回，反馈：{feedback}"]
}
elif choice == "q":
raise KeyboardInterrupt("人工终止流水线")
else:
print("无效输入，请输入 y / f / q")
# ============================================================
# Step 8: 最终输出
# ============================================================
def node_final_output(state: PipelineState) -> dict:
if not validate_state_field(state, "cases", str, "Step8-最终输出"):
return {
"final_output": "⚠️ 流水线未能生成有效测试用例，请检查上游环节。",
"error": "用例为空",
"logs": ["[Step8] ❌ 最终输出失败：用例为空"]
}
result = call_llm_with_retry(final_agent, state["cases"])
return {
"final_output": result,
"error": "",
"logs": ["[Step8] 最终审核输出完成"]
}
5.3 条件路由函数
def route_analysis_review(state: PipelineState) -> str:
"""Step 2 评审后：通过 → 制定计划，不通过 → 重做或人工介入"""
# 检查上游是否出错
if state.get("error") and not state.get("analysis"):
return "human_escalation"
review = state.get("analysis_review", {})
score = review.get("score", 0)
passed = review.get("pass_status", False)
retry_count = state.get("analysis_retry_count", 0)
if passed or score >= 80:
return "pass"
if retry_count >= MAX_ANALYSIS_RETRY:
logger.warning(
f"⚠️ 需求分析连续 {MAX_ANALYSIS_RETRY} 轮未通过，转人工审核"
)
return "human_escalation"
return "retry"
def route_plan_review(state: PipelineState) -> str:
"""Step 4 评审后：通过 → 生成用例，不通过 → 重做或人工介入"""
if state.get("error") and not state.get("plan"):
return "human_escalation"
review = state.get("plan_review", {})
score = review.get("score", 0)
passed = review.get("pass_status", False)
retry_count = state.get("plan_retry_count", 0)
if passed or score >= 80:
return "pass"
if retry_count >= MAX_PLAN_RETRY:
logger.warning(
f"⚠️ 计划连续 {MAX_PLAN_RETRY} 轮未通过，转人工审核"
)
return "human_escalation"
return "retry"
def route_case_review(state: PipelineState) -> str:
"""Step 6 评审后：通过 → 人工确认，不通过 → 重做或人工介入"""
if state.get("error") and not state.get("cases"):
return "human_escalation"
review = state.get("case_review", {})
score = review.get("score", 0)
passed = review.get("pass_status", False)
retry_count = state.get("case_retry_count", 0)
if passed or score >= 80:
return "pass"
if retry_count >= MAX_CASE_RETRY:
logger.warning(
f"⚠️ 用例连续 {MAX_CASE_RETRY} 轮未通过，转人工审核"
)
return "human_escalation"
return "retry"
def route_human_review(state: PipelineState) -> str:
"""Step 7 人工审核后：通过 → 最终输出，打回 → 重新生成"""
if state.get("human_approved"):
return "approved"
return "rejected"
5.4 构建完整有向图
def build_pipeline() -> StateGraph:
graph = StateGraph(PipelineState)
# ========== 添加所有节点（按流水线顺序） ==========
graph.add_node("analyze", node_analyze) # Step 1
graph.add_node("review_analysis", node_review_analysis) # Step 2
graph.add_node("plan", node_plan) # Step 3
graph.add_node("review_plan", node_review_plan) # Step 4
graph.add_node("generate_cases", node_generate_cases) # Step 5
graph.add_node("review_cases", node_review_cases) # Step 6
graph.add_node("human_review", node_human_review) # Step 7
graph.add_node("final_output", node_final_output) # Step 8
# ========== 入口 ==========
graph.set_entry_point("analyze")
# ========== 固定边 ==========
graph.add_edge("analyze", "review_analysis")
graph.add_edge("generate_cases", "review_cases")
graph.add_edge("final_output", END)
# ========== 条件边：Step 2 需求分析评审 ==========
graph.add_conditional_edges(
"review_analysis",
route_analysis_review,
{
"pass": "plan", # 通过 → 制定计划
"retry": "analyze", # 重做 → 回到需求分析
"human_escalation": "human_review" # 超限 → 人工介入
}
)
# ========== 固定边 ==========
graph.add_edge("plan", "review_plan")
# ========== 条件边：Step 4 计划评审 ==========
graph.add_conditional_edges(
"review_plan",
route_plan_review,
{
"pass": "generate_cases", # 通过 → 生成用例
"retry": "plan", # 重做 → 回到计划
"human_escalation": "human_review" # 超限 → 人工介入
}
)
# ========== 条件边：Step 6 用例评审 ==========
graph.add_conditional_edges(
"review_cases",
route_case_review,
{
"pass": "human_review", # 通过 → 人工确认
"retry": "generate_cases", # 重做 → 回到生成
"human_escalation": "human_review" # 超限 → 人工介入
}
)
# ========== 条件边：Step 7 人工审核 ==========
graph.add_conditional_edges(
"human_review",
route_human_review,
{
"approved": "final_output", # 通过 → 最终输出
"rejected": "generate_cases" # 打回 → 重新生成
}
)
return graph
5.5 运行入口（含检查点持久化与人工审核）
from langgraph.checkpoint.memory import MemorySaver
def main():
user_requirement = """
【需求描述】电商系统 - 用户登录功能
1. 用户通过手机号 + 密码登录
2. 密码错误3次后账号锁定30分钟
3. 支持"记住密码"功能（7天免登录）
4. 登录成功后跳转到首页
5. 需要图形验证码
"""
# 构建图
graph = build_pipeline()
# ============================================================
# 检查点持久化：支持断点恢复
# 即使流程中断（网络断、进程挂），可以从上次的状态继续
# 生产中可用 SqliteSaver 或 PostgresSaver 替代
# ============================================================
checkpointer = MemorySaver()
app = graph.compile(
checkpointer=checkpointer,
# human_review 节点会在此中断，等待外部输入
interrupt_before=["human_review"]
)
# thread_id 用于标识一次完整流水线，支持多条流水线并行
config = {"configurable": {"thread_id": "test-run-001"}}
try:
result = app.invoke(
{
"requirement": user_requirement,
"analysis_retry_count": 0,
"plan_retry_count": 0,
"case_retry_count": 0,
"human_approved": False,
"human_feedback": "",
"error": "",
"logs": []
},
config=config
)
# 输出执行日志
print("\n" + "=" * 60)
print("📊 执行日志")
print("=" * 60)
for log in result.get("logs", []):
print(f" {log}")
# 检查是否有错误
if result.get("error"):
print(f"\n⚠️ 流水线存在未解决的错误: {result['error']}")
# 保存最终结果
with open("test_output.md", "w", encoding="utf-8") as f:
f.write(result.get("final_output", "无输出"))
print("\n✅ 结果已保存到 test_output.md")
except KeyboardInterrupt:
print("\n🛑 流水线被人工终止")
print("💡 提示：已保存检查点，下次运行可从断点恢复")
except Exception as e:
logger.error(f"流水线执行异常: {type(e).__name__}: {e}")
print(f"\n❌ 流水线执行失败: {e}")
print("💡 提示：已保存检查点，修复问题后可从断点恢复")
if __name__ == "__main__":
main()
5.6 完整流程图
┌──────────┐
│ 开始 │
└────┬─────┘
▼
┌──────────────┐
│ Step1 需求分析 │
└──────┬───────┘
▼
┌──────────────┐
│Step2 需求评审 │◄──────────────────┐
└──────┬───────┘ │
通过 │ 不通过 │ 反馈
│ ┌───────────────┘
│ │
│ ▼
│ ┌──────────────┐
│ │ 重试次数 < 3? │
│ └──────┬───────┘
│ 是 │ 否
│ │ │
│ 回到S1 人工介入 ──→ Step7
│
▼
┌──────────────────┐
│ Step3 制定测试计划 │◄──────────────────┐
└────────┬─────────┘ │
▼ │
┌──────────────────┐ ┌─────┴─────┐
│ Step4 评审测试计划 │── 不通过 ──→ │ 重试次数<3? │
└────────┬─────────┘ └─────┬─────┘
通过 │ 是 │ 否
│ ┌────────────────┘
│ │ │
│ ▼ ▼
│ 回到S3 ┌──────────┐
│ │ 人工介入 │
▼ └────┬─────┘
┌──────────────────┐ │
│ Step5 生成测试用例 │◄──────────────┘
└────────┬─────────┘ （打回时带反馈）
│ ▲
▼ │
┌──────────────────┐│
│ Step6 评审测试用例 ││── 不通过 ──→ 重试次数<3?
└────────┬─────────┘│ │是 │否
通过 │ 回到S5 回到S5 人工介入
│
▼
┌──────────────────┐
│ Step7 人工审核确认 │◄── interrupt_before（暂停点）
└────────┬─────────┘
│
┌────────┴────────┐
│ │
通过(✅) 打回(❌)
│ │
▼ ▼
┌──────────────┐ 回到S5 重新生成
│Step8 最终输出 │ （带人工反馈）
└──────┬───────┘
▼
┌──────────┐
│ 结束 │
└──────────┘
关键设计点：
- Step1-Step2 循环：需求分析 → 评审 → 不通过就带反馈重做，最多 3 次——地基先校准再往上搭
- Step3-Step4 循环：测试计划自动生成 + 评审打回，最多 3 次
- Step5-Step6 循环：测试用例自动生成 + 评审打回，最多 3 次
- Step7 人工节点：所有 AI 自动生成的内容都经过人工确认才输出
- 超限转人工：任何环节连续 3 次不通过，不再死循环，直接转人工介入
- 错误感知路由：每个路由函数先检查上游 error 字段，如果上游彻底失败则直接转人工，避免带着空数据继续跑
- 检查点持久化：流程中断后可从断点恢复，不丢失已生成的中间产物
六、完整异常处理体系
本章汇总整条流水线的异常处理设计，覆盖从 LLM 调用到流水线级别的全链路容错。
6.1 分层异常定义
class PipelineError(Exception):
"""流水线基础异常"""
pass
class LLMCallError(PipelineError):
"""LLM 调用失败（网络、超时、限流等）"""
pass
class JSONParseError(PipelineError):
"""LLM 返回内容无法解析为 JSON"""
pass
class ContextOverflowError(PipelineError):
"""输入超过模型上下文窗口限制"""
pass
class ContentFilterError(PipelineError):
"""内容被安全过滤器拦截"""
pass
class StateValidationError(PipelineError):
"""节点间状态传递异常（字段缺失或类型错误）"""
pass
class ReviewTimeoutError(PipelineError):
"""评审轮次超过最大重试次数"""
pass
class PipelineTimeoutError(PipelineError):
"""整条流水线执行超时"""
pass
异常分层关系：
PipelineError（基础异常）
├── LLMCallError → LLM 调用层
├── JSONParseError → 解析层
├── ContextOverflowError → LLM 调用层（输入过长）
├── ContentFilterError → LLM 调用层（内容安全）
├── StateValidationError → 节点间传递层
├── ReviewTimeoutError → 流程控制层
└── PipelineTimeoutError → 流水线级
6.2 LLM 调用层异常处理
| 异常类型 | HTTP 状态码 | 是否可重试 | 处理策略 |
|---|---|---|---|
| RateLimitError | 429 | ✅ 可重试 | 指数退避：1s → 2s → 4s |
| APITimeoutError | 超时 | ✅ 可重试 | 指数退避，检查网络 |
| APIConnectionError | 网络错误 | ✅ 可重试 | 指数退避，检查 DNS/代理 |
| BadRequestError(上下文溢出) | 400 | ✅ 截断后重试 | 截断输入至70%后重试 |
| BadRequestError(内容过滤) | 400 | ❌ 不可重试 | 直接抛出 ContentFilterError |
| BadRequestError(参数错误) | 400 | ❌ 不可重试 | 直接抛出，检查请求格式 |
| AuthenticationError | 401 | ❌ 不可重试 | 直接终止，检查 API Key |
| 未知异常 | - | ✅ 有限重试 | 最多重试3次后抛出 |
核心调用函数见 5.2 节的 call_llm_with_retry，此处补充一个流水线级超时保护：
import signal
class PipelineTimeout:
"""流水线级超时保护（仅 Unix 系统）
⚠️ 版本兼容说明：
ChatOpenAI.invoke 的 timeout 参数行为取决于 langchain-openai 版本：
- langchain-openai >= 0.2.0：支持 config={"timeout": 秒数}
- langchain-openai < 0.2.0：需要通过 request_timeout 参数或 httpx.Timeout 对象
建议锁定 langchain-openai >= 0.2.0 以使用本文的 timeout 写法。
"""
def __init__(self, timeout_seconds: int = PIPELINE_TIMEOUT):
self.timeout = timeout_seconds
def __enter__(self):
def timeout_handler(signum, frame):
raise PipelineTimeoutError(
f"流水线执行超过 {self.timeout}s 超时限制"
)
self.old_handler = signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(self.timeout)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
signal.alarm(0)
signal.signal(signal.SIGALRM, self.old_handler)
return False
# 使用方式
# with PipelineTimeout(600):
# result = app.invoke(initial_state, config=config)
6.3 节点间状态校验
每个节点在开始执行前，先校验上游输出是否有效：
# 每个节点函数的入口处统一加校验
def node_plan(state: PipelineState) -> dict:
# 校验上游输出
if not validate_state_field(state, "analysis", str, "Step3-测试计划"):
return {
"plan": "",
"error": "需求分析报告为空，无法制定计划",
"logs": ["[Step3] ❌ 测试计划失败：上游输出为空"]
}
# ... 正常逻辑
validate_state_field 函数检查三件事：
def validate_state_field(state: dict, field: str,
field_type=str, node_name: str = "") -> bool:
"""校验状态字段：1.是否存在 2.类型是否正确 3.是否为空"""
value = state.get(field)
if value is None: # 字段不存在
return False
if not isinstance(value, field_type): # 类型错误
return False
if field_type == str and len(value.strip()) == 0: # 空字符串
return False
return True
这样即使某个节点因异常返回了空值，下游节点不会带着空数据继续跑，而是立即标记错误并走降级路径。
6.4 流水线级容错策略
| 策略 | 说明 | 代码位置 |
|---|---|---|
| 节点级 try-except | 捕获节点异常，记录到 error 字段，不中断流水线 | safe_node 装饰器 |
| 错误感知路由 | 路由函数先检查 error 字段，上游失败直接转人工 | route_* 函数 |
| 评审超限转人工 | 自动重试 N 次后不再死循环，转人工介入 | route_* 函数 |
| 检查点持久化 | 流程中断后可从断点恢复，不丢失中间产物 | MemorySaver / SqliteSaver |
| 人工审核兜底 | 所有自动环节的最终出口都有人工确认 | node_human_review |
| 流水线超时 | 整条流水线执行超过阈值自动终止 | PipelineTimeout |
| 非交互环境降级 | CI/CD 中人工审核节点自动通过或跳过 | EOFError 捕获 |
safe_node 装饰器实现：
def safe_node(func):
"""节点执行的装饰器：捕获异常并记录到日志和状态"""
def wrapper(state: PipelineState) -> dict:
node_name = func.__name__
try:
logger.info(f"▶ 开始执行节点: {node_name}")
result = func(state)
logger.info(f"✅ 节点完成: {node_name}")
return result
except KeyboardInterrupt:
raise
except Exception as e:
error_msg = f"❌ 节点 {node_name} 执行失败: {type(e).__name__}: {e}"
logger.error(error_msg)
return {
"logs": [error_msg],
"error": str(e)
}
return wrapper
# 使用方式
# @safe_node
# def node_analyze(state: PipelineState) -> dict:
# ...
6.5 生产环境监控与进阶建议
运行指标采集
class PipelineMetrics:
"""流水线运行指标采集"""
@staticmethod
def record_run(thread_id: str, duration: float,
total_tokens: int, steps_completed: int,
final_score: float, error_count: int):
"""
每次流水线运行结束后记录：
- thread_id: 流水线实例 ID
- duration: 总耗时（秒）
- total_tokens: 消耗的总 token 数
- steps_completed: 完成的步骤数 / 总步骤数
- final_score: 最终评审得分
- error_count: 过程中发生的错误次数
可对接 Prometheus / Grafana / 企业自研监控平台
"""
logger.info(
f"[Metrics] thread={thread_id} | "
f"duration={duration:.1f}s | "
f"tokens={total_tokens} | "
f"steps={steps_completed}/8 | "
f"score={final_score} | "
f"errors={error_count}"
)
熔断器模式（Circuit Breaker）
当 LLM 服务长时间不可用时，连续重试会浪费时间和 Token。熔断器模式的核心思想：连续失败 N 次后，短路后续调用，直接返回降级结果。
class CircuitBreaker:
"""简易熔断器
状态机：CLOSED → OPEN → HALF_OPEN → CLOSED
CLOSED（正常）：请求正常通过
OPEN（熔断）：连续失败达到阈值，拒绝所有请求
HALF_OPEN（试探）：冷却时间过后，放行一次请求试探
"""
def __init__(self, failure_threshold: int = 5,
cooldown_seconds: int = 60):
self.failure_threshold = failure_threshold
self.cooldown_seconds = cooldown_seconds
self.failure_count = 0
self.last_failure_time = 0
self.state = "CLOSED" # CLOSED / OPEN / HALF_OPEN
def record_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = "OPEN"
logger.warning(
f"[CircuitBreaker] 熔断器打开：连续失败 {self.failure_count} 次，"
f"冷却 {self.cooldown_seconds}s"
)
def record_success(self):
self.failure_count = 0
self.state = "CLOSED"
def allow_request(self) -> bool:
if self.state == "CLOSED":
return True
if self.state == "OPEN":
elapsed = time.time() - self.last_failure_time
if elapsed >= self.cooldown_seconds:
self.state = "HALF_OPEN"
logger.info("[CircuitBreaker] 进入半开状态，放行试探请求")
return True
return False
if self.state == "HALF_OPEN":
return True
return False
# 使用方式
# breaker = CircuitBreaker(failure_threshold=5, cooldown_seconds=60)
#
# def node_with_breaker(state):
# if not breaker.allow_request():
# logger.error("[节点] 熔断器打开，跳过本次调用")
# return {"error": "LLM 服务熔断中", "logs": ["熔断器跳过"]}
# try:
# result = call_llm_with_retry(agent, input)
# breaker.record_success()
# return result
# except Exception as e:
# breaker.record_failure()
# raise
适用场景：当 LLM API 提供商出现大面积故障时（如 OpenAI 服务中断），熔断器可以在连续失败 5 次后直接跳过调用，等待 60 秒后再试探，而不是每轮都重试 3 次白白消耗时间和 token。
分层异常处理总览
┌──────────────────────────────────────────────────────────────┐
│ 异常处理分层架构 │
├──────────────┬───────────────────┬───────────────────────────┤
│ 层级 │ 机制 │ 覆盖场景 │
├──────────────┼───────────────────┼───────────────────────────┤
│ LLM 调用层 │ call_llm_with_ │ 限流/超时/网络→退避重试 │
│ │ retry │ 上下文溢出→截断重试 │
│ │ │ 认证失败→直接终止 │
│ │ │ 内容过滤→直接抛出 │
├──────────────┼───────────────────┼───────────────────────────┤
│ JSON 解析层 │ call_llm_json_ │ 解析失败→追加提示词重试 │
│ │ with_retry │ 字段缺失→追加提示词重试 │
│ │ │ 全部失败→返回兜底结构 │
├──────────────┼───────────────────┼───────────────────────────┤
│ 节点间传递层 │ validate_state_ │ 字段不存在→标记 error │
│ │ field │ 类型错误→标记 error │
│ │ │ 空字符串→标记 error │
├──────────────┼───────────────────┼───────────────────────────┤
│ 业务逻辑层 │ _run_with_review │ 评审不达标→带反馈重做 │
│ │ │ 重试超限→返回最后版本 │
│ │ safe_node 装饰器 │ 节点异常→记录日志+error │
├──────────────┼───────────────────┼───────────────────────────┤
│ 路由决策层 │ route_* 函数 │ 上游 error→转人工介入 │
│ │ │ 评审超限→转人工介入 │
├──────────────┼───────────────────┼───────────────────────────┤
│ 流水线级 │ PipelineTimeout │ 整体执行超时→强制终止 │
│ │ Checkpointer │ 流程中断→断点恢复 │
│ │ CircuitBreaker │ 服务持续不可用→熔断降级 │
├──────────────┼───────────────────┼───────────────────────────┤
│ 人工兜底 │ node_human_review │ 所有自动环节出口→人工确认 │
│ │ │ CI/CD 环境→自动降级通过 │
└──────────────┴───────────────────┴───────────────────────────┘
七、单 Agent vs 多 Agent 效果对比
7.1 需求分析的三种审查方案（如何选择）
需求分析是整条流水线的地基。地基歪了，上面全歪——需求分析阶段的错误会逐级放大：
功能点遗漏 1 个
→ 测试计划少覆盖 1 个模块
→ 测试用例少生成 10~20 条
→ 上线后该模块 0 覆盖 → 线上 bug
业务规则理解错误 1 条
→ 测试策略方向性错误
→ 用例预期结果全部写反
→ 评审打回重做，浪费大量 token 和时间
需求分析相比后续阶段有一个特殊风险：它没有上游文档可以对照。测试计划可以对照需求分析来评审，测试用例可以对照测试计划来评审，但需求分析面对的是原始需求文档——而原始需求本身就可能不完整、有歧义。
根据项目复杂度和风险等级，可以选择三种审查方案：
| 方案 | API 调用增加 | 质量提升 | 适用场景 |
|---|---|---|---|
| 不加需求审核 | 0 | 基准线 | 快速原型、需求很清晰、非核心功能 |
| 加单一审核 Agent | +2~5 次 | 明显提升 | 大多数正式项目，性价比最高 |
| 双视角分析 + 审核 | +5~10 次 | 显著提升 | 需求模糊、复杂业务、金融医疗等高风险项目 |
方案一：不加需求审核
原始需求 → [需求分析 Agent] → 直接进入测试计划
适合需求文档已经非常清晰、功能简单的场景。优点是速度快、成本低，但风险是需求理解错误无法被发现。
方案二：加单一审核 Agent（本文默认方案）
原始需求 → [需求分析 Agent] → [需求评审 Agent] → 不通过则打回重做 → 进入测试计划
在需求分析之后加一个审核 Agent（即本文的 Agent 2），对比原始需求文档和分析报告，检查功能覆盖度、业务规则准确性、边界条件充分性、隐含需求合理性。性价比最高，推荐大多数正式项目使用。
方案三：双视角分析 + 审核
原始需求
│
├──────────────────┐
▼ ▼
[功能视角分析师] [风险视角分析师]
│ │
▼ ▼
分析报告A 分析报告B
│ │
└────────┬─────────┘
▼
[差异对比 Agent] ← 合并两份报告，标注"高置信度/待审查/有冲突"
│
▼
[需求评审 Agent]
│
▼
进入测试计划
让两个不同视角的 Agent 分别分析，然后对比差异。就像代码 Review 需要另一个人看一样，需求分析也需要"另一双眼睛"。
| 场景 | 单一分析师 | 双视角对比 |
|---|---|---|
| 功能A：两个分析师都提到 | 不确定是否遗漏 | 高置信度 ✅ |
| 功能B：只有分析师A提到 | 不确定对不对 | 自动标记待审查 ⚠️ |
| 功能C：分析师A和B理解不同 | 无法发现 | 自动标记有冲突 ❌ |
选择建议：需求清晰的简单功能 → 方案一；中等复杂度的正式项目 → 方案二（本文实现）；需求模糊 / 业务复杂 / 高风险行业 → 方案三。
7.2 对比实验设计
# 单 Agent：一个 Prompt 做完所有事
SINGLE_AGENT_PROMPT = """
你是一名测试专家。请根据以下需求，直接输出完整的测试用例。
需求：{requirement}
"""
# 多 Agent：七步流水线（本文实现的完整 Pipeline）
对比维度：
| 评估维度 | 说明 |
|---|---|
| 用例总数 | 生成的用例数量 |
| 场景覆盖 | 正向 / 反向 / 边界场景是否齐全 |
| 预期结果质量 | 是否具体可验证，而非模糊描述 |
| 步骤可执行性 | 新人能否直接按步骤执行 |
| 格式规范性 | 编号、字段、格式是否统一 |
7.3 实测对比结果
以下数据基于同一需求（电商登录功能），各跑 3 次取平均值。
┌─────────────────┬──────────────┬──────────────┐
│ 评估维度 │ 单 Agent │ 多 Agent │
├─────────────────┼──────────────┼──────────────┤
│ 用例总数 │ 8~12 条 │ 25~35 条 │
├─────────────────┼──────────────┼──────────────┤
│ 正向场景覆盖 │ ✅ 基本覆盖 │ ✅ 完整覆盖 │
├─────────────────┼──────────────┼──────────────┤
│ 反向场景覆盖 │ ⚠️ 遗漏较多 │ ✅ 覆盖率高 │
│ │ (仅2~3条) │ (8~12条) │
├─────────────────┼──────────────┼──────────────┤
│ 边界值覆盖 │ ❌ 几乎没有 │ ✅ 每个功能 │
│ │ │ 至少1条 │
├─────────────────┼──────────────┼──────────────┤
│ 预期结果质量 │ ⚠️ 部分模糊 │ ✅ 具体可 │
│ │ "显示正确" │ 量化验证 │
├─────────────────┼──────────────┼──────────────┤
│ 步骤可执行性 │ ⚠️ 有时笼统 │ ✅ 新人可 │
│ │ │ 直接执行 │
├─────────────────┼──────────────┼──────────────┤
│ 评审 AI 平均分 │ 55~65 分 │ 82~92 分 │
├─────────────────┼──────────────┼──────────────┤
│ API 调用次数 │ 1 次 │ 8~20 次 │
│ │ │ (含审核+重试)│
├─────────────────┼──────────────┼──────────────┤
│ 耗时 │ ~10 秒 │ ~3~6 分钟 │
├─────────────────┼──────────────┼──────────────┤
│ Token 成本 │ ~2k tokens │ ~40k tokens │
└─────────────────┴──────────────┴──────────────┘
7.4 结论与选择建议
质量差异总结：
✅ 反向场景数量：多 Agent 提升 3~4 倍
✅ 边界值覆盖：从 0 到基本全覆盖
✅ 预期结果可验证性：从"显示正确"到"HTTP 200，响应含 token 字段"
✅ 评审通过率：从 55 分到 85 分以上
选择建议：
| 场景 | 推荐方案 |
|---|---|
| 快速原型、不重要的功能 | 单 Agent（速度快、成本低） |
| 核心业务、上线前回归 | 多 Agent 流水线（质量可控） |
| 合规要求高的行业（金融、医疗） | 多 Agent + 人工审核节点 |
八、进阶优化建议
8.1 引入 RAG 知识库
如果公司有历史测试用例或行业标准，可以用 RAG 让 Agent 参考：
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.runnables import RunnablePassthrough
# 加载历史用例到向量数据库
vectorstore = Chroma.from_documents(
documents,
OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
# 在 Prompt 中加入检索到的参考用例
rag_prompt = ChatPromptTemplate.from_messages([
("system", """你是测试用例设计师。以下是历史参考用例：
{context}
请参考以上用例的风格和质量标准来设计新用例。"""),
("human", "{input}")
])
rag_chain = (
{"context": retriever, "input": RunnablePassthrough()}
| rag_prompt
| llm
| parser
)
8.2 并行执行
如果某些 Agent 之间没有依赖，可以并行：
from langchain_core.runnables import RunnableParallel
# 功能测试用例和性能测试用例可以同时生成
parallel_chain = RunnableParallel(
functional=functional_case_chain,
performance=performance_case_chain
)
results = parallel_chain.invoke({"input": plan})
8.3 成本控制
多 Agent 流水线最大的隐性成本是 API 调用费。优化方法：
| 策略 | 说明 |
|---|---|
| 分级模型 | 生成用 GPT-4o，评审用 GPT-4o-mini（便宜10倍） |
| 结果缓存 | 相同输入直接返回缓存结果 |
| 输出压缩 | 限制每步输出长度，减少后续步骤的输入 token |
| 打回上限 | 设置最大重试次数，防止死循环 |
from langchain.globals import set_llm_cache
from langchain_community.cache import SQLiteCache
# 启用缓存，相同问题不重复调用
set_llm_cache(SQLiteCache(database_path=".langchain.db"))
九、完整项目结构
test-pipeline/
├── environment.yml # conda 环境文件
├── requirements.txt # pip 依赖
│
├── config.py # API Key、模型配置、超时参数
├── main.py # 入口
├── errors.py # 自定义异常（6.1）
├── utils.py # 工具函数（JSON解析、状态校验、重试等）
├── metrics.py # 运行指标采集（6.5）
├── circuit_breaker.py # 熔断器（6.5）
│
├── prompts/ # Prompt 定义（七个 Agent，按流水线顺序）
│ ├── __init__.py
│ ├── requirement.py # Agent 1: 需求分析师
│ ├── analysis_review.py # Agent 2: 需求分析评审员
│ ├── plan.py # Agent 3: 测试计划师
│ ├── plan_review.py # Agent 4: 计划评审员
│ ├── case.py # Agent 5: 用例设计师
│ ├── case_review.py # Agent 6: 用例评审员
│ └── final.py # Agent 7: 审核输出员
│
├── agents/ # Agent 封装
│ ├── __init__.py
│ └── base_agent.py # TestAgent 基类（3.2）
│
├── pipeline/ # 流水线实现
│ ├── __init__.py
│ ├── simple_sequential.py # 简单版（纯 Python 脚本，3.5）
│ ├── langchain_lcel.py # LCEL 管道版（第四章）
│ └── langgraph_flow.py # LangGraph 图版（第五章，推荐）
│
├── knowledge/ # RAG 知识库（可选）
│ └── historical_cases/
│
├── output/ # 输出目录
│ └── test_output.md
│
└── tests/ # 单元测试
├── test_agents.py # Agent 单元测试
├── test_pipeline.py # 流水线集成测试
├── test_prompts.py # Prompt 质量测试
└── test_error_handling.py # 异常处理测试
# environment.yml (conda)
name: test-pipeline
dependencies:
- python=3.11
- pip:
- langchain>=0.3.0,<0.4.0
- langchain-openai>=0.2.0,<0.3.0
- langchain-community>=0.3.0
- langgraph>=0.2.0,<0.3.0
- openai>=1.50.0
- chromadb>=0.5.0
- tiktoken>=0.7.0
- pydantic>=2.0
# requirements.txt
langchain>=0.3.0,<0.4.0
langchain-openai>=0.2.0,<0.3.0
langchain-community>=0.3.0
langgraph>=0.2.0,<0.3.0
openai>=1.50.0
chromadb>=0.5.0
tiktoken>=0.7.0
pydantic>=2.0
十、总结
| 层次 | 核心要点 |
|---|---|
| Prompt 层 | 角色具体化、任务步骤化、格式示例化、负面约束、Few-Shot、量化评分 |
| 架构层 | 七个 Agent 按流水线顺序：①需求分析→②需求评审→③计划→④计划评审→⑤用例→⑥用例评审→⑦输出 |
| 流程控制 | 生成 → 评审 → 打回重做循环，最多 N 次重试，超限转人工 |
| 需求保障 | 需求分析评审是地基校验——三种方案按项目风险等级选择 |
| 异常处理 | 七层覆盖：LLM 调用（限流/超时/溢出/过滤）→ JSON 解析（重试+兜底）→ 状态校验 → 业务降级 → 路由感知 → 流水线级（超时/检查点/熔断器）→ 人工兜底 |
| 框架选择 | 简单场景用 Python 脚本，标准场景用 LCEL，复杂条件分支用 LangGraph |
| 进阶优化 | RAG 知识增强、并行执行、分级模型、缓存降本、检查点持久化、熔断器 |
最后一句话：这套系统的质量上限取决于 Prompt 的质量，代码只是把好 Prompt 串起来的胶水。先打磨 Prompt，再搭架构。而需求分析阶段的审核是整条链路中最值得投入的一环——地基正了，上面才不会歪。
如果这篇文章对你有帮助，欢迎点赞收藏。有问题可以在评论区交流。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
15
0- 0
已为社区贡献2条内容
所有评论(0)