API 契约测试在大模型服务中的应用：电商促销场景实战

2. 核心契约测试用例

import requests
import pytest

class TestHolySheepAPIContract:
    """HolySheep AI API 契约测试套件"""
    
    @pytest.fixture(autouse=True)
    def setup(self):
        """测试前置配置"""
        self.base_url = "https://api.holysheep.ai/v1"
        self.api_key = "YOUR_HOLYSHEEP_API_KEY"
        self.headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        self.tester = HolySheepContractTester(self.api_key)
    
    def test_basic_chat_completion_contract(self):
        """测试用例 1: 基础对话补全契约"""
        payload = {
            "model": "gpt-4.1",
            "messages": [
                {"role": "system", "content": "你是一个电商客服助手"},
                {"role": "user", "content": "请问双十一有哪些优惠活动？"}
            ],
            "temperature": 0.7,
            "max_tokens": 500
        }
        
        start_time = time.time()
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=self.headers,
            json=payload,
            timeout=10
        )
        latency_ms = (time.time() - start_time) * 1000
        
        assert response.status_code == 200, f"Expected 200, got {response.status_code}"
        
        data = response.json()
        
        # 契约验证：响应结构
        is_valid, errors = self.tester.verify_response_fields(data)
        assert is_valid, f"Contract violation: {', '.join(errors)}"
        
        # 契约验证：延迟 SLA
        assert latency_ms < 800, f"Latency {latency_ms:.2f}ms exceeds SLA of 800ms"
        
        # 契约验证：内容质量
        assert len(data["choices"][0]["message"]["content"]) > 0, "Response content is empty"
        
        print(f"✓ Basic chat completion passed: {latency_ms:.2f}ms, {data['usage']['total_tokens']} tokens")
    
    def test_streaming_response_contract(self):
        """测试用例 2: 流式响应契约验证"""
        payload = {
            "model": "deepseek-v3.2",
            "messages": [{"role": "user", "content": "写一段产品描述"}],
            "stream": True,
            "max_tokens": 200
        }
        
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=self.headers,
            json=payload,
            stream=True,
            timeout=15
        )
        
        assert response.status_code == 200
        
        stream_chunks = []
        received_fields = set()
        
        for line in response.iter_lines():
            if line:
                line_text = line.decode('utf-8')
                if line_text.startswith('data: '):
                    data_str = line_text[6:]
                    if data_str == '[DONE]':
                        break
                    chunk = json.loads(data_str)
                    stream_chunks.append(chunk)
                    
                    # 收集字段用于契约验证
                    if 'choices' in chunk:
                        for choice in chunk['choices']:
                            if 'delta' in choice:
                                received_fields.update(choice['delta'].keys())
        
        # 契约验证：流式响应必须包含 delta 字段
        assert len(stream_chunks) > 0, "No streaming chunks received"
        assert 'delta' in received_fields, "Streaming response missing 'delta' field"
        
        print(f"✓ Streaming contract passed: {len(stream_chunks)} chunks")
    
    def test_concurrent_requests_stability(self):
        """测试用例 3: 并发稳定性契约（模拟双十一场景）"""
        import concurrent.futures
        
        def send_request(request_id: int) -> ContractTestResult:
            payload = {
                "model": "gpt-4.1",
                "messages": [
                    {"role": "user", "content": f"订单号{request_id}的物流状态"}
                ],
                "max_tokens": 100
            }
            
            start = time.time()
            try:
                response = requests.post(
                    f"{self.base_url}/chat/completions",
                    headers=self.headers,
                    json=payload,
                    timeout=10
                )
                latency = (time.time() - start) * 1000
                
                if response.status_code != 200:
                    return ContractTestResult(
                        contract_name="concurrent_stability",
                        passed=False,
                        latency_ms=latency,
                        error_message=f"HTTP {response.status_code}"
                    )
                
                data = response.json()
                is_valid, errors = self.tester.verify_response_fields(data)
                
                return ContractTestResult(
                    contract_name="concurrent_stability",
                    passed=is_valid,
                    latency_ms=latency,
                    error_message=', '.join(errors) if errors else None,
                    response_data=data
                )
            except Exception as e:
                return ContractTestResult(
                    contract_name="concurrent_stability",
                    passed=False,
                    latency_ms=(time.time() - start) * 1000,
                    error_message=str(e)
                )
        
        # 模拟 100 并发请求（双十一流量的缩略版本）
        with concurrent.futures.ThreadPoolExecutor(max_workers=50) as executor:
            futures = [executor.submit(send_request, i) for i in range(100)]
            results = [f.result() for f in concurrent.futures.as_completed(futures)]
        
        passed = sum(1 for r in results if r.passed)
        failed = len(results) - passed
        avg_latency = sum(r.latency_ms for r in results) / len(results)
        max_latency = max(r.latency_ms for r in results)
        
        print(f"✓ Concurrency test: {passed}/100 passed, avg={avg_latency:.2f}ms, max={max_latency:.2f}ms")
        
        # 契约 SLA：成功率 >= 99%，平均延迟 < 500ms，最大延迟 < 2000ms
        assert passed >= 99, f"Success rate {passed}% below SLA of 99%"
        assert avg_latency < 500, f"Average latency {avg_latency:.2f}ms exceeds SLA"
        assert max_latency < 2000, f"Max latency {max_latency:.2f}ms exceeds threshold"
    
    def test_cost_estimation_contract(self):
        """测试用例 4: 成本估算契约（HolySheep 汇率优势验证）"""
        payload = {
            "model": "gpt-4.1",
            "messages": [{"role": "user", "content": "解释量子计算原理"}],
            "max_tokens": 1000
        }
        
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=self.headers,
            json=payload
        )
        
        data = response.json()
        usage = data.get("usage", {})
        
        prompt_tokens = usage.get("prompt_tokens", 0)
        completion_tokens = usage.get("completion_tokens", 0)
        total_tokens = usage.get("total_tokens", 0)
        
        # HolySheep 价格表（2026年主流模型 output 价格）
        price_per_mtok = {
            "gpt-4.1": 8.0,           # $8/MTok
            "claude-sonnet-4.5": 15.0, # $15/MTok
            "gemini-2.5-flash": 2.50,  # $2.50/MTok
            "deepseek-v3.2": 0.42     # $0.42/MTok
        }
        
        model = data.get("model", "gpt-4.1")
        cost_usd = (completion_tokens / 1_000_000) * price_per_mtok.get(model, 8.0)
        
        # HolySheep 汇率：¥1=$1，实际成本即为美元金额
        # 对比官方：¥7.3=$1，节省超过 85%
        cost_cny = cost_usd  # HolySheep 无损汇率
        
        print(f"✓ Cost estimation: {completion_tokens} output tokens, ${cost_usd:.4f} (~¥{cost_cny:.4f})")
        
        # 契约验证：成本必须在合理范围内
        assert cost_usd < 0.02, f"Cost ${cost_usd} exceeds budget"
        assert total_tokens == prompt_tokens + completion_tokens, "Token count mismatch"

# .github/workflows/api-contract-test.yml
name: AI API Contract Tests

on:
  push:
    branches: [main, develop]
  schedule:
    # 每日凌晨执行完整契约测试
    - cron: '0 2 * * *'
  pull_request:
    branches: [main]

jobs:
  contract-test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ['3.10', '3.11']
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Set up Python ${{ matrix.python-version }}
        uses: actions/setup-python@v5
        with:
          python-version: ${{ matrix.python-version }}
      
      - name: Install dependencies
        run: |
          pip install pytest requests pytest-cov pytest-xdist
          pip install -r requirements.txt
      
      - name: Run contract tests with HolySheep AI
        env:
          HOLYSHEEP_API_KEY: ${{ secrets.HOLYSHEEP_API_KEY }}
        run: |
          pytest tests/contract/ \
            --verbose \
            --cov=src \
            --cov-report=xml \
            --cov-report=html \
            --junitxml=results.xml
      
      - name: Upload test results
        uses: actions/upload-artifact@v4
        with:
          name: contract-test-results
          path: results.xml
      
      - name: Slack notification on failure
        if: failure()
        uses: slackapi/slack-github-action@v1
        with:
          payload: |
            {
              "text": "🚨 API Contract Test Failed!",
              "blocks": [
                {
                  "type": "section",
                  "text": {
                    "type": "mrkdwn",
                    "text": "*HolySheep AI API 契约测试失败*\n日期: ${{ github.event.head_commit.timestamp }}"
                  }
                }
              ]
            }

# 错误信息
requests.exceptions.HTTPError: 401 Client Error: Unauthorized

原因分析
1. API Key 未正确设置或为空
2. API Key 格式错误（缺少 Bearer 前缀）
3. 使用了占位符而非真实密钥

解决方案
import os

✅ 正确方式：从环境变量读取
api_key = os.environ.get("HOLYSHEEP_API_KEY")
if not api_key or api_key == "YOUR_HOLYSHEEP_API_KEY":
    raise ValueError("请设置有效的 HOLYSHEEP_API_KEY 环境变量")

headers = {
    "Authorization": f"Bearer {api_key}",  # 必须包含 Bearer 前缀
    "Content-Type": "application/json"
}

✅ 验证密钥格式
assert api_key.startswith("sk-"), "HolySheep API Key 必须以 sk- 开头"

✅ 测试连接
response = requests.get(
    "https://api.holysheep.ai/v1/models",
    headers=headers
)
print(f"连接成功，可用模型: {response.json()}")

# 错误信息
{"error": {"message": "Invalid request payload", "type": "invalid_request_error"}}

常见原因及解决方案

原因 1: messages 字段格式错误
❌ 错误
payload = {"messages": "你好"}  # 字符串而非数组

✅ 正确
payload = {
    "model": "gpt-4.1",
    "messages": [
        {"role": "system", "content": "你是助手"},
        {"role": "user", "content": "你好"}
    ]
}

原因 2: temperature/max_tokens 超出范围
❌ 错误
payload = {"temperature": 3.0}  # 超出 0-2 范围

✅ 正确
payload = {
    "model": "gpt-4.1",
    "messages": [...],
    "temperature": 0.7,   # 有效范围 0-2
    "max_tokens": 1000   # 有效范围 1-模型最大值
}

原因 3: model 名称拼写错误
❌ 错误
payload = {"model": "gpt-4"}  # 不是有效模型名

✅ 正确（2026年主流模型）
valid_models = [
    "gpt-4.1",
    "claude-sonnet-4.5",
    "gemini-2.5-flash",
    "deepseek-v3.2"
]
payload = {"model": "gpt-4.1"}

通用验证函数
def validate_request_payload(payload: dict) -> list:
    """验证请求 payload，返回错误列表"""
    errors = []
    
    if "messages" not in payload:
        errors.append("Missing required field: messages")
    elif not isinstance(payload["messages"], list):
        errors.append("Field 'messages' must be an array")
    elif len(payload["messages"]) == 0:
        errors.append("Field 'messages' cannot be empty")
    
    if "model" not in payload:
        errors.append("Missing required field: model")
    elif payload["model"] not in valid_models:
        errors.append(f"Invalid model: {payload['model']}")
    
    if "temperature" in payload:
        if not 0 <= payload["temperature"] <= 2:
            errors.append("temperature must be between 0 and 2")
    
    if "max_tokens" in payload:
        if not 1 <= payload["max_tokens"] <= 128000:
            errors.append("max_tokens out of valid range")
    
    return errors

使用验证
errors = validate_request_payload(payload)
if errors:
    raise ValueError(f"Payload validation failed: {', '.join(errors)}")

# 错误信息
{"error": {"message": "Rate limit exceeded", "type": "rate_limit_error"}}

解决方案：实现指数退避重试机制

import time
import random
from functools import wraps

def retry_with_exponential_backoff(
    max_retries=5,
    base_delay=1.0,
    max_delay=60.0,
    jitter=True
):
    """指数退避装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    
                    if "429" in str(e) or "rate limit" in str(e).lower():
                        # 计算延迟时间
                        delay = min(base_delay * (2 ** attempt), max_delay)
                        
                        # 添加随机抖动（0.5-1.5倍）
                        if jitter:
                            delay = delay * (0.5 + random.random())
                        
                        print(f"⏳ Rate limited, retrying in {delay:.2f}s (attempt {attempt+1}/{max_retries})")
                        time.sleep(delay)
                    else:
                        # 非限流错误，直接重试
                        time.sleep(base_delay)
            
            raise last_exception
        return wrapper
    return decorator

@retry_with_exponential_backoff(max_retries=5, base_delay=2.0)
def call_holysheep_api_with_retry(payload: dict) -> dict:
    """带重试的 HolySheep API 调用"""
    response = requests.post(
        "https://api.holysheep.ai/v1/chat/completions",
        headers={
            "Authorization": f"Bearer {os.environ.get('HOLYSHEEP_API_KEY')}",
            "Content-Type": "application/json"
        },
        json=payload,
        timeout=30
    )
    
    if response.status_code == 429:
        # 读取重试建议
        retry_after = response.headers.get("Retry-After", "60")
        wait_time = int(retry_after)
        print(f"⏳ Server suggests waiting {wait_time}s")
        time.sleep(wait_time)
        raise Exception("429 Rate Limit")  # 触发装饰器重试
    
    response.raise_for_status()
    return response.json()

使用示例
result = call_holysheep_api_with_retry({
    "model": "gpt-4.1",
    "messages": [{"role": "user", "content": "你好"}]
})