去年双十一，我们公司的 AI 客服系统经历了上线以来最严峻的考验。大促前夜，团队信心满满地部署了新版本的大语言模型 API，想着终于可以支撑住万级并发了。结果零点一到，订单咨询响应时间从 200ms 暴涨到 8 秒，退款请求直接返回 500 错误——整整宕机 47 分钟，客诉电话被打爆，直接损失超过 80 万营收。这场灾难的根源，事后复盘才发现：新版本 API 改变了响应字段结构，而我们下游的订单系统完全没有感知到变更。
这就是为什么我今天要系统性地讲解「AI 服务变更的契约测试」。我自己在 HolySheep AI 平台上部署了 12 个生产环境项目，踩过无数次 API 变更的坑，终于总结出一套行之有效的自动化测试方案。接下来，我会从实战场景出发，带你构建完整的 AI 服务契约测试体系。
什么是 AI 服务契约测试
契约测试（Contract Testing）的核心思想是：在分布式系统中，每个服务之间通过「契约」来约定数据交互的格式和语义。当某个服务更新时，自动化验证它是否符合已有的契约定义。
在 AI 服务场景下，契约测试需要覆盖以下几个维度：
- 接口契约：请求参数格式、响应字段结构、状态码定义
- 语义契约：相同输入是否产生语义一致的输出（容忍 LLM 的微小随机性）
- 性能契约：响应延迟 SLA、并发吞吐量、TPS 限制
- 成本契约：Token 消耗是否符合预算、费用不会突增
实战场景：电商促销日 AI 客服系统
让我以一个真实的电商 AI 客服场景为例，展示完整的契约测试实现。假设我们有这样一套系统架构：
┌─────────────┐ ┌──────────────┐ ┌─────────────────┐
│ 用户请求 │ ──▶ │ 网关层 │ ──▶ │ AI 客服 API │
│ (促销页面) │ │ (限流/鉴权) │ │ (HolySheep) │
└─────────────┘ └──────────────┘ └────────┬────────┘
│
┌──────────────┐ │
│ 订单系统 │ ◀────────────┘
│ (处理业务) │
└──────────────┘在双十一这样的高并发场景下，AI 服务可能随时需要切换模型或调整参数。契约测试的作用就是确保这些变更不会破坏下游系统的正常工作。
使用 HolySheep API 构建基础测试框架
首先，我需要搭建一个基础的 AI 服务客户端和测试框架。HolyShehe AI 的优势在于：国内直连延迟低于 50ms，注册即送免费额度，且汇率是 ¥1=$1（官方是 ¥7.3=$1），对于我们这种日均调用量超过百万 Token 的团队来说，每月能节省超过 85% 的 API 费用。
import requests
import hashlib
import json
import time
from typing import Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
import asyncio
from aiohttp import ClientSession, TCPConnector
class TestSeverity(Enum):
CRITICAL = "critical" # 阻断性问题
MAJOR = "major" # 功能受损
MINOR = "minor" # 非关键问题
@dataclass
class ContractTestResult:
test_name: str
passed: bool
severity: TestSeverity
expected: Any
actual: Any
latency_ms: float
error_message: Optional[str] = None
@dataclass
class AIResponse:
content: str
model: str
usage: Dict[str, int] = field(default_factory=dict)
latency_ms: float = 0.0
finish_reason: str = ""
class HolySheepClient:
"""HolySheep AI API 客户端封装"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url.rstrip('/')
self.session = None
async def create_session(self):
"""创建异步 HTTP 会话（复用连接提升性能）"""
connector = TCPConnector(limit=100, limit_per_host=20)
self.session = ClientSession(connector=connector)
async def close(self):
if self.session:
await self.session.close()
async def chat_completion(
self,
messages: list,
model: str = "gpt-4.1",
temperature: float = 0.7,
max_tokens: int = 1000,
timeout: float = 30.0
) -> AIResponse:
"""
调用 HolySheep AI 聊天补全接口
参数:
messages: 对话消息列表
model: 模型名称（支持 gpt-4.1, claude-sonnet-4.5, gemini-2.5-flash 等）
temperature: 温度参数（控制随机性）
max_tokens: 最大生成 Token 数
timeout: 超时时间（秒）
返回:
AIResponse 对象
"""
url = f"{self.base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
start_time = time.perf_counter()
try:
async with self.session.post(
url,
json=payload,
headers=headers,
timeout=timeout
) as response:
response.raise_for_status()
data = await response.json()
latency = (time.perf_counter() - start_time) * 1000
return AIResponse(
content=data["choices"][0]["message"]["content"],
model=data.get("model", model),
usage=data.get("usage", {}),
latency_ms=latency,
finish_reason=data["choices"][0].get("finish_reason", "stop")
)
except Exception as e:
latency = (time.perf_counter() - start_time) * 1000
raise RuntimeError(f"API 调用失败 (延迟 {latency:.1f}ms): {str(e)}")构建契约测试套件
接下来，我需要设计一个完整的契约测试框架。这个框架需要能够：自动化验证 API 响应结构、监控性能指标、检测语义一致性，并且在测试失败时生成详细的诊断报告。
import pytest
import pytest_asyncio
from contract_tester import HolySheepClient, ContractTestResult, TestSeverity
import jsonschema
from typing import Callable
class AIContractTester:
"""AI 服务契约测试器"""
# 定义响应契约 schema（根据业务需求调整）
RESPONSE_SCHEMA = {
"type": "object",
"required": ["choices", "model", "usage"],
"properties": {
"id": {"type": "string"},
"object": {"type": "string"},
"created": {"type": "integer"},
"model": {"type": "string"},
"choices": {
"type": "array",
"minItems": 1,
"items": {
"type": "object",
"required": ["message", "finish_reason", "index"],
"properties": {
"message": {
"type": "object",
"required": ["role", "content"],
"properties": {
"role": {"type": "string", "enum": ["assistant"]},
"content": {"type": "string"}
}
},
"finish_reason": {
"type": "string",
"enum": ["stop", "length", "content_filter"]
},
"index": {"type": "integer", "minimum": 0}
}
}
},
"usage": {
"type": "object",
"required": ["prompt_tokens", "completion_tokens", "total_tokens"],
"properties": {
"prompt_tokens": {"type": "integer", "minimum": 0},
"completion_tokens": {"type": "integer", "minimum": 0},
"total_tokens": {"type": "integer", "minimum": 0}
}
}
}
}
# 性能 SLA 定义（毫秒）
PERFORMANCE_SLA = {
"p50": 100, # 中位数延迟
"p95": 500, # 95 分位延迟
"p99": 1000, # 99 分位延迟
}
def __init__(self, client: HolySheepClient):
self.client = client
self.test_results: list[ContractTestResult] = []
def _validate_schema(self, response_data: dict, test_name: str) -> ContractTestResult:
"""验证响应 JSON Schema"""
try:
jsonschema.validate(instance=response_data, schema=self.RESPONSE_SCHEMA)
return ContractTestResult(
test_name=test_name,
passed=True,
severity=TestSeverity.CRITICAL,
expected="符合 schema 定义",
actual="验证通过",
latency_ms=0
)
except jsonschema.ValidationError as e:
return ContractTestResult(
test_name=test_name,
passed=False,
severity=TestSeverity.CRITICAL,
expected="符合 schema 定义",
actual=f"Schema 验证失败: {e.message}",
latency_ms=0,
error_message=str(e)
)
async def test_interface_contract(self) -> ContractTestResult:
"""
测试接口契约：验证 API 响应结构是否符合规范
HolySheep API 的优势在于返回格式完全兼容 OpenAI 标准，
这意味着我们的契约测试可以直接复用 OpenAI 的 schema 定义
"""
test_prompt = "请用一句话介绍你自己"
messages = [{"role": "user", "content": test_prompt}]
response = await self.client.chat_completion(
messages=messages,
model="gpt-4.1",
max_tokens=50
)
# 构造完整的 API 响应数据（用于 schema 验证）
response_data = {
"id": f"test-{int(time.time())}",
"object": "chat.completion",
"created": int(time.time()),
"model": response.model,
"choices": [{
"message": {"role": "assistant", "content": response.content},
"finish_reason": response.finish_reason,
"index": 0
}],
"usage": response.usage
}
return self._validate_schema(response_data, "interface_contract_schema")
async def test_performance_contract(
self,
sample_size: int = 100,
concurrency: int = 10
) -> list[ContractTestResult]:
"""
测试性能契约：验证响应延迟是否满足 SLA 要求
在电商促销场景下，AI 客服的响应延迟直接关系到用户体验和转化率
根据我们的经验，P95 延迟超过 500ms 会导致用户满意度显著下降
"""
results = []
test_prompt = "查询订单状态，订单号 A123456789"
messages = [{"role": "user", "content": test_prompt}]
async def single_request():
return await self.client.chat_completion(
messages=messages,
model="gpt-4.1",
max_tokens=100
)
# 使用信号量控制并发
semaphore = asyncio.Semaphore(concurrency)
async def controlled_request():
async with semaphore:
return await single_request()
# 并发执行请求
tasks = [controlled_request() for _ in range(sample_size)]
responses = await asyncio.gather(*tasks, return_exceptions=True)
latencies = [r.latency_ms for r in responses if isinstance(r, AIResponse)]
if not latencies:
results.append(ContractTestResult(
test_name="performance_contract",
passed=False,
severity=TestSeverity.CRITICAL,
expected="至少部分请求成功",
actual="所有请求均失败",
latency_ms=0
))
return results
# 计算分位数
sorted_latencies = sorted(latencies)
p50_idx = int(len(sorted_latencies) * 0.50)
p95_idx = int(len(sorted_latencies) * 0.95)
p99_idx = int(len(sorted_latencies) * 0.99)
p50 = sorted_latencies[p50_idx]
p95 = sorted_latencies[p95_idx]
p99 = sorted_latencies[p99_idx]
avg = sum(latencies) / len(latencies)
# P50 测试
results.append(ContractTestResult(
test_name="performance_p50",
passed=p50 <= self.PERFORMANCE_SLA["p50"],
severity=TestSeverity.MAJOR,
expected=f"P50 <= {self.PERFORMANCE_SLA['p50']}ms",
actual=f"P50 = {p50:.1f}ms",
latency_ms=p50
))
# P95 测试
results.append(ContractTestResult(
test_name="performance_p95",
passed=p95 <= self.PERFORMANCE_SLA["p95"],
severity=TestSeverity.MAJOR,
expected=f"P95 <= {self.PERFORMANCE_SLA['p95']}ms",
actual=f"P95 = {p95:.1f}ms",
latency_ms=p95
))
# P99 测试
results.append(ContractTestResult(
test_name="performance_p99",
passed=p99 <= self.PERFORMANCE_SLA["p99"],
severity=TestSeverity.MINOR,
expected=f"P99 <= {self.PERFORMANCE_SLA['p99']}ms",
actual=f"P99 = {p99:.1f}ms",
latency_ms=p99
))
return results
async def test_semantic_contract(
self,
test_cases: list[dict]
) -> list[ContractTestResult]:
"""
测试语义契约：验证相同输入是否产生语义一致的输出
这里使用关键词匹配和简单的相似度计算来验证输出语义
在生产环境中，建议使用 embedding 相似度或 LLM-as-Judge 方式
"""
results = []
for case in test_cases:
response = await self.client.chat_completion(
messages=[{"role": "user", "content": case["input"]}],
model="gpt-4.1",
temperature=0.0, # 使用确定性模式
max_tokens=200
)
content = response.content.lower()
# 检查关键词是否出现在输出中
matched_keywords = [
kw for kw in case.get("expected_keywords", [])
if kw.lower() in content
]
match_rate = len(matched_keywords) / len(case.get("expected_keywords", [1])) if case.get("expected_keywords") else 1.0
passed = match_rate >= case.get("min_match_rate", 0.8)
results.append(ContractTestResult(
test_name=f"semantic_contract_{case['id']}",
passed=passed,
severity=TestSeverity.MAJOR,
expected=f"匹配率 >= {case.get('min_match_rate', 0.8) * 100}%",
actual=f"匹配率 = {match_rate * 100:.1}% (关键词: {matched_keywords})",
latency_ms=response.latency_ms,
error_message=None if passed else f"语义不匹配，输出: {content[:100]}..."
))
return results
async def run_full_suite(self) -> dict:
"""运行完整测试套件"""
print("🚀 开始运行 AI 服务契约测试套件...")
all_results = []
# 1. 接口契约测试
print("📋 测试 1: 接口契约验证...")
interface_result = await self.test_interface_contract()
all_results.append(interface_result)
# 2. 性能契约测试
print("⚡ 测试 2: 性能 SLA 验证...")
perf_results = await self.test_performance_contract(sample_size=50, concurrency=5)
all_results.extend(perf_results)
# 3. 语义契约测试
print("🔍 测试 3: 语义一致性验证...")
semantic_cases = [
{
"id": "order_status",
"input": "我的订单 A123456789 现在是什么状态？",
"expected_keywords": ["订单", "状态", "已发货", "配送中", "签收"],
"min_match_rate": 0.6
},
{
"id": "refund_policy",
"input": "商品不满意可以退货吗？",
"expected_keywords": ["退货", "退款", "7天", "30天", "政策"],
"min_match_rate": 0.6
}
]
semantic_results = await self.test_semantic_contract(semantic_cases)
all_results.extend(semantic_results)
# 生成报告
passed = sum(1 for r in all_results if r.passed)
failed = len(all_results) - passed
print(f"\n📊 测试完成: {passed} 通过, {failed} 失败")
return {
"summary": {"passed": passed, "failed": failed, "total": len(all_results)},
"results": all_results
}
运行测试
@pytest_asyncio.fixture
async def client():
"""创建测试客户端 fixture"""
client = HolySheepClient(api_key="YOUR_HOLYSHEEP_API_KEY")
await client.create_session()
yield client
await client.close()
@pytest.mark.asyncio
async def test_ai_service_contracts(client):
"""主测试入口"""
tester = AIContractTester(client)
report = await tester.run_full_suite()
# 断言所有关键测试通过
assert report["summary"]["failed"] == 0, \
f"存在 {report['summary']['failed']} 个测试失败"
print("\n✅ 所有契约测试通过！")部署监控告警系统
测试通过只是第一步，更重要的是在生产环境中持续监控契约的合规性。我设计了一个实时监控脚本，当 API 响应不符合契约时自动告警：
import logging
import smtplib
from email.mime.text import MIMEText
from typing import Optional
from dataclasses import dataclass
from datetime import datetime
import redis
import json
@dataclass
class AlertConfig:
latency_threshold_ms: int = 500 # 延迟告警阈值
error_rate_threshold: float = 0.05 # 错误率告警阈值
schema_violation_threshold: int = 3 # Schema 违规次数阈值
class ProductionMonitor:
"""生产环境契约监控器"""
def __init__(self, redis_host: str = "localhost", redis_port: int = 6379):
self.redis_client = redis.Redis(host=redis_host, port=redis_port, db=0)
self.logger = logging.getLogger(__name__)
self.alert_config = AlertConfig()
self.metrics_key = "ai_contract_metrics"
def record_request(self, request_id: str, latency_ms: float, success: bool, schema_valid: bool):
"""记录单个请求的指标"""
timestamp = datetime.now().isoformat()
metric = {
"request_id": request_id,
"timestamp": timestamp,
"latency_ms": latency_ms,
"success": success,
"schema_valid": schema_valid
}
# 存储到 Redis（保留最近 10000 条）
self.redis_client.lpush(self.metrics_key, json.dumps(metric))
self.redis_client.ltrim(self.metrics_key, 0, 9999)
# 实时检查告警条件
self._check_alerts(metric)
def _check_alerts(self, metric: dict):
"""检查是否触发告警条件"""
alerts = []
# 延迟告警
if metric["latency_ms"] > self.alert_config.latency_threshold_ms:
alerts.append(f"⚠️ 延迟告警: {metric['latency_ms']:.1f}ms > {self.alert_config.latency_threshold_ms}ms")
# Schema 违规告警
if not metric["schema_valid"]:
alerts.append(f"🚨 Schema 违规: request_id={metric['request_id']}")
if alerts:
self._send_alert(alerts)
def _send_alert(self, messages: list):
"""发送告警通知"""
alert_text = "\n".join(messages)
self.logger.warning(f"告警触发:\n{alert_text}")
# 这里可以接入企业微信、钉钉、飞书等通知渠道
# 企业微信 webhook 示例：
# webhook_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"
# requests.post(webhook_url, json={"msgtype": "text", "text": {"content": alert_text}})
def get_metrics_summary(self, time_window_minutes: int = 5) -> dict:
"""获取指标汇总（用于 Dashboard）"""
metrics = self.redis_client.lrange(self.metrics_key, 0, -1)
metrics = [json.loads(m) for m in metrics]
if not metrics:
return {"error": "暂无数据"}
# 计算统计指标
latencies = [m["latency_ms"] for m in metrics]
success_count = sum(1 for m in metrics if m["success"])
schema_valid_count = sum(1 for m in metrics if m["schema_valid"])
return {
"total_requests": len(metrics),
"success_rate": success_count / len(metrics) * 100,
"schema_valid_rate": schema_valid_count / len(metrics) * 100,
"avg_latency_ms": sum(latencies) / len(latencies),
"p95_latency_ms": sorted(latencies)[int(len(latencies) * 0.95)] if latencies else 0,
"p99_latency_ms": sorted(latencies)[int(len(latencies) * 0.99)] if latencies else 0,
}
使用示例
if __name__ == "__main__":
monitor = ProductionMonitor()
# 模拟请求监控
for i in range(100):
import random
monitor.record_request(
request_id=f"req_{i}",
latency_ms=random.uniform(50, 800), # 模拟不同延迟
success=random.random() > 0.05, # 模拟 5% 错误率
schema_valid=random.random() > 0.02 # 模拟 2% schema 违规
)
print("📊 监控指标汇总:")
print(json.dumps(monitor.get_metrics_summary(), indent=2, ensure_ascii=False))在 CI/CD 流程中集成契约测试
我强烈建议将契约测试集成到持续集成/持续部署流程中。这样可以在模型或 API 版本变更时自动验证兼容性，避免有问题的版本部署到生产环境。
# .github/workflows/ai-contract-test.yml
name: AI Service Contract Tests
on:
push:
branches: [main, develop]
pull_request:
branches: [main]
schedule:
# 每天凌晨 3 点运行一次基线测试
- cron: '0 3 * * *'
jobs:
contract-test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
pip install pytest pytest-asyncio aiohttp jsonschema redis requests
- name: Run Contract Tests
env:
HOLYSHEEP_API_KEY: ${{ secrets.HOLYSHEEP_API_KEY }}
run: |
pytest tests/test_contracts.py -v --tb=short --junitxml=results.xml
- name: Upload Test Results
uses: actions/upload-artifact@v3
if: always()
with:
name: contract-test-results
path: results.xml
- name: Publish to Dashboard
if: github.ref == 'refs/heads/main'
run: |
# 将测试结果发布到监控平台
curl -X POST "${{ secrets.DASHBOARD_WEBHOOK }}" \
-H "Content-Type: application/json" \
-d "{\"test_results\": \"passed\", \"commit\": \"${{ github.sha }}\"}"在实际项目中，我会为不同的 API 版本维护不同的契约定义。当 HolySheep AI 平台更新模型版本时（比如从 GPT-4.1 切换到新版本），我只需要更新对应的契约配置，而不需要修改测试逻辑。这种设计让我在 2025 年 Q4 成功完成了 7 次模型版本升级，每次升级都实现了零故障切换。
常见报错排查
错误 1：Schema 验证失败 - missing required field 'usage'
问题描述：API 返回的响应缺少 usage 字段，导致契约测试失败。
原因分析：HolySheep AI 的部分端点在特定情况下（如流式响应）可能不会立即返回 usage 信息。
解决方案：更新契约定义，使 usage 字段变为可选，或者在测试代码中处理这种情况：
# 修改后的 schema 定义
RESPONSE_SCHEMA_FLEXIBLE = {
"type": "object",
"required": ["choices", "model"],
"properties": {
"choices": {"type": "array", "minItems": 1},
"model": {"type": "string"},
"usage": { # 改为可选
"type": "object",
"properties": {
"prompt_tokens": {"type": "integer"},
"completion_tokens": {"type": "integer"},
"total_tokens": {"type": "integer"}
}
}
}
}
在测试代码中添加兼容性处理
def safe_validate_response(data: dict) -> tuple[bool, str]:
"""安全地验证响应，支持部分字段缺失"""
try:
# 首先尝试严格模式验证
jsonschema.validate(instance=data, schema=RESPONSE_SCHEMA)
return True, "严格验证通过"
except jsonschema.ValidationError as e:
# 如果严格模式失败，尝试宽松模式
try:
jsonschema.validate(instance=data, schema=RESPONSE_SCHEMA_FLEXIBLE)
return True, f"宽松验证通过 (严格模式失败: {e.message})"
except jsonschema.ValidationError:
return False, f"验证失败: {e.message}"错误 2：超时错误 - TimeoutError: API call exceeded 30s
问题描述：在高并发测试时，大量请求超时，测试失败。
原因分析：默认 30 秒超时对于并发场景过于严格，或者触发了 API 的限流机制。
解决方案：实现指数退避重试机制和动态超时配置：
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential
class ResilientHolySheepClient(HolySheepClient):
"""带重试机制的 HolySheep 客户端"""
async def chat_completion_with_retry(
self,
messages: list,
model: str = "gpt-4.1",
max_retries: int = 3,
base_timeout: float = 30.0
) -> AIResponse:
"""带指数退避重试的 API 调用"""
last_exception = None
for attempt in range(max_retries):
try:
# 动态调整超时时间（随重试次数增加）
timeout = base_timeout * (1 + attempt * 0.5)
response = await self.chat_completion(
messages=messages,
model=model,
timeout=timeout
)
return response
except asyncio.TimeoutError:
last_exception = asyncio.TimeoutError(
f"Attempt {attempt + 1}/{max_retries}: 请求超时 (timeout={timeout:.1f}s)"
)
print(f"⚠️ {last_exception}")
except requests.exceptions.HTTPError as e:
if e.response.status_code == 429: # Rate limit
last_exception = e
wait_time = 2 ** attempt # 指数退避等待
print(f"⚠️ Rate limit triggered, waiting {wait_time}s...")
await asyncio.sleep(wait_time)
else:
raise
raise RuntimeError(f"所有重试均失败: {last_exception}")
使用示例
async def test_with_retry():
client = ResilientHolySheepClient(api_key="YOUR_HOLYSHEEP_API_KEY")
await client.create_session()
try:
response = await client.chat_completion_with_retry(
messages=[{"role": "user", "content": "测试消息"}],
max_retries=5
)
print(f"✅ 成功: {response.content[:50]}...")
finally:
await client.close()错误 3：并发测试时返回不一致的内容
问题描述：相同的输入在并发测试中返回了不同的内容，导致语义契约测试失败。
原因分析：LLM 本身具有随机性，即使设置 temperature=0 也可能在极端情况下产生微小差异，或者触发了不同的模型实例。
解决方案：调整语义验证的容错度，使用更宽松的匹配规则：
import re
from difflib import SequenceMatcher
class SemanticValidator:
"""语义契约验证器 - 支持模糊匹配"""
def __init__(self, strict_mode: bool = False):
self.strict_mode = strict_mode
def compute_similarity(self, text1: str, text2: str) -> float:
"""计算两段文本的相似度"""
return SequenceMatcher(None, text1.lower(), text2.lower()).ratio()
def extract_entities(self, text: str) -> set:
"""提取文本中的关键实体（订单号、金额、日期等）"""
entities = set()
# 订单号模式
order_pattern = r'[A-Z]{1,3}[0-9]{6,12}'
entities.update(re.findall(order_pattern, text))
# 金额模式
amount_pattern = r'¥?\d+\.?\d*[元块]?'
entities.update(re.findall(amount_pattern, text))
# 日期模式
date_pattern = r'\d{4}[-/年]\d{1,2}[-/月]\d{1,2}[日]?'
entities.update(re.findall(date_pattern, text))
return entities
def validate_semantic(
self,
actual_output: str,
expected_keywords: list[str],
min_keyword_match: float = 0.6,
min_entity_match: float = 0.8
) -> tuple[bool, dict]:
"""
语义验证
返回: (是否通过, 详细结果)
"""
results = {
"keyword_match_rate": 0.0,
"entity_match_rate": 0.0,
"matched_keywords": [],
"missing_keywords": [],
"matched_entities": [],
"missing_entities": []
}
actual_lower = actual_output.lower()
# 关键词匹配
for keyword in expected_keywords:
if keyword.lower() in actual_lower:
results["matched_keywords"].append(keyword)
else:
results["missing_keywords"].append(keyword)
keyword_rate = len(results["matched_keywords"]) / len(expected_keywords) if expected_keywords else 1.0
results["keyword_match_rate"] = keyword_rate
# 实体匹配
expected_entities = self.extract_entities(" ".join(expected_keywords))
actual_entities = self.extract_entities(actual_output)
if expected_entities:
matched = expected_entities & actual_entities
results["matched_entities"] = list(matched)
results["missing_entities"] = list(expected_entities - actual_entities)
entity_rate = len(matched) / len(expected_entities)
results["entity_match_rate"] = entity_rate
else:
results["entity_match_rate"] = 1.0
# 综合判断
if self.strict_mode:
passed = keyword_rate >= min_keyword_match and entity_rate >= min_entity_match
else:
# 宽松模式：关键词或实体匹配其一即可
passed = keyword_rate >= min_keyword_match * 0.8 or entity_rate >= min_entity_match
return passed, results
使用示例
validator = SemanticValidator(strict_mode=False)
test_output = "您的订单 A123456789 正在配送中，预计 3-5 个工作日送达。金额 ¥299.00 已扣款。"
keywords = ["订单", "配送", "送达", "金额"]
passed, details = validator.validate_semantic(
actual_output=test_output,
expected_keywords=keywords,
min_keyword_match=0.6
)
print(f"验证结果: {'✅ 通过' if passed else '❌ 失败'}")
print(f"关键词匹配率: {details['keyword_match_rate']*100:.0f}%")
print(f"实体匹配率: {details['entity_match_rate']*100:.0f}%")
print(f"缺失关键词: {details['missing_keywords']}")错误 4：费用异常增长 - 月账单超出预算 300%
问题描述：某天突然发现 API 费用暴涨，排查后发现是测试环境忘记关闭，大规模并发测试导致 Token 消耗激增。
原因分析：缺少费用监控和预算告警机制。
解决方案：实现实时费用监控和自动熔断：
import time
from collections import defaultdict
from