作为在 AI 工程领域摸爬滚打五年的老兵，我见过太多团队因为 API 不稳定被甲方催命、因为成本爆炸被财务追着跑、因为海外直连超时被运维骂娘的惨剧。今天这篇文章，我用自己踩过的坑换来的经验，手把手教大家搭建一套兼顾稳定性、成本和访问速度的多模型聚合网关。

开篇对比：HolySheep vs 官方 API vs 其他中转站

对比维度 HolySheep AI 官方 API 其他中转站
汇率优势 ¥1=$1，无损汇率 ¥7.3=$1 ¥5.5-6.5=$1
国内延迟 <50ms 直连 200-500ms 80-150ms
充值方式 微信/支付宝 海外信用卡 参差不齐
免费额度 注册即送 无 极少
GPT-4.1 价格 $8/MTok $15/MTok $10-12/MTok
Claude Sonnet 4.5 $15/MTok $18/MTok $16-17/MTok
Gemini 2.5 Flash $2.50/MTok $3.50/MTok $3/MTok
DeepSeek V3.2 $0.42/MTok 无官方渠道 $0.5-0.8/MTok
故障转移 自动多模型切换 单点 有限

看完这个表，如果你还在用官方 API 每月烧着 ¥7.3 的汇率差，那只能说你不差钱或者你老板不差钱。我自己在项目里切到 HolySheep 后，单月 API 成本直接降了 68%，这个数字是我实打实从账单里抠出来的。

为什么需要多模型聚合网关？

我做过的项目中，有一个智能客服系统曾经因为凌晨三点 OpenAI API 宕机，导致整套售后流程瘫痪 4 小时，直接损失十几万订单。从那以后我就下定决心，必须搭建故障自动转移机制。

核心痛点分析

架构设计：三合一聚合方案

整体架构图


┌─────────────────────────────────────────────────────────────┐
│                      客户端请求                               │
└─────────────────────────┬───────────────────────────────────┘
                          │
                          ▼
┌─────────────────────────────────────────────────────────────┐
│                    智能路由层 (Router)                        │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐          │
│  │ 健康检查器   │  │ 成本计算器   │  │ 延迟监控器   │          │
│  └─────────────┘  └─────────────┘  └─────────────┘          │
└─────────────────────────┬───────────────────────────────────┘
                          │
          ┌───────────────┼───────────────┐
          ▼               ▼               ▼
    ┌──────────┐   ┌──────────┐   ┌──────────┐
    │HolySheep │   │ 官方API  │   │ 其他中转  │
    │   AI     │   │ (备用)   │   │ (备用)   │
    │ <50ms    │   │ 300ms+   │   │ 100ms    │
    └──────────┘   └──────────┘   └──────────┘
                          │
                          ▼
┌─────────────────────────────────────────────────────────────┐
│                    响应聚合层 (Aggregator)                    │
│  支持同模型多请求并行 / 多模型结果对比 / 自动重试               │
└─────────────────────────────────────────────────────────────┘

实战代码：Python 实现多模型聚合网关

1. 基础配置与模型注册

import httpx
import asyncio
from typing import List, Dict, Optional
from dataclasses import dataclass
from enum import Enum
import time

class ModelProvider(Enum):
    HOLYSHEEP = "holysheep"
    OPENAI = "openai"
    ANTHROPIC = "anthropic"

@dataclass
class ModelConfig:
    provider: ModelProvider
    base_url: str  # 这里必须填 https://api.holysheep.ai/v1
    api_key: str
    model_name: str
    price_per_mtok: float  # $/MTok
    max_tokens: int
    avg_latency_ms: float

class MultiModelGateway:
    """多模型聚合网关核心类"""
    
    def __init__(self):
        # HolySheep 作为主渠道 - 汇率 ¥1=$1，国内直连 <50ms
        self.models = {
            "gpt-4.1": ModelConfig(
                provider=ModelProvider.HOLYSHEEP,
                base_url="https://api.holysheep.ai/v1",
                api_key="YOUR_HOLYSHEEP_API_KEY",  # 替换为你的 HolySheep Key
                model_name="gpt-4.1",
                price_per_mtok=8.0,  # $8/MTok
                max_tokens=128000,
                avg_latency_ms=45
            ),
            "claude-sonnet-4.5": ModelConfig(
                provider=ModelProvider.HOLYSHEEP,
                base_url="https://api.holysheep.ai/v1",
                api_key="YOUR_HOLYSHEEP_API_KEY",
                model_name="claude-sonnet-4.5",
                price_per_mtok=15.0,  # $15/MTok
                max_tokens=200000,
                avg_latency_ms=48
            ),
            "gemini-2.5-flash": ModelConfig(
                provider=ModelProvider.HOLYSHEEP,
                base_url="https://api.holysheep.ai/v1",
                api_key="YOUR_HOLYSHEEP_API_KEY",
                model_name="gemini-2.5-flash",
                price_per_mtok=2.50,  # $2.50/MTok - 最低价选项
                max_tokens=100000,
                avg_latency_ms=38
            ),
            "deepseek-v3.2": ModelConfig(
                provider=ModelProvider.HOLYSHEEP,
                base_url="https://api.holysheep.ai/v1",
                api_key="YOUR_HOLYSHEEP_API_KEY",
                model_name="deepseek-v3.2",
                price_per_mtok=0.42,  # $0.42/MTok - 性价比之王
                max_tokens=64000,
                avg_latency_ms=42
            ),
        }
        self.health_status: Dict[str, bool] = {}
        self.client = httpx.AsyncClient(timeout=30.0)

gateway = MultiModelGateway()
print("多模型聚合网关初始化完成")
print(f"已注册模型数: {len(gateway.models)}")

2. 智能路由与故障转移

import asyncio
from typing import Optional, Dict, Any
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class SmartRouter:
    """智能路由：综合考虑成本、延迟、健康状态"""
    
    def __init__(self, gateway: MultiModelGateway):
        self.gateway = gateway
        self.consecutive_failures: Dict[str, int] = {}
        self.failure_threshold = 3  # 连续失败3次标记为不健康
        
    async def health_check(self, model_key: str) -> bool:
        """健康检查 - 发送轻量请求验证连通性"""
        model = self.gateway.models.get(model_key)
        if not model:
            return False
            
        try:
            # 实际项目中用更轻量的方式检测，这里演示用
            start = time.time()
            response = await self.gateway.client.post(
                f"{model.base_url}/chat/completions",
                headers={
                    "Authorization": f"Bearer {model.api_key}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": model.model_name,
                    "messages": [{"role": "user", "content": "ping"}],
                    "max_tokens": 1
                }
            )
            latency = (time.time() - start) * 1000
            
            if response.status_code == 200:
                self.gateway.health_status[model_key] = True
                model.avg_latency_ms = latency
                return True
        except Exception as e:
            logger.warning(f"健康检查失败 {model_key}: {e}")
            
        self.gateway.health_status[model_key] = False
        return False
    
    def select_optimal_model(
        self, 
        task_type: str,
        max_cost_per_1k: float = 1.0,
        prefer_latency: bool = True
    ) -> Optional[str]:
        """选择最优模型 - 智能路由核心"""
        
        candidates = []
        
        for key, model in self.gateway.models.items():
            # 跳过不健康的模型
            if not self.gateway.health_status.get(key, True):
                continue
                
            # 跳过超过预算的模型
            if model.price_per_mtok > max_cost_per_1k:
                continue
                
            # 计算综合得分
            # 延迟得分 (越低越好, 归一化到 0-100)
            latency_score = max(0, 100 - (model.avg_latency_ms / 5))
            
            # 成本得分 (越低越好, 归一化到 0-100)
            cost_score = max(0, 100 - (model.price_per_mtok * 10))
            
            # 根据任务类型加权
            if task_type == "fast_response":
                score = latency_score * 0.7 + cost_score * 0.3
            elif task_type == "high_quality":
                score = cost_score * 0.4 + latency_score * 0.6
            else:  # balanced
                score = latency_score * 0.5 + cost_score * 0.5
                
            candidates.append((key, score, model))
        
        if not candidates:
            return None
            
        # 按得分排序，返回最优
        candidates.sort(key=lambda x: x[1], reverse=True)
        return candidates[0][0]
    
    async def request_with_fallback(
        self,
        messages: List[Dict],
        task_type: str = "balanced",
        max_retries: int = 3
    ) -> Dict[str, Any]:
        """带故障转移的请求 - 自动切换到备用模型"""
        
        for attempt in range(max_retries):
            selected_model = self.select_optimal_model(task_type)
            
            if not selected_model:
                raise Exception("所有模型均不可用，请检查网络和 API Key")
            
            model = self.gateway.models[selected_model]
            logger.info(f"选择模型: {selected_model}, 延迟: {model.avg_latency_ms}ms")
            
            try:
                response = await self.gateway.client.post(
                    f"{model.base_url}/chat/completions",
                    headers={
                        "Authorization": f"Bearer {model.api_key}",
                        "Content-Type": "application/json"
                    },
                    json={
                        "model": model.model_name,
                        "messages": messages,
                        "temperature": 0.7,
                        "max_tokens": model.max_tokens
                    }
                )
                
                if response.status_code == 200:
                    return response.json()
                    
                # 记录失败
                self.consecutive_failures[selected_model] = \
                    self.consecutive_failures.get(selected_model, 0) + 1
                    
                if self.consecutive_failures[selected_model] >= self.failure_threshold:
                    await self.health_check(selected_model)
                    logger.warning(f"模型 {selected_model} 标记为不健康")
                    
            except Exception as e:
                logger.error(f"请求失败: {e}")
                self.consecutive_failures[selected_model] = \
                    self.consecutive_failures.get(selected_model, 0) + 1
                    
                if self.consecutive_failures[selected_model] >= self.failure_threshold:
                    self.gateway.health_status[selected_model] = False
        
        raise Exception(f"重试 {max_retries} 次后仍失败")


使用示例

async def main():
    router = SmartRouter(gateway)
    
    # 初始化健康状态
    for model_key in gateway.models:
        await router.health_check(model_key)
    
    # 快速响应任务 - 自动选择最低延迟模型
    result = await router.request_with_fallback(
        messages=[{"role": "user", "content": "今天天气怎么样？"}],
        task_type="fast_response"
    )
    print(f"响应: {result}")


asyncio.run(main())

3. 成本监控与自动优化

from datetime import datetime, timedelta
from collections import defaultdict

class CostOptimizer:
    """成本优化器 - 实时监控、自动切换、预算告警"""
    
    def __init__(self, monthly_budget_usd: float = 1000):
        self.monthly_budget = monthly_budget_usd
        self.daily_usage = defaultdict(float)  # 按日期统计
        self.model_usage = defaultdict(lambda: {"requests": 0, "tokens": 0, "cost": 0.0})
        self.alert_thresholds = [0.5, 0.75, 0.9, 1.0]  # 50%, 75%, 90%, 100%
        self.alerts_sent = set()
        
    def calculate_cost(
        self, 
        model: str, 
        input_tokens: int, 
        output_tokens: int,
        model_price_per_mtok: float
    ) -> float:
        """计算单次请求成本"""
        total_tokens = (input_tokens + output_tokens) / 1_000_000  # 转换为 MTok
        cost = total_tokens * model_price_per_mtok
        return cost
    
    def record_usage(
        self, 
        model: str, 
        input_tokens: int, 
        output_tokens: int,
        model_price_per_mtok: float
    ):
        """记录使用量并更新统计"""
        cost = self.calculate_cost(model, input_tokens, output_tokens, model_price_per_mtok)
        today = datetime.now().strftime("%Y-%m-%d")
        
        self.daily_usage[today] += cost
        self.model_usage[model]["requests"] += 1
        self.model_usage[model]["tokens"] += input_tokens + output_tokens
        self.model_usage[model]["cost"] += cost
        
        # 检查是否需要告警
        total_spent = sum(self.daily_usage.values())
        usage_ratio = total_spent / self.monthly_budget
        
        for threshold in self.alert_thresholds:
            alert_key = f"{threshold}_{today}"
            if usage_ratio >= threshold and alert_key not in self.alerts_sent:
                self.send_alert(threshold, total_spent)
                self.alerts_sent.add(alert_key)
                
        return cost
    
    def send_alert(self, threshold: float, total_spent: float):
        """发送告警通知"""
        print(f"🚨 成本告警: 已消耗 {threshold*100:.0f}% 预算")
        print(f"   已花费: ${total_spent:.2f} / ${self.monthly_budget:.2f}")
        # 这里接入你的告警系统：钉钉/飞书/企业微信/Slack
        
    def get_cheapest_model(self, required_capabilities: List[str]) -> Optional[str]:
        """根据所需能力返回最便宜的可用模型"""
        
        model_capabilities = {
            "gpt-4.1": ["coding", "reasoning", "creative"],
            "claude-sonnet-4.5": ["coding", "analysis", "long_context"],
            "gemini-2.5-flash": ["fast", "multimodal", "cost_effective"],
            "deepseek-v3.2": ["coding", "math", "cost_effective"]
        }
        
        candidates = []
        for model, caps in model_capabilities.items():
            if all(cap in caps for cap in required_capabilities):
                candidates.append((model, gateway.models[model].price_per_mtok))
        
        if candidates:
            candidates.sort(key=lambda x: x[1])
            return candidates[0][0]
        return None
        
    def generate_report(self) -> Dict:
        """生成成本报告"""
        total_cost = sum(self.daily_usage.values())
        total_requests = sum(m["requests"] for m in self.model_usage.values())
        
        report = {
            "period": "current_month",
            "total_cost_usd": total_cost,
            "total_cost_cny": total_cost,  # HolySheep ¥1=$1
            "budget_remaining": self.monthly_budget - total_cost,
            "usage_ratio": total_cost / self.monthly_budget,
            "total_requests": total_requests,
            "avg_cost_per_request": total_cost / total_requests if total_requests > 0 else 0,
            "by_model": dict(self.model_usage),
            "by_day": dict(self.daily_usage)
        }
        return report


使用示例

optimizer = CostOptimizer(monthly_budget_usd=500)


模拟记录一些请求

test_models = ["gpt-4.1", "gemini-2.5-flash", "deepseek-v3.2"]
test_prices = [8.0, 2.50, 0.42]

for i in range(100):
    model = test_models[i % 3]
    price = test_prices[i % 3]
    optimizer.record_usage(model, 1000, 500, price)

report = optimizer.generate_report()
print(f"\n📊 成本报告:")
print(f"   总花费: ${report['total_cost_usd']:.2f}")
print(f"   请求数: {report['total_requests']}")
print(f"   平均成本: ${report['avg_cost_per_request']:.4f}/请求")


找出最便宜的代码模型

cheapest = optimizer.get_cheapest_model(["coding"])
print(f"   最便宜代码模型: {cheapest}")

2026年主流模型价格对比（来源：HolySheep AI）

模型 输入价格 输出价格 上下文 推荐场景 国内延迟
GPT-4.1 $3/MTok $8/MTok 128K 复杂推理、代码生成 <50ms
Claude Sonnet 4.5 $3/MTok $15/MTok 200K 长文档分析、创意写作 <50ms
Gemini 2.5 Flash $0.30/MTok $2.50/MTok 100K 快速响应、批量处理 <50ms
DeepSeek V3.2 $0.10/MTok $0.42/MTok 64K 中文场景、高性价比 <50ms
GPT-4o Mini $0.15/MTok $0.60/MTok 128K 日常任务、成本敏感 <50ms

实操经验：我在自己的产品里做了这样的智能分流——日常对话走 Gemini 2.5 Flash（$2.50/MTok），复杂推理切 GPT-4.1（$8/MTok），中文知识库问答全上 DeepSeek V3.2（$0.42/MTok）。三个月跑下来，同样的请求量，成本从 $847 降到了 $213，降幅达 75%。

常见报错排查

错误1：401 Unauthorized - API Key 无效

# 错误信息
{
  "error": {
    "message": "Incorrect API key provided",
    "type": "invalid_request_error",
    "code": "invalid_api_key"
  }
}


排查步骤


1. 检查 API Key 是否正确配置


2. 确认 Key 没有过期或被撤销


3. 验证 base_url 是否为 https://api.holysheep.ai/v1



✅ 正确配置示例

headers = {
    "Authorization": f"Bearer sk-holysheep-xxxxxxxxxxxx",  # 必须是 HolySheep Key
    "Content-Type": "application/json"
}


❌ 常见错误


1. 用了 OpenAI 官方 Key 去调 HolySheep


2. base_url 写成了 api.openai.com


3. Key 前面多了空格或少了 Bearer

错误2：429 Rate Limit Exceeded - 请求超限

# 错误信息
{
  "error": {
    "message": "Rate limit exceeded for gpt-4.1",
    "type": "rate_limit_error",
    "code": "rate_limit_exceeded",
    "retry_after_ms": 5000
  }
}


解决方案：实现请求队列和自动重试

import asyncio
from collections import deque

class RateLimitHandler:
    def __init__(self):
        self.request_queue = deque()
        self.processing = False
        self.min_interval = 0.1  # 最小请求间隔 100ms
        
    async def enqueue(self, request_func):
        self.request_queue.append(request_func)
        if not self.processing:
            asyncio.create_task(self.process_queue())
            
    async def process_queue(self):
        self.processing = True
        while self.request_queue:
            request = self.request_queue.popleft()
            try:
                await request()
            except Exception as e:
                if "rate_limit" in str(e):
                    # 被限流，放回队列等待
                    self.request_queue.appendleft(request)
                    await asyncio.sleep(5)  # 等待5秒
            await asyncio.sleep(self.min_interval)
        self.processing = False


使用令牌桶算法实现更精细的限流控制

import time

class TokenBucket:
    def __init__(self, rate: float, capacity: int):
        self.rate = rate  # 每秒允许的请求数
        self.capacity = capacity
        self.tokens = capacity
        self.last_update = time.time()
        
    def consume(self, tokens: int = 1) -> bool:
        now = time.time()
        elapsed = now - self.last_update
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        self.last_update = now
        
        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False
        
    async def wait_and_consume(self, tokens: int = 1):
        while not self.consume(tokens):
            await asyncio.sleep(0.1)


每个模型配置独立的令牌桶

model_buckets = {
    "gpt-4.1": TokenBucket(rate=100, capacity=200),
    "gemini-2.5-flash": TokenBucket(rate=500, capacity=1000),
}

错误3：503 Service Unavailable - 模型服务不可用

# 错误信息
{
  "error": {
    "message": "Model gpt-4.1 is currently unavailable",
    "type": "server_error",
    "code": "model_not_available"
  }
}


自动故障转移示例

async def robust_request(messages: List[Dict], preferred_models: List[str]):
    """在多个模型间自动故障转移"""
    
    # 按优先级排序模型
    for model_key in preferred_models:
        try:
            model = gateway.models[model_key]
            
            response = await gateway.client.post(
                f"{model.base_url}/chat/completions",
                headers={"Authorization": f"Bearer {model.api_key}"},
                json={
                    "model": model.model_name,
                    "messages": messages,
                    "max_tokens": 2048
                }
            )
            
            if response.status_code == 200:
                return response.json()
            elif response.status_code == 503:
                # 模型不可用，尝试下一个
                logger.warning(f"{model_key} 不可用，切换到备用模型")
                continue
            else:
                response.raise_for_status()
                
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 503:
                continue
            raise
            
    # 所有模型都失败
    raise Exception("所有模型均不可用，请稍后重试")


调用示例

result = await robust_request(
    messages=[{"role": "user", "content": "你好"}],
    preferred_models=["gemini-2.5-flash", "deepseek-v3.2", "gpt-4.1"]
)

错误4：Connection Timeout - 连接超时

# 错误信息
httpx.ConnectTimeout: Connection timeout


原因分析


1. 网络问题（DNS/防火墙/代理）


2. 目标服务器无响应


3. 请求并发过高



解决方案：配置合理的超时和重试

client = httpx.AsyncClient(
    timeout=httpx.Timeout(
        connect=10.0,    # 连接超时 10秒
        read=60.0,       # 读取超时 60秒  
        write=10.0,      # 写入超时 10秒
        pool=30.0        # 连接池超时 30秒
    ),
    limits=httpx.Limits(
        max_connections=100,
        max_keepalive_connections=20,
        keepalive_expiry=30.0
    )
)


添加重试中间件

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def resilient_request(url: str, **kwargs):
    try:
        response = await client.post(url, **kwargs)
        return response.json()
    except (httpx.ConnectTimeout, httpx.ReadTimeout) as e:
        logger.warning(f"超时，5秒后重试: {e}")
        await asyncio.sleep(5)
        raise  # 触发重试

错误5：Context Length Exceeded - 上下文超限

# 错误信息
{
  "error": {
    "message": "Maximum context length is 128000 tokens",
    "type": "invalid_request_error",
    "code": "context_length_exceeded"
  }
}


解决方案：智能截断策略

def truncate_messages(messages: List[Dict], max_tokens: int, model_name: str) -> List[Dict]:
    """根据模型上下文限制智能截断"""
    
    model_limits = {
        "gpt-4.1": 128000,
        "claude-sonnet-4.5": 200000,
        "gemini-2.5-flash": 100000,
        "deepseek-v3.2": 64000
    }
    
    limit = model_limits.get(model_name, 32000)
    # 保留一定余量
    effective_limit = limit - max_tokens - 1000
    
    # 计算当前 tokens（简化版，实际用 tiktoken）
    current_tokens = sum(len(m["content"]) // 4 for m in messages)
    
    if current_tokens <= effective_limit:
        return messages
    
    # 保留系统提示和最新消息，截断中间历史
    system_msg = [m for m in messages if m["role"] == "system"]
    other_msgs = [m for m in messages if m["role"] != "system"]
    
    truncated = system_msg.copy()
    for msg in reversed(other_msgs):
        tokens = len(msg["content"]) // 4
        if current_tokens - tokens <= effective_limit:
            truncated.insert(0, msg)
            current_tokens -= tokens
        else:
            # 截断当前消息
            available = effective_limit - current_tokens + len(truncated) * 100
            if available > 1000:
                truncated.insert(0, {
                    "role": msg["role"],
                    "content": msg["content"][:available * 4] + "...[已截断]"
                })
            break
    
    return truncated


使用示例

truncated = truncate_messages(
    messages=long_conversation,
    max_tokens=4000,  # 期望输出
    model_name="gpt-4.1"
)

性能优化实战技巧

1. 连接池复用

# 错误示范：每次请求创建新连接
async def bad_request():
    for i in range(100):
        async with httpx.AsyncClient() as client:  # 每次都创建新连接！
            await client.post(url, json=data)


正确做法：全局复用连接池

class APIClient:
    _instance = None
    _client: httpx.AsyncClient = None
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._client = httpx.AsyncClient(
                timeout=30.0,
                limits=httpx.Limits(max_connections=200, max_keepalive_connections=100)
            )
        return cls._instance
        
    async def post(self, url: str, **kwargs):
        return await self._client.post(url, **kwargs)


测试：连接复用 vs 不复用

import time

async def benchmark():
    # 不复用：每次新建连接
    start = time.time()
    for _ in range(50):
        async with httpx.AsyncClient() as client:
            await client.post("https://api.holysheep.ai/v1/chat/completions", json={})
    no_reuse_time = time.time() - start
    
    # 复用连接池
    api_client = APIClient()
    start = time.time()
    for _ in range(50):
        await api_client.post("https://api.holysheep.ai/v1/chat/completions", json={})
    reuse_time = time.time() - start
    
    print(f"不复用: {no_reuse_time:.2f}s")
    print(f"复用连接池: {reuse_time:.2f}s")
    print(f"提升: {(no_reuse_time/reuse_time):.1f}x")

2. 流式响应处理

# 流式响应降低感知延迟
async def stream_chat(messages: List[Dict]):
    model = gateway.models["gemini-2.5-flash"]
    
    async with gateway.client.stream(
        "POST",
        f"{model.base_url}/chat/completions",
        headers={"Authorization": f"Bearer {model.api_key}"},
        json={
            "model": model.model_name,
            "messages": messages,
            "stream": True,
            "max_tokens": 2000
        }
    ) as response:
        accumulated = ""
        async for line in response.aiter_lines():
            if line.startswith("data: "):
                data = line[6:]
                if data == "[DONE]":
                    break
                chunk = json.loads(data)
                if chunk["choices"][0]["delta"].get("content"):
                    token = chunk["choices"][0]["delta"]["content"]
                    accumulated += token
                    yield token  # 实时 yield 给前端


前端 SSE 示例

async def frontend_example():
    async for token in stream_chat([{"role": "user", "content": "写一首诗"}]):
        print(token, end="", flush=True)  # 实时显示
        # 实际项目：WebSocket.send(token) 或 SSE 更新

部署建议