云原生 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}]}, timeout5.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_usage85%} return {status: error, message: f未知的工具名称: {func_name}} def test_agent_integration_mock_llm(): orchestrator AgentOrchestrator(llm_endpointhttp://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_valuemock_response): result orchestrator.step(检查数据库状态) assert result[status] tool_executed assert cpu_usage85% in result[result]使用 pytest 自动化测试框架可以高效捕获并运行集成测试用例并在控制台生成清晰的堆栈追踪pytest test_agent_integration.py -s --tbshort重试次数和超时应按模型服务的限流策略、任务预算和工具幂等性配置。示例中的 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 appagent-executor -o wide kubectl logs -n ai-test-env -l appagent-executor --tail100 | grep tool_call_loop_count拟真演练捕获日志如下[示例输出] [WARN] tool_call_loop_count reached limit5 for toolexecute_sql_query, interrupting agent session context当需要在 E2E 阶段向测试 Pod 施加并发负载并监控资源开销时结合 kubectl top 工具可以实时获取 CPU 与内存消耗数据kubectl top pod -n ai-test-env -l appagent-executor压测时应记录单 Pod 的 CPU、内存、工具循环次数和请求延迟并以部署环境的资源限制作为判断依据。示例日志只说明熔断器应能留下可检索的原因不能替代真实容量测试。端到端测试可定期向测试集群投放固定的评测集比对期望输出并计算指标。镜像或编排逻辑更新后应关注指标是否回退分层测试能降低回归风险但不能替代上线后的监控和人工审核。