AI营销中台服务合同签订指南：五项可量化验收条款的设计范式

# 交付物清单 - 结构化定义
from dataclasses import dataclass
from enum import Enum
from typing import List

class DeliverableType(Enum):
    API_SPEC = "api_spec"
    DASHBOARD = "dashboard"
    MODEL_WEIGHT = "model_weight"
    AVATAR_ASSET = "avatar_asset"
    CONFIG = "config"
    REPORT = "report"

@dataclass
class DeliverableItem:
    id: str                          # 唯一标识，如 "DEL-001"
    name: str                        # 交付物名称
    type: DeliverableType            # 交付物类型
    storage_uri: str                 # 存储路径，如 "cos://bucket/ai-model/v1.0/"
    checksum: str                    # SHA256校验值，确保完整性
    dependencies: List[str]          # 依赖清单，如 ["pytorch==2.0.1", "transformers>=4.30.0"]
    expected_artifact_count: int     # 预期文件数量
    format_spec: str                 # 格式规范，如 "JSONL", "ONNX", "MP4"

from datetime import datetime, timedelta

@dataclass
class Milestone:
    name: str                        # 里程碑名称，如 "环境就绪"
    trigger_condition: str           # 触发条件，如 "甲方已提供全部API凭证"
    t0_base: datetime                # 以触发条件满足日为 T+0
    due_delta_days: int              # 相对T+0的截止天数
    penalty_ratio: float             # 延期惩罚比例
    required_throughput_qps: int     # 要求的吞吐量阈值

# SLA 不可用判定逻辑
def is_service_unavailable(
    error_rate: float,               # 过去5分钟的调用错误率
    duration_seconds: int,           # 持续时间
    threshold_rate: float = 0.05,    # 默认 5%
    threshold_duration: int = 300    # 默认 300秒
) -> bool:
    """判定服务是否处于不可用状态"""
    return error_rate > threshold_rate and duration_seconds >= threshold_duration

# 自动扣款触发
def calculate_penalty(milestone: Milestone, actual_delivery: datetime) -> float:
    """计算延期惩罚金额比例"""
    delay_days = (actual_delivery - milestone.t0_base).days - milestone.due_delta_days
    if delay_days > 0:
        return min(delay_days * milestone.penalty_ratio, 0.20)  # 封顶20%
    return 0.0

import re
import time
from typing import List, Tuple, Dict

# ========== 验收标准 1：事实性幻觉检测 ==========
def extract_claims(text: str) -> List[str]:
    """从生成文本中抽取事实性断言（NLP抽取）"""
    # 实际实现可使用NER或关系抽取模型
    # 此处为伪代码示意
    return re.findall(r'[^。;]*?(?:是|有|达到|超过)[^。;]*?。', text)

def verify_against_kb(claim: str, kb_entries: List[dict]) -> bool:
    """验证断言是否与知识库一致"""
    for entry in kb_entries:
        if claim.strip() in entry.get("content", ""):
            return True
    return False

def validate_factual_accuracy(
    generated_text: str,
    knowledge_base_entries: List[dict],
    hallucination_threshold: float = 0.03   # 幻觉率不超过3%
) -> Tuple[bool, dict]:
    """校验生成内容是否与知识库一致"""
    claims = extract_claims(generated_text)
    if not claims:
        return True, {"total_claims": 0, "errors": 0, "rate": 0.0}
    
    errors = sum(1 for claim in claims if not verify_against_kb(claim, knowledge_base_entries))
    hallucination_rate = errors / len(claims)
    passed = hallucination_rate <= hallucination_threshold
    return passed, {
        "total_claims": len(claims),
        "errors": errors,
        "hallucination_rate": round(hallucination_rate, 4)
    }

# ========== 验收标准 2：数字人唇音同步误差检测 ==========
def extract_lip_motion_timestamps(video_path: str) -> List[float]:
    """提取唇部运动时间戳（使用MediaPipe或类似库）"""
    # 实际实现调用CV库
    return [0.0, 0.5, 1.0, 1.5]  # 占位返回

def extract_phoneme_timestamps(audio_transcript: str) -> List[float]:
    """提取音素时间戳"""
    return [0.05, 0.48, 0.98, 1.52]  # 占位返回

def align_timestamps(lip_ts: List[float], audio_ts: List[float]) -> List[Tuple[float, float]]:
    """对齐音视频时间戳"""
    return list(zip(lip_ts, audio_ts))

def validate_lip_sync(
    video_path: str,
    audio_transcript: str,
    max_latency_ms: int = 150
) -> Tuple[bool, dict]:
    """检测音视频同步延迟"""
    lip_movement_ts = extract_lip_motion_timestamps(video_path)
    audio_phoneme_ts = extract_phoneme_timestamps(audio_transcript)
    
    if not lip_movement_ts or not audio_phoneme_ts:
        return False, {"error": "无法提取时间戳"}
    
    sync_latencies = [
        abs(lip_ts - audio_ts) * 1000  # 转换为毫秒
        for lip_ts, audio_ts in align_timestamps(lip_movement_ts, audio_phoneme_ts)
    ]
    avg_latency = sum(sync_latencies) / len(sync_latencies) if sync_latencies else 999
    max_latency = max(sync_latencies) if sync_latencies else 999
    
    passed = avg_latency <= max_latency_ms
    return passed, {
        "avg_latency_ms": round(avg_latency, 2),
        "max_latency_ms": round(max_latency, 2),
        "samples": len(sync_latencies)
    }

# ========== 验收标准 3：API响应性能与成功率验收 ==========
def validate_schema(response: dict, expected_schema: dict) -> bool:
    """校验响应是否符合Schema定义"""
    for key, expected_type in expected_schema.items():
        if key not in response:
            return False
        if not isinstance(response[key], expected_type):
            return False
    return True

def call_api(request: dict) -> dict:
    """模拟API调用"""
    # 实际实现发送HTTP请求
    return {"status": 200, "data": {"result": "success"}}

def validate_api_sla(
    test_requests: List[dict],
    expected_schema: dict,
    max_response_time_ms: int = 2000,
    min_success_rate: float = 0.99
) -> Tuple[bool, dict]:
    """批量API验收测试"""
    results = []
    for req in test_requests:
        start = time.time()
        resp = call_api(req)
        elapsed = (time.time() - start) * 1000  # 转换为毫秒
        valid = resp.get("status") == 200 and validate_schema(resp, expected_schema)
        results.append({"success": valid, "latency": elapsed})
    
    success_count = sum(1 for r in results if r["success"])
    success_rate = success_count / len(results) if results else 0.0
    avg_latency = sum(r["latency"] for r in results) / len(results) if results else 999
    
    passed = (success_rate >= min_success_rate) and (avg_latency <= max_response_time_ms)
    return passed, {
        "success_rate": round(success_rate, 4),
        "avg_latency_ms": round(avg_latency, 2),
        "total_requests": len(results)
    }

import hashlib
from datetime import datetime, timedelta
from typing import List, Tuple

# ========== 数据主权：导出格式可迁移性验证 ==========
REQUIRED_FIELDS = ["user_id", "interaction_id", "prompt", "response", "timestamp", "model_version"]

def validate_data_export(export_package: dict) -> Tuple[bool, dict]:
    """验证数据导出格式符合可迁移性要求"""
    errors = []
    
    # 检查格式
    if export_package.get("format") != "JSONL":
        errors.append("数据格式需为 JSONL")
    
    # 检查Schema完整性
    sample_record = export_package.get("sample_record", {})
    missing_fields = [f for f in REQUIRED_FIELDS if f not in sample_record]
    if missing_fields:
        errors.append(f"导出数据缺少必要字段: {missing_fields}")
    
    # 检查是否包含明文敏感信息（手机号、身份证等）
    import re
    phone_pattern = r'1[3-9]\d{9}'
    sensitive_found = []
    for record in export_package.get("records", [])[:10]:  # 抽样检查
        if re.search(phone_pattern, str(record)):
            sensitive_found.append(record.get("user_id", "unknown"))
    
    if sensitive_found:
        errors.append(f"导出数据包含明文手机号: {sensitive_found}")
    
    passed = len(errors) == 0
    return passed, {"errors": errors, "total_records": export_package.get("total_count", 0)}

# ========== 数据主权：删除验证 ==========
def request_audit_log(bucket_uri: str, after_date: datetime) -> List[dict]:
    """请求对象存储的删除操作审计日志"""
    # 实际实现调用云厂商的审计API
    return []  # 占位返回

def fetch_object(uri: str, timeout: int = 5) -> object:
    """尝试获取对象"""
    # 实际实现发送HTTP GET请求
    # 返回模拟的响应对象
    class MockResponse:
        status_code = 404
        content = b""
    return MockResponse()

def verify_data_deletion(
    bucket_uri: str,
    known_file_hashes: List[Tuple[str, str]],  # [(uri, expected_sha256)]
    deletion_window_days: int = 7
) -> Tuple[bool, dict]:
    """
    合同终止后，验证服务方是否彻底删除数据
    通过抽样核查方式确认
    """
    deleted_uris = []
    retained_uris = []
    
    # 要求乙方提供对象存储的删除操作审计日志
    audit_log = request_audit_log(
        bucket_uri,
        after_date=datetime.now() - timedelta(days=deletion_window_days)
    )
    
    for uri, expected_hash in known_file_hashes:
        try:
            resp = fetch_object(uri, timeout=5)
            if resp.status_code == 200:
                actual_hash = hashlib.sha256(resp.content).hexdigest()
                if actual_hash == expected_hash:
                    retained_uris.append(uri)   # 数据仍存在，未删除
                else:
                    deleted_uris.append(uri)     # 内容已被覆盖
            elif resp.status_code == 404:
                deleted_uris.append(uri)         # 已删除
            else:
                deleted_uris.append(uri)         # 不可达，视为已清理
        except Exception:
            deleted_uris.append(uri)             # 异常，视为已清理
    
    passed = len(retained_uris) == 0
    return passed, {
        "deleted_count": len(deleted_uris),
        "retained_count": len(retained_uris),
        "retained_uris": retained_uris[:10],     # 仅返回前10个，避免数据量过大
        "audit_log_available": len(audit_log) > 0
    }

from enum import Enum
from typing import Dict, List

class TerminationLevel(Enum):
    NORMAL = 0          # 正常履约
    OBSERVATION = 1     # 进入观察期
    TERMINATE = 2       # 触发终止

@dataclass
class IncidentRecord:
    timestamp: datetime
    duration_seconds: int
    resolution: str
    affected_service: str

def evaluate_contract_status(
    milestone_records: Dict[str, bool],        # 各里程碑通过情况 {milestone_name: passed}
    incident_log: List[IncidentRecord],        # 故障记录
    max_incident_duration_24h: int = 1800,     # 24小时内最长允许故障时长（秒）
    consecutive_failures_threshold: int = 2,   # 连续阶段性验收不通过次数
    observation_window_days: int = 14          # 观察期长度
) -> TerminationLevel:
    """合同状态自动评估"""
    
    # 条件1：检查累计不可用时长
    now = datetime.now()
    last_24h_incidents = [
        inc for inc in incident_log 
        if (now - inc.timestamp).total_seconds() <= 86400
    ]
    total_downtime = sum(inc.duration_seconds for inc in last_24h_incidents)
    
    if total_downtime > max_incident_duration_24h:
        return TerminationLevel.OBSERVATION
    
    # 条件2：检查连续验收失败次数
    if len(milestone_records) >= consecutive_failures_threshold:
        milestone_names = list(milestone_records.keys())
        last_n_results = [
            milestone_records[name] 
            for name in milestone_names[-consecutive_failures_threshold:]
        ]
        # 最近N次全部失败 -> 触发终止
        if not any(last_n_results):
            return TerminationLevel.TERMINATE
    
    return TerminationLevel.NORMAL

# ========== 终止后清算逻辑 ==========
def calculate_refund(
    contract_value: float,
    terminated_phase: int,          # 终止时已执行到第几阶段（从1开始）
    total_phases: int,
    breach_penalty_rate: float = 0.10   # 违约扣除比例（甲方违约）
) -> dict:
    """
    阶梯式退款计算
    返回: {"refund_amount": float, "formula": str}
    """
    if terminated_phase <= 1:
        refund = contract_value * 0.90
        formula = f"90%（执行不足1个阶段）"
    elif terminated_phase >= total_phases:
        refund = 0.0
        formula = "0%（已全部交付）"
    else:
        remaining_ratio = (total_phases - terminated_phase) / total_phases
        refund = contract_value * remaining_ratio * (1 - breach_penalty_rate)
        formula = f"{round(remaining_ratio * 100)}% × (1-{breach_penalty_rate*100}%)"
    
    return {
        "refund_amount": round(refund, 2),
        "formula": formula,
        "terminated_phase": terminated_phase,
        "total_phases": total_phases
    }

# ========== 示例：合同状态评估主流程 ==========
def main_contract_health_check(
    contract: dict,
    milestone_status: Dict[str, bool],
    incidents: List[IncidentRecord]
) -> dict:
    """合同健康检查主入口"""
    level = evaluate_contract_status(milestone_status, incidents)
    
    response = {
        "status": level.name,
        "timestamp": datetime.now().isoformat(),
        "recommendation": ""
    }
    
    if level == TerminationLevel.NORMAL:
        response["recommendation"] = "继续履约，按计划推进下一阶段"
    elif level == TerminationLevel.OBSERVATION:
        response["recommendation"] = f"进入{14}天观察期，需在观察期内完成整改"
    else:  # TERMINATE
        refund_info = calculate_refund(
            contract_value=contract.get("total_value", 0.0),
            terminated_phase=contract.get("current_phase", 1),
            total_phases=contract.get("total_phases", 4)
        )
        response["recommendation"] = "触发终止条件，启动退出清算流程"
        response["refund"] = refund_info
    
    return response

import pytest
from typing import List, Dict
import requests

# ========== 测试配置 ==========
TEST_CONFIG = {
    "api_base_url": "https://api.example.com/v1",
    "test_timeout": 30,
    "knowledge_base_endpoint": "/kb/verify",
    "inference_endpoint": "/agent/chat",
    "export_endpoint": "/data/export"
}

# ========== 测试用例集合 ==========
class TestAIMarketingContract:
    """合同验收测试集 - 所有用例通过即视为合同履约完成"""
    
    def setup_method(self):
        """每个测试用例执行前的初始化"""
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {os.getenv('TEST_API_KEY')}",
            "X-Request-ID": f"acceptance-test-{datetime.now().strftime('%Y%m%d%H%M%S')}"
        })
    
    # -------- 测试组1：交付物完整性 --------
    def test_deliverable_integrity(self):
        """测试用例 TC-001：校验交付物清单完整性与校验和"""
        deliverable_list = self._fetch_deliverable_manifest()
        assert len(deliverable_list) > 0, "交付物清单为空"
        
        for item in deliverable_list:
            # 校验存储路径可访问
            resp = self.session.head(item["storage_uri"], timeout=10)
            assert resp.status_code in [200, 302], f"交付物 {item['id']} 不可访问"
            
            # 校验文件完整性（SHA256）
            if item.get("checksum"):
                resp = self.session.get(item["storage_uri"], timeout=30)
                actual_sha = hashlib.sha256(resp.content).hexdigest()
                assert actual_sha == item["checksum"], \
                    f"交付物 {item['id']} 完整性校验失败"
    
    # -------- 测试组2：模型推理准确性 --------
    def test_model_inference_accuracy(self):
        """测试用例 TC-002：模型推理准确性回归测试"""
        test_cases = self._load_golden_dataset()
        failures = []
        
        for case in test_cases:
            resp = self.session.post(
                f"{TEST_CONFIG['api_base_url']}{TEST_CONFIG['inference_endpoint']}",
                json={"prompt": case["input"], "temperature": 0.0}
            )
            assert resp.status_code == 200, f"推理请求失败: {case['id']}"
            
            result = resp.json()
            if result.get("confidence", 0) < case.get("threshold", 0.85):
                failures.append({
                    "case_id": case["id"],
                    "expected_threshold": case.get("threshold", 0.85),
                    "actual_confidence": result.get("confidence", 0)
                })
        
        assert len(failures) == 0, f"以下用例未通过置信度阈值: {failures}"
    
    # -------- 测试组3：API性能与成功率 --------
    def test_api_performance_sla(self):
        """测试用例 TC-003：API响应性能与成功率验收"""
        test_payloads = self._generate_test_payloads(count=100)
        results = []
        
        for payload in test_payloads:
            start = time.time()
            resp = self.session.post(
                f"{TEST_CONFIG['api_base_url']}/agent/chat",
                json=payload,
                timeout=TEST_CONFIG["test_timeout"]
            )
            elapsed_ms = (time.time() - start) * 1000
            
            is_success = resp.status_code == 200 and resp.json().get("status") == "success"
            results.append({
                "success": is_success,
                "latency_ms": elapsed_ms
            })
        
        success_count = sum(1 for r in results if r["success"])
        success_rate = success_count / len(results)
        avg_latency = sum(r["latency_ms"] for r in results) / len(results)
        p95_latency = sorted(r["latency_ms"] for r in results)[int(len(results) * 0.95)]
        
        assert success_rate >= 0.99, f"成功率 {success_rate:.2%} 低于 99%"
        assert avg_latency <= 2000, f"平均延迟 {avg_latency:.0f}ms 超过 2000ms"
        assert p95_latency <= 3000, f"P95延迟 {p95_latency:.0f}ms 超过 3000ms"
    
    # -------- 测试组4：数据主权与可迁移性 --------
    def test_data_sovereignty_export(self):
        """测试用例 TC-004：验证数据导出格式符合可迁移性要求"""
        resp = self.session.post(
            f"{TEST_CONFIG['api_base_url']}{TEST_CONFIG['export_endpoint']}",
            json={"format": "JSONL", "fields": REQUIRED_FIELDS}
        )
        assert resp.status_code == 200, "数据导出请求失败"
        
        export_data = resp.json()
        assert export_data.get("format") == "JSONL", "数据格式需为 JSONL"
        
        # 检查导出文件是否包含必要字段
        sample = export_data.get("sample_records", [])
        assert len(sample) > 0, "导出数据为空"
        
        missing_fields = []
        for field in REQUIRED_FIELDS:
            if field not in sample[0]:
                missing_fields.append(field)
        assert len(missing_fields) == 0, f"缺少必要字段: {missing_fields}"
        
        # 检查是否包含明文敏感信息
        import re
        phone_pattern = r'1[3-9]\d{9}'
        id_card_pattern = r'\d{17}[\dXx]'
        
        sensitive_found = []
        for record in sample:
            record_str = str(record)
            if re.search(phone_pattern, record_str):
                sensitive_found.append("phone")
            if re.search(id_card_pattern, record_str):
                sensitive_found.append("id_card")
        
        assert len(sensitive_found) == 0, f"导出数据包含明文敏感信息: {set(sensitive_found)}"
    
    # -------- 测试组5：优雅降级与容错 --------
    def test_graceful_degradation(self):
        """测试用例 TC-005：服务组件异常时系统表现"""
        # 模拟知识库超时
        resp = self.session.post(
            f"{TEST_CONFIG['api_base_url']}{TEST_CONFIG['inference_endpoint']}",
            json={
                "prompt": "测试降级场景",
                "mock_fault": "knowledge_base_timeout",
                "timeout_ms": 5000
            }
        )
        
        assert resp.status_code == 200, "降级场景下应返回200"
        result = resp.json()
        # 预期：在知识库超时时，应返回缓存结果或兜底回答
        assert result.get("status") in ["CACHE_HIT", "FALLBACK"], \
            f"未按约定实现优雅降级，实际状态: {result.get('status')}"
    
    # -------- 辅助方法 --------
    def _fetch_deliverable_manifest(self) -> List[Dict]:
        """获取交付物清单"""
        resp = self.session.get(f"{TEST_CONFIG['api_base_url']}/manifest")
        return resp.json().get("items", [])
    
    def _load_golden_dataset(self) -> List[Dict]:
        """加载黄金测试数据集"""
        # 实际实现从文件或数据库加载
        return [
            {"id": "GC-001", "input": "请介绍产品的核心功能", "threshold": 0.85},
            {"id": "GC-002", "input": "与其他竞品相比的优势", "threshold": 0.80},
        ]
    
    def _generate_test_payloads(self, count: int) -> List[Dict]:
        """生成测试请求负载"""
        prompts = [
            "产品有哪些特点？",
            "价格是多少？",
            "如何购买？",
            "售后服务如何？",
            "有没有试用版？"
        ]
        import random
        return [
            {"prompt": random.choice(prompts), "temperature": 0.0}
            for _ in range(count)
        ]