云函数无缝集成DeepSeek：从技术选型到生产环境实践

1次阅读

共计 3002 个字符，预计需要花费 8 分钟才能阅读完成。

在云函数 (Cloud Function) 场景下集成 AI 服务时，开发者常遇到几个典型问题：

冷启动延迟：云函数冷启动时加载模型的时间可能达到秒级，严重影响响应速度
上下文长度限制：云函数通常有内存和执行时间限制，处理长文本时容易超出限制
计费不可控：AI 服务按请求计费，突发流量可能导致意外费用飙升
连接管理困难：云函数无状态特性使得保持长连接或连接池管理复杂化

接入方式	延迟	成本	扩展性	适用场景
REST API	高(~500ms)	按请求计费	简单	低频调用、快速原型开发
SDK	中(~200ms)	混合计费	中等	常规业务集成
WebAssembly	低(~50ms)	固定成本	复杂	高性能、高频调用场景

from typing import Optional, Dict
import time
import jwt
from datetime import datetime, timedelta

class DeepSeekClient:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self._token_cache: Optional[str] = None
        self._token_expiry: Optional[datetime] = None

    def _generate_jwt(self) -> str:
        """生成带有效期的 JWT Token"""
        now = datetime.utcnow()
        payload = {
            'iss': 'deepseek-client',
            'exp': now + timedelta(minutes=30),
            'iat': now
        }
        return jwt.encode(payload, self.api_key, algorithm='HS256')

    def get_token(self) -> str:
        """获取有效 Token，自动刷新过期 Token"""
        if self._token_cache and self._token_expiry > datetime.utcnow():
            return self._token_cache

        new_token = self._generate_jwt()
        self._token_cache = new_token
        self._token_expiry = datetime.utcnow() + timedelta(minutes=25)  # 提前 5 分钟刷新
        return new_token

import asyncio
from typing import List

async def batch_inference(
    client: DeepSeekClient,
    inputs: List[str],
    batch_size: int = 5
) -> List[str]:
    """
    批处理推理请求
    :param client: DeepSeek 客户端实例
    :param inputs: 输入文本列表
    :param batch_size: 每批数量
    :return: 结果列表
    """
    results = []

    for i in range(0, len(inputs), batch_size):
        batch = inputs[i:i + batch_size]
        tasks = [_single_inference(client, text)
            for text in batch
        ]
        batch_results = await asyncio.gather(*tasks)
        results.extend(batch_results)

    return results

async def _single_inference(client: DeepSeekClient, text: str) -> str:
    """单个推理请求实现"""
    token = client.get_token()
    # 实际请求逻辑...
    return f"processed: {text}"

预热函数：定期调用保持实例活跃
最小化依赖：减少冷启动时的包加载时间
保持连接：复用 HTTP 连接而非每次新建

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["deepseek:Inference"],
      "Resource": "*"
    }
  ]
}

from collections import deque
import time

class RateLimiter:
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.request_times = deque()

    def check_limit(self) -> bool:
        now = time.time()
        # 移除窗口外的记录
        while self.request_times and self.request_times[0] <= now - self.window_seconds:
            self.request_times.popleft()

        if len(self.request_times) >= self.max_requests:
            return False

        self.request_times.append(now)
        return True

import random

async def call_with_retry(func, max_retries=3):
    """指数退避重试机制"""
    for attempt in range(max_retries):
        try:
            return await func()
        except Exception as e:
            if "429" not in str(e):
                raise

            wait_time = (2 ** attempt) + random.uniform(0, 1)
            await asyncio.sleep(wait_time)

    raise Exception("Max retries exceeded")

def chunk_text(text: str, max_length: int) -> List[str]:
    """按最大长度分块文本"""
    if len(text) <= max_length:
        return [text]

    chunks = []
    words = text.split()
    current_chunk = []
    current_length = 0

    for word in words:
        if current_length + len(word) + 1 > max_length:
            chunks.append(' '.join(current_chunk))
            current_chunk = []
            current_length = 0

        current_chunk.append(word)
        current_length += len(word) + 1

    if current_chunk:
        chunks.append(' '.join(current_chunk))

    return chunks

在云函数集成 AI 模型时，如何平衡模型精度与函数执行时间限制？可以考虑：