Agent 系统集成测试模拟用户行为的多轮对话自动化测试一、Agent 上线 3 天后才发现查订单时偶尔会卡死Agent 功能上线前通过了单元测试和单轮工具调用测试。但第三天用户反馈连续对话 5 轮以上时Agent 偶尔会陷入正在思考的循环既不回复也不报错。排查发现根因Agent 在第 3 轮调用了 A 工具第 4 轮调用了 B 工具B 工具的错误码让 LLM 重新调用 A 工具形成 A→B→A→B 的死循环。单元测试覆盖不了这种场景因为单个工具调用是正常的。只有多轮对话的集成测试才能发现这类状态累积导致的问题。二、Agent 集成测试框架Agent 集成测试框架主要由四个核心模块协同工作测试用例构造、测试执行引擎、断言层以及测试报告。首先测试用例构造阶段负责定义多步用户行为场景并将其转化为测试脚本明确每一步的用户输入与期望行为。其次测试执行引擎负责初始化 Agent加载工具与知识库按步骤执行输入并验证最终清理状态并重置。在执行过程中断言层会对每一步进行多维度校验包括工具调用断言是否调用正确工具、响应断言是否包含关键信息、性能断言响应延迟是否低于阈值以及安全断言未执行危险操作。最后测试报告模块根据断言结果生成报告通过则标记为通过失败则记录详细日志并提供 Agent 全链路 Trace 以便排查。三、Python 集成测试框架实现import time import json from typing import List, Dict, Any, Optional, Callable from dataclasses import dataclass, field from enum import Enum class TestStatus(Enum):PASS pass FAIL fail TIMEOUT timeout ERROR errordataclassclass StepAssertion:单步断言tool_called: Optional[str] None # 期望调用的工具tool_not_called: Optional[str] None # 期望不调用的工具response_contains: Optional[List[str]] None # 响应中应包含的关键词response_not_contains: Optional[List[str]] None # 响应中不应包含的max_latency_ms: int 10000 # 最大响应延迟max_tool_calls: int 5 # 单步最大工具调用次数dataclassclass TestScenario:测试场景——多步对话定义name: strdescription: strsteps: List[Dict[str, Any]] # [{user_input: ..., assertion: StepAssertion}]dataclassclass TestStepResult:单步测试结果step_index: intuser_input: stragent_response: strtool_calls: List[str]latency_ms: floatstatus: TestStatusfailures: List[str] field(default_factorylist)class AgentIntegrationTest:Agent 集成测试框架def __init__(self, agent_factory: Callable, timeout_seconds: int 60): agent_factory: Agent 工厂函数每次测试创建新的 Agent 实例 timeout_seconds: 整体测试超时 self.agent_factory agent_factory self.timeout_seconds timeout_seconds self.results: List[TestStepResult] [] def run_scenario(self, scenario: TestScenario) - Dict[str, Any]: 运行一个测试场景 print(f\n{*60}) print(f测试场景: {scenario.name}) print(f描述: {scenario.description}) print(f{*60}) # 创建新的 Agent 实例状态隔离 agent self.agent_factory() scenario_results [] passed True for i, step in enumerate(scenario.steps): user_input step[user_input] assertion step.get(assertion, StepAssertion()) print(f\n Step {i1}: 用户输入 - {user_input[:50]}...) result self._run_step(agent, i, user_input, assertion) scenario_results.append(result) if result.status ! TestStatus.PASS: passed False print(f [FAIL] {result.failures}) else: print(f [PASS] 延迟: {result.latency_ms:.0f}ms, 工具调用: {result.tool_calls}) return { scenario: scenario.name, passed: passed, total_steps: len(scenario.steps), passed_steps: sum(1 for r in scenario_results if r.status TestStatus.PASS), results: scenario_results, } def _run_step( self, agent, step_index: int, user_input: str, assertion: StepAssertion, ) - TestStepResult: 运行单个对话步骤 start time.time() failures [] tool_calls [] response try: # 执行 Agent带超时控制 import threading result_container {response: None, tools: []} error_container {error: None} def _execute(): try: resp, tools agent.run(user_input) result_container[response] resp result_container[tools] tools except Exception as e: error_container[error] e thread threading.Thread(target_execute, daemonTrue) thread.start() thread.join(timeoutassertion.max_latency_ms / 1000) if thread.is_alive(): return TestStepResult( step_indexstep_index, user_inputuser_input, agent_response, tool_calls[], latency_ms(time.time() - start) * 1000, statusTestStatus.TIMEOUT, failures[f响应超时: {assertion.max_latency_ms}ms], ) if error_container[error]: raise error_container[error] response result_container[response] or tool_calls result_container[tools] or [] except Exception as e: return TestStepResult( step_indexstep_index, user_inputuser_input, agent_responsestr(e), tool_calls[], latency_ms(time.time() - start) * 1000, statusTestStatus.ERROR, failures[f执行异常: {e}], ) latency_ms (time.time() - start) * 1000 # 执行断言 # 1. 检查工具调用 if assertion.tool_called: if assertion.tool_called not in tool_calls: failures.append( f期望调用 {assertion.tool_called}, f实际调用: {tool_calls} ) if assertion.tool_not_called: if assertion.tool_not_called in tool_calls: failures.append( f不应调用 {assertion.tool_not_called} ) # 2. 检查响应内容 if assertion.response_contains: for keyword in assertion.response_contains: if keyword not in response: failures.append(f响应中缺少关键词: {keyword}) if assertion.response_not_contains: for keyword in assertion.response_not_contains: if keyword in response: failures.append(f响应中不应包含: {keyword}) # 3. 检查延迟 if latency_ms assertion.max_latency_ms: failures.append( f延迟超标: {latency_ms:.0f}ms {assertion.max_latency_ms}ms ) # 4. 检查工具调用次数防止死循环 if len(tool_calls) assertion.max_tool_calls: failures.append( f工具调用次数超标: {len(tool_calls)} {assertion.max_tool_calls} ) status TestStatus.PASS if not failures else TestStatus.FAIL return TestStepResult( step_indexstep_index, user_inputuser_input, agent_responseresponse, tool_callstool_calls, latency_mslatency_ms, statusstatus, failuresfailures, )### 定义测试场景 python # 定义测试场景——覆盖常见的多轮对话模式 CHATBOT_SCENARIOS [ # 场景一正常多轮对话 TestScenario( name多轮正常对话, description用户连续提问Agent 正确调用工具并回复, steps[ { user_input: 帮我查一下订单 12345 的状态, assertion: StepAssertion( tool_calledquery_order, response_contains[订单状态, 12345], max_latency_ms5000, ), }, { user_input: 这个订单能退款吗, assertion: StepAssertion( tool_calledquery_refund_policy, response_contains[退款], max_latency_ms5000, ), }, { user_input: 好的帮我申请退款, assertion: StepAssertion( tool_calledrefund_order, response_contains[退款申请], max_latency_ms8000, ), }, ], ), # 场景二死循环检测 TestScenario( name死循环检测, descriptionAgent 不应在连续工具调用中陷入循环, steps[ { user_input: 查询一个不存在的订单 ABCDEFG, assertion: StepAssertion( response_contains[不存在, 未找到], max_tool_calls2, # 关键限制工具调用次数 max_latency_ms10000, ), }, ], ), # 场景三权限边界测试 TestScenario( name权限边界测试, descriptionAgent 不应调用超出权限的工具, steps[ { user_input: 删除订单 12345, assertion: StepAssertion( tool_not_calleddelete_order, # 不应调用删除工具 response_contains[权限不足, 无法], max_latency_ms5000, ), }, ], ), # 场景四上下文记忆测试 TestScenario( name上下文记忆测试, descriptionAgent 应在多轮对话中记住之前的上下文, steps[ { user_input: 我想买一个蓝牙耳机, assertion: StepAssertion( response_contains[蓝牙耳机, 推荐], ), }, { user_input: 预算 200 元以内, # 注意此时 Agent 应该结合之前的蓝牙耳机上下文 assertion: StepAssertion( response_contains[蓝牙耳机, 200], ), }, ], ), ]测试运行器class TestRunner: 测试运行器——批量执行 生成报告 def __init__(self, test_framework: AgentIntegrationTest): self.framework test_framework self.all_results [] def run_all(self, scenarios: List[TestScenario]) - Dict: 运行所有测试场景 total_passed 0 total_scenarios len(scenarios) for scenario in scenarios: result self.framework.run_scenario(scenario) self.all_results.append(result) if result[passed]: total_passed 1 # 生成汇总报告 summary { total_scenarios: total_scenarios, passed_scenarios: total_passed, failed_scenarios: total_scenarios - total_passed, pass_rate: f{total_passed/total_scenarios*100:.1f}%, details: self.all_results, } self._print_summary(summary) return summary def _print_summary(self, summary: Dict): print(f\n{*60}) print(f测试汇总) print(f{*60}) print(f总场景: {summary[total_scenarios]}) print(f通过: {summary[passed_scenarios]}) print(f失败: {summary[failed_scenarios]}) print(f通过率: {summary[pass_rate]})四、边界分析与 Trade-offs测试的真实性模拟的用户行为可能无法覆盖真实用户的所有模式。建议配合线上 A/B 测试和用户反馈日志持续补充测试场景。状态隔离的必要性每个测试场景必须使用独立的 Agent 实例。共享状态会让测试结果之间相互影响导致假通过或假失败。Mock vs 真实服务工具调用层建议综合使用 Stub快速 真实服务保真。单元测试用 Stub集成测试用真实服务。测试维护成本随着 Agent 能力的增加测试场景也在增长。建议使用基于日志回放的方式生成测试用例——从生产日志中提取真实对话序列作为测试输入。五、总结Agent 的集成测试和传统服务的测试有本质区别多轮依赖于状态——测试必须覆盖状态累积效应工具调用不可预测——断言要覆盖调用链而不仅是单次调用死循环是常见故障——必须限制工具调用次数LLM 的非确定性——需要容忍一定范围的输出差异自动化集成测试的目标不是 100% 覆盖所有场景而是覆盖已知的风险模式死循环、权限越界、上下文遗忘、延迟超标。这些是 80% 的生产故障来源。