HolySheep Agent SaaS 商业化清单：统一计费、客户隔离、模型降级与异常请求追踪

智能模型路由与降级策略

这是整套方案最有技术含量的部分。我的核心设计思路是：先用小模型做"探测"，根据任务复杂度动态选择是否升级到贵模型。

class SmartModelRouter:
    """
    智能模型路由 + 自动降级
    HolySheep 支持 8+ 主流模型，一次接入全覆盖
    """
    
    def __init__(self, billing_engine: HolySheepBillingEngine):
        self.engine = billing_engine
    
    async def classify_task_complexity(self, prompt: str, system: str) -> int:
        """
        返回任务复杂度等级 1-5
        1: 简单问答/翻译
        5: 复杂推理/代码生成
        """
        complexity_prompt = f"""
        分析以下任务复杂度，返回 1-5 的数字:
        1 = 简单问答、翻译、格式转换
        2 = 一般内容创作、摘要
        3 = 技术文档、代码审查
        4 = 复杂推理、架构设计
        5 = 多步推理、代码生成、数学证明
        
        System: {system[:200]}
        User: {prompt[:500]}
        
        只返回一个数字，不要解释。
        """
        
        # 用 DeepSeek V3.2 做分类，成本极低
        response = await self.engine.holysheep_client.post(
            "/chat/completions",
            json={
                "model": "deepseek-v3.2",
                "messages": [{"role": "user", "content": complexity_prompt}],
                "max_tokens": 5,
                "temperature": 0
            }
        )
        
        try:
            result = int(response.json()["choices"][0]["message"]["content"].strip())
            return min(max(result, 1), 5)
        except:
            return 3  # 默认中等复杂度
    
    async def select_model(
        self, 
        tenant: Tenant, 
        prompt: str, 
        system: str,
        force_model: Optional[str] = None
    ) -> str:
        """
        根据租户权限、任务复杂度、预算选择最优模型
        """
        if force_model and force_model in tenant.model_permissions:
            return force_model
        
        # 1. 判断任务复杂度
        complexity = await self.classify_task_complexity(prompt, system)
        
        # 2. 根据复杂度选择模型池
        if complexity <= 2:
            # 简单任务: 优先低成本模型
            candidates = ["deepseek-v3.2", "gpt-4o-mini", "gemini-2.5-flash"]
        elif complexity <= 4:
            # 中等任务: 平衡成本和效果
            candidates = ["gemini-2.5-flash", "gpt-4.1"]
        else:
            # 复杂任务: 用最强模型
            candidates = ["gpt-4.1", "claude-sonnet-4.5"]
        
        # 3. 选择租户有权限的最优候选
        for model in candidates:
            if model in tenant.model_permissions:
                return model
        
        # 兜底: 用租户最低权限模型
        return tenant.model_permissions[0]
    
    async def execute_with_fallback(
        self,
        tenant: Tenant,
        messages: List[Dict],
        max_cost_estimate: float = 0.50
    ) -> Dict:
        """
        执行请求，失败时自动降级
        HolySheep 国内直连延迟 < 50ms，Fallback 体验流畅
        """
        primary_model = await self.select_model(
            tenant,
            messages[-1]["content"],
            messages[0]["content"] if messages else ""
        )
        
        for model in [primary_model] + tenant.fallback_chain:
            try:
                start = datetime.utcnow()
                
                response = await self.engine.holysheep_client.post(
                    "/chat/completions",
                    json={
                        "model": model,
                        "messages": messages,
                        "max_tokens": 2048,
                        "temperature": 0.7
                    }
                )
                
                latency = (datetime.utcnow() - start).total_seconds() * 1000
                
                if response.status_code == 200:
                    result = response.json()
                    
                    # 计算实际成本
                    usage = result.get("usage", {})
                    output_tokens = usage.get("completion_tokens", 0)
                    cost = (output_tokens / 1_000_000) * self.engine.model_costs[model]
                    
                    return {
                        "success": True,
                        "model": model,
                        "response": result,
                        "latency_ms": latency,
                        "cost_usd": cost,
                        "fallback_attempted": model != primary_model
                    }
                    
            except Exception as e:
                print(f"Model {model} failed: {e}")
                continue
        
        raise Exception("All models failed")

========== 实际使用示例 ==========
async def demo():
    engine = HolySheepBillingEngine(
        redis_url="redis://localhost:6379",
        holysheep_api_key="YOUR_HOLYSHEEP_API_KEY"  # 替换为真实 Key
    )
    
    router = SmartModelRouter(engine)
    
    # 模拟租户请求
    tenant = await engine.verify_tenant("demo_key")
    
    messages = [
        {"role": "system", "content": "你是一个专业的代码审查助手"},
        {"role": "user", "content": "审查这段 Python 代码的性能问题:\n\ndef fib(n):\n    if n <= 1:\n        return n\n    return fib(n-1) + fib(n-2)"}
    ]
    
    result = await router.execute_with_fallback(tenant, messages)
    
    print(f"模型: {result['model']}")
    print(f"延迟: {result['latency_ms']:.0f}ms")  # HolySheep 国内 < 50ms
    print(f"成本: ${result['cost_usd']:.4f}")
    print(f"降级执行: {result['fallback_attempted']}")

asyncio.run(demo())

class AnomalyTracker:
    """
    异常请求追踪系统
    检测: 刷接口 / 短时间大量失败 / 异常 prompt 模式
    """
    
    def __init__(self, redis: redis.Redis):
        self.redis = redis
    
    async def track_request(
        self, 
        tenant_id: str, 
        endpoint: str,
        tokens_used: int,
        latency_ms: float,
        status: str,
        ip_address: str
    ):
        """记录每次请求到 Redis Stream"""
        event = {
            "tenant_id": tenant_id,
            "endpoint": endpoint,
            "tokens": tokens_used,
            "latency": latency_ms,
            "status": status,
            "ip": ip_address,
            "timestamp": datetime.utcnow().isoformat()
        }
        
        await self.redis.xadd("requests_stream", event)
        
        # 实时统计窗口
        window_key = f"stats:{tenant_id}:{datetime.utcnow().strftime('%Y%m%d%H%M')}"
        pipe = self.redis.pipeline()
        pipe.hincrby(window_key, "requests", 1)
        pipe.hincrbyfloat(window_key, "tokens", tokens_used)
        pipe.hincrbyfloat(window_key, "latency_sum", latency_ms)
        pipe.expire(window_key, 3600)
        await pipe.execute()
    
    async def detect_anomaly(self, tenant_id: str) -> Optional[Dict]:
        """
        异常检测规则:
        1. 60秒内请求 > 500 次
        2. 平均延迟 > 10秒 (可能滥用)
        3. 失败率 > 30%
        4. Token 效率 < 0.3 (输出 tokens / 总 tokens)
        """
        rules = [
            {
                "name": "burst_requests",
                "check": await self._check_burst(tenant_id),
                "threshold": 500,
                "window": 60
            },
            {
                "name": "high_latency",
                "check": await self._check_latency(tenant_id),
                "threshold": 10000,
                "window": 300
            },
            {
                "name": "low_efficiency",
                "check": await self._check_efficiency(tenant_id),
                "threshold": 0.3,
                "window": 3600
            }
        ]
        
        violations = []
        for rule in rules:
            if rule["check"] is not None and rule["check"] > rule["threshold"]:
                violations.append({
                    "type": rule["name"],
                    "value": rule["check"],
                    "threshold": rule["threshold"]
                })
        
        if violations:
            return {
                "tenant_id": tenant_id,
                "violations": violations,
                "action": "warn" if len(violations) == 1 else "ban",
                "timestamp": datetime.utcnow().isoformat()
            }
        
        return None
    
    async def _check_burst(self, tenant_id: str) -> int:
        """检查 60 秒内请求数"""
        keys = []
        for i in range(6):
            t = datetime.utcnow() - timedelta(seconds=i*10)
            keys.append(f"stats:{tenant_id}:{t.strftime('%Y%m%d%H%M')}")
        
        pipe = self.redis.pipeline()
        for key in keys:
            pipe.get(key)
        results = await pipe.execute()
        
        total = sum(int(r.split(':')[1]) if r else 0 for r in results if r)
        return total
    
    async def execute_tracking_pipeline(
        self, 
        api_key: str, 
        request_data: Dict
    ):
        """完整追踪管道"""
        tenant = await self.verify_tenant(api_key)
        
        # 1. 检查是否已被封禁
        ban_key = f"ban:{tenant.tenant_id}"
        if await self.redis.exists(ban_key):
            raise PermissionError(f"Tenant {tenant.tenant_id} is banned")
        
        # 2. 执行追踪
        await self.track_request(
            tenant.tenant_id,
            request_data["endpoint"],
            request_data.get("tokens", 0),
            request_data.get("latency_ms", 0),
            request_data.get("status", "success"),
            request_data.get("ip", "unknown")
        )
        
        # 3. 检测异常
        anomaly = await self.detect_anomaly(tenant.tenant_id)
        if anomaly:
            if anomaly["action"] == "ban":
                await self.redis.setex(ban_key, 86400, json.dumps(anomaly))
                await self.send_alert(tenant.tenant_id, anomaly)
            else:
                await self.send_warning(tenant.tenant_id, anomaly)
        
        return tenant, anomaly
    
    async def send_alert(self, tenant_id: str, anomaly: Dict):
        """发送告警到 Slack/企微"""
        # 实际生产发送真实通知
        print(f"🚨 ALERT: Tenant {tenant_id} banned: {anomaly}")
    
    async def send_warning(self, tenant_id: str, anomaly: Dict):
        """发送警告邮件"""
        print(f"⚠️ WARNING: Tenant {tenant_id}: {anomaly}")

# 问题原因：租户 RPM 超过限制
排查步骤：
1. 检查 Redis: GET ratelimit:{tenant_id}:{minute}
2. 确认租户 tier 配置是否正确
3. 查看 HOLYSHEEP_API_QUOTA 是否耗尽

解决方案：
if not await engine.check_rate_limit(tenant):
    raise httpx.HTTPStatusError(
        "Rate limit exceeded",
        request=None,
        response=httpx.Response(
            429,
            json={"error": "rate_limit_exceeded", "retry_after": 60}
        )
    )

# 问题原因：月度 token 额度用尽
排查步骤：
1. 检查租户 used_tokens vs monthly_token_quota
2. 确认是否月初重置 (reset_at 字段)
3. 查看 billings 表确认无漏计

解决方案：支持额度透支 (企业版)
async def check_quota_with_overdraft(tenant: Tenant, needed: int) -> bool:
    overdraft_limit = {
        TenantTier.FREE: 0,
        TenantTier.BASIC: 10_000,
        TenantTier.PRO: 100_000,
        TenantTier.ENTERPRISE: float('inf')
    }[tenant.tier]
    
    return (tenant.used_tokens + needed) <= (tenant.monthly_token_quota + overdraft_limit)

# 问题原因：租户未购买该模型权限
排查步骤：
1. 打印 tenant.model_permissions 确认权限列表
2. 检查 HolySheep 控制台模型可用性
3. 确认 API Key 对应的套餐

解决方案：自动降级到可用模型
async def safe_model_select(tenant: Tenant, requested: str) -> str:
    if requested in tenant.model_permissions:
        return requested
    
    # 降级到第一个可用模型
    available = tenant.model_permissions[0] if tenant.model_permissions else "gpt-4o-mini"
    logger.warning(f"Model {requested} not available, fallback to {available}")
    return available

# 1. 注册获取 API Key
访问 https://www.holysheep.ai/register

2. 安装 SDK
pip install holyheep-agent-saas

3. 接入代码
from holyheep_agent_saas import AgentSaaS

client = AgentSaaS(
    api_key="YOUR_HOLYSHEEP_API_KEY",  # 替换为真实 Key
    base_url="https://api.holysheep.ai/v1"  # HolySheep 官方端点
)

创建租户
tenant = client.create_tenant(
    tier="pro",
    rate_limit=300,
    quota=1_000_000
)

执行请求
result = client.chat.completions.create(
    tenant_id=tenant.id,
    model="gpt-4.1",
    messages=[{"role": "user", "content": "Hello"}]
)