云原生 Agent 怎么测:Prompt 契约、外部依赖与端到端评测
示例场景:在 Kubernetes 集群中部署处理多步 SQL 查询与外部 API 调用的 LLM Agent 服务时,如果没有细粒度日志和结构化断言,团队往往只能靠终端日志与少量 Prompt 手工验证。这类验证难以稳定发现编排逻辑缺陷和工具调用循环。较稳妥的做法是把单元测试、集成测试和端到端(E2E)测试分层执行。
graph TD A[开发者提交 Agent 编排代码] --> B[单元测试: Prompt 模板解析与 Tool 签名校验] B --> C[集成测试: Mock 大模型 API 与外部 HTTP/DB 工具] C --> D[端到端测试: 部署至 K8s 影子集群运行真实 Worker 任务] D --> E[输出评价指标: Tool 调用准确率 / 响应延迟 / Token 消耗]单元测试阶段:抓取 Prompt 模板解析与 Tool Calling 模式校验
在单元测试阶段,验证逻辑应当严格剥离真实大模型 API 调用。本阶段的核心目标在于确认 Prompt 字符串拼接是否合规、系统提示词(System Prompt)格式是否完备,以及工具函数(Tools)导出的 Schema 是否完全契约化。
在 Agent 开发中,工具参数的类型、必填项和描述一旦与执行函数不一致,就可能在运行时失败。下述代码演示如何测试入参校验与 Schema 生成逻辑;实际字段格式应以所使用 SDK 的 Tool Calling 文档为准:
import json import unittest from typing import Dict, Any def generate_tool_schema(func) -> Dict[str, Any]: """提取函数签名并转换为 OpenAI Tool Calling 格式""" if not hasattr(func, "__doc__") or not func.__doc__: raise ValueError(f"工具函数 {func.__name__} 缺失 docstring 说明") return { "type": "function", "function": { "name": func.__name__, "description": func.__doc__.strip(), "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "查询 SQL 语句"} }, "required": ["query"] } } } def execute_sql_query(query: str) -> str: """在只读数据库镜像上执行查询""" if "DROP" in query.upper() or "DELETE" in query.upper(): raise PermissionError("禁止在只读工具中执行高风险破坏性语句") return json.dumps([{"id": 101, "status": "active"}]) class TestAgentUnitTest(unittest.TestCase): def test_schema_generation(self): schema = generate_tool_schema(execute_sql_query) self.assertEqual(schema["function"]["name"], "execute_sql_query") self.assertIn("query", schema["function"]["parameters"]["properties"]) def test_tool_execution_boundary(self): with self.assertRaises(PermissionError): execute_sql_query("DELETE FROM users WHERE id = 1") result = execute_sql_query("SELECT * FROM users LIMIT 1") self.assertIn("active", result) if __name__ == "__main__": unittest.main()单元测试脚本的执行应当无缝接入 Docker 构建流水线的前置步骤,通过下述命令行可以直接在 CI 容器环境中进行快速验证:
python3 -m unittest test_agent_unit.py -vCI 可以检查关键 Tool 的 Schema、必填字段和异常路径;字段描述有助于模型选择工具,但不能单独保证调用正确。
集成测试阶段:Mock 外部大模型 API 与外部 HTTP/DB 工具响应
当 Agent 推进至多步编排逻辑测试时,网络波动与 LLM 响应的不确定性容易导致持续集成(CI)流水线产生偶发性失败。集成测试的必要手段在于构建只读 HTTP Server 拦截 LLM API 请求,返回预设的 Tool Call JSON 结构与控制信号。
在这个层级中,测试的重点是验证 Agent 决策引擎接收到模型返回的 tool_calls 结构后,能否正确路由并分发至底层子服务,同时妥善处理 HTTP 404、数据库连接超时或 API 限流(HTTP 429)等异常边界。
import pytest from unittest.mock import Mock, patch import requests class AgentOrchestrator: def __init__(self, llm_endpoint: str): self.llm_endpoint = llm_endpoint def step(self, prompt: str) -> dict: try: resp = requests.post( self.llm_endpoint, json={"messages": [{"role": "user", "content": prompt}]}, timeout=5.0 ) resp.raise_for_status() data = resp.json() # 处理 Tool Calling 路由逻辑 if "tool_calls" in data.get("choices", [{}])[0].get("message", {}): tool_call = data["choices"][0]["message"]["tool_calls"][0] return self._invoke_tool(tool_call) return {"status": "completed", "output": data["choices"][0]["message"]["content"]} except requests.exceptions.Timeout: return {"status": "error", "message": "LLM 响应超时,触发降级保护策略"} except Exception as e: return {"status": "error", "message": str(e)} def _invoke_tool(self, tool_call: dict) -> dict: func_name = tool_call.get("function", {}).get("name") if func_name == "get_db_metrics": return {"status": "tool_executed", "result": "cpu_usage=85%"} return {"status": "error", "message": f"未知的工具名称: {func_name}"} def test_agent_integration_mock_llm(): orchestrator = AgentOrchestrator(llm_endpoint="http://mock-llm.local/v1/chat") # Mock LLM 返回标准的 Tool Call 响应 mock_response = Mock() mock_response.status_code = 200 mock_response.json.return_value = { "choices": [{ "message": { "tool_calls": [{ "function": {"name": "get_db_metrics", "arguments": "{}"} }] } }] } with patch("requests.post", return_value=mock_response): result = orchestrator.step("检查数据库状态") assert result["status"] == "tool_executed" assert "cpu_usage=85%" in result["result"]使用 pytest 自动化测试框架可以高效捕获并运行集成测试用例,并在控制台生成清晰的堆栈追踪:
pytest test_agent_integration.py -s --tb=short重试次数和超时应按模型服务的限流策略、任务预算和工具幂等性配置。示例中的 3 次与 5 秒只是起点,失败路径还应确保请求连接被关闭。
端到端测试阶段:云原生环境下的 LLM Agent 自动化测试链路与评测指标
在真实 Kubernetes 集群环境中部署 Agent 服务时,验证策略需要进一步延伸至集群基础设施层面。端到端(E2E)测试必须完整校验部署的 Deployment 资源、Service 路由、Pod 间 DNS 解析以及 IAM 权限配置。
系统评测除最终输出外,还应至少记录以下指标:
- Tool 调用准确度:比较实际工具路径与标注集(Golden Dataset)的差异,并按工具风险分级设定通过线。
- 任务延迟(Latency P95):分别统计模型推理和本地工具耗时。文中的 1.2 秒为示例演练值,真实目标需以交互预期和容量基线确定。
- 循环保护:测试同一工具被连续调用超过设定上限时,系统能中断会话并告警。上限不应脱离任务类型固定为 5 次。
运维与测试工程师可通过以下命令行直接提取影子测试命名空间中的 Pod 状态及链路日志:
kubectl get pods -n ai-test-env -l app=agent-executor -o wide kubectl logs -n ai-test-env -l app=agent-executor --tail=100 | grep "tool_call_loop_count"拟真演练捕获日志如下:
[示例输出] [WARN] tool_call_loop_count reached limit=5 for tool=execute_sql_query, interrupting agent session context当需要在 E2E 阶段向测试 Pod 施加并发负载并监控资源开销时,结合 kubectl top 工具可以实时获取 CPU 与内存消耗数据:
kubectl top pod -n ai-test-env -l app=agent-executor压测时应记录单 Pod 的 CPU、内存、工具循环次数和请求延迟,并以部署环境的资源限制作为判断依据。示例日志只说明熔断器应能留下可检索的原因,不能替代真实容量测试。
端到端测试可定期向测试集群投放固定的评测集,比对期望输出并计算指标。镜像或编排逻辑更新后,应关注指标是否回退;分层测试能降低回归风险,但不能替代上线后的监控和人工审核。