聊聊我在杭州搞AI应用性能优化踩过的坑

说实话，写这篇博客的时候我刚从工位上站起来，泡了杯咖啡，窗外杭州的夜色已经挺深了。作为一个在杭州混了好几年的DevOps工程师，日常就是跟各种CI/CD流水线、容器编排、监控告警打交道。用MacBook Pro写脚本，Windows虚拟机里跑测试，这基本就是我的日常。

本来想着今年安安稳稳混到年底，结果去年Q4的时候，我们团队接了个挺有意思的项目，硬生生把我从纯运维的舒适区拽出来，搞了一波AI应用的性能优化。今天就来聊聊这段经历，算是个技术探索与实践总结吧。

事情是这样的

去年10月份，产品那边突然拉了个会，说要搞一个AI内容生成的平台。我当时心里就咯噔一下——产品一兴奋，运维就得脱层皮，这是铁律。

需求大概是这样的：要接入大模型做文本生成，同时支持图片生成，还得有语义搜索的能力。技术选型上，文本生成用的是腾讯混元，图片生成选了Pika，语义搜索这块需要用到Embedding模型。

"这有啥难的，不就是调几个API嘛。"当时我们组有个年轻的后端兄弟说了这么一句。我当时就笑了，心想这小子还是太年轻。

果不其然，第一版上线的时候，问题就来了。

第一波暴击：延迟高到离谱

测试环境跑起来之后，产品经理自己试了一下，直接在工作群里@我："这个响应速度也太慢了吧，等半天才出结果，用户早跑了。"

我一看监控，好家伙，文本生成接口平均响应时间4.8秒，图片生成更夸张，Pika那边平均要12秒。这还没算上Embedding做语义检索的时间。

说实话，当时看到这些数据我头都大了。我们SLA要求是P99延迟控制在3秒以内，这差得也太多了。

先来分析一下问题出在哪。我拉了链路追踪的数据一看，主要瓶颈在这几个地方：

接口	平均耗时	P99耗时	主要瓶颈
腾讯混元文本生成	4.8s	8.2s	模型推理 + 网络IO
Pika图片生成	12.3s	18.5s	模型推理 + 异步回调
Embedding语义检索	1.2s	2.8s	向量计算 + 数据库查询

问题很明确了。腾讯混元和Pika都是第三方服务，模型推理时间我们控制不了，但网络IO、请求调度、结果缓存这些是我们能优化的。Embedding这块倒是可以好好搞一搞。

先拿Embedding开刀

Embedding这块相对来说是最可控的，毕竟模型可以自部署。我们用的是BGE-large-zh模型，一开始是直接跑的CPU版本，那速度确实感人。

第一步，上GPU。这个没啥好说的，把模型迁移到A10上，推理速度直接提升了6倍。

# 优化前的Embedding服务
from sentence_transformers import SentenceTransformer

class EmbeddingService:
    def __init__(self):
        # CPU版本，慢得一批
        self.model = SentenceTransformer('BAAI/bge-large-zh')
    
    def encode(self, texts: list) -> list:
        # 单条处理，没有batch
        embeddings = []
        for text in texts:
            emb = self.model.encode(text)
            embeddings.append(emb.tolist())
        return embeddings

当时我看到这段代码的时候，真的想砸电脑。for循环逐条处理？这兄弟是认真的吗？

# 优化后的Embedding服务
import torch
from sentence_transformers import SentenceTransformer
from concurrent.futures import ThreadPoolExecutor

class EmbeddingServiceOptimized:
    def __init__(self):
        # GPU版本，指定设备
        device = 'cuda' if torch.cuda.is_available() else 'cpu'
        self.model = SentenceTransformer(
            'BAAI/bge-large-zh',
            device=device,
            # 开启half精度，显存占用减半，速度还能再快点
        ).half()
        # 批处理大小，根据显存调整
        self.batch_size = 64
    
    def encode(self, texts: list) -> list:
        # batch处理，这才是正确姿势
        embeddings = self.model.encode(
            texts,
            batch_size=self.batch_size,
            convert_to_tensor=True,
            show_progress_bar=False
        )
        return embeddings.cpu().numpy().tolist()
    
    def encode_with_cache(self, texts: list) -> list:
        # 加上Redis缓存，重复查询直接命中
        results = []
        miss_texts = []
        miss_indices = []
        
        for i, text in enumerate(texts):
            cache_key = f"emb:{hash(text)}"
            cached = self.redis_client.get(cache_key)
            if cached:
                results.append((i, json.loads(cached)))
            else:
                miss_texts.append(text)
                miss_indices.append(i)
        
        if miss_texts:
            miss_embeddings = self.encode(miss_texts)
            for idx, emb in zip(miss_indices, miss_embeddings):
                results.append((idx, emb))
                # 缓存24小时
                cache_key = f"emb:{hash(texts[idx])}"
                self.redis_client.setex(
                    cache_key, 86400, json.dumps(emb)
                )
        
        # 按原始顺序返回
        results.sort(key=lambda x: x[0])
        return [r[1] for r in results]

这一波优化下来，Embedding接口的平均耗时从1.2s降到了0.18s，P99也从2.8s降到了0.35s。效果还是很明显的。

不过这里有个坑要提醒大家：batch_size不是越大越好。我一开始贪心设了256，结果显存直接OOM了。后来压测下来64是个比较平衡的值，既不会爆显存，吞吐量也够。

腾讯混元这边怎么搞

腾讯混元是第三方API，推理时间我们没法优化，但调用方式可以优化。

原来的代码是这样的：

# 原来的调用方式，同步阻塞
def generate_text(prompt: str) -> str:
    response = hunyuan_client.chat(
        model="hunyuan-pro",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

同步阻塞调用，一个请求占一个线程，并发一上来线程池就满了。

我改成了异步+流式的方式：

import asyncio
import aiohttp
from typing import AsyncGenerator

class HunyuanService:
    def __init__(self):
        self.api_key = settings.HUNYUAN_API_KEY
        self.base_url = "https://api.hunyuan.cloud.tencent.com/v1"
        # 连接池，复用TCP连接
        self.connector = aiohttp.TCPConnector(
            limit=100,           # 连接池大小
            limit_per_host=20,   # 单host连接数
            ttl_dns_cache=300,   # DNS缓存
            keepalive_timeout=30
        )
        self.session = aiohttp.ClientSession(connector=self.connector)
    
    async def generate_text_stream(
        self, 
        prompt: str,
        max_tokens: int = 2048
    ) -> AsyncGenerator[str, None]:
        """流式输出，用户体验更好"""
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": "hunyuan-pro",
            "messages": [{"role": "user", "content": prompt}],
            "max_tokens": max_tokens,
            "stream": True,
            # 温度参数，控制创造性
            "temperature": 0.7,
            # top_p采样
            "top_p": 0.9
        }
        
        async with self.session.post(
            f"{self.base_url}/chat/completions",
            headers=headers,
            json=payload,
            timeout=aiohttp.ClientTimeout(total=30)
        ) as response:
            async for line in response.content:
                if line:
                    decoded = line.decode('utf-8')
                    if decoded.startswith('data: '):
                        data = decoded[6:]
                        if data != '[DONE]':
                            chunk = json.loads(data)
                            content = chunk['choices'][0]['delta'].get('content', '')
                            if content:
                                yield content
    
    async def generate_text_batch(
        self, 
        prompts: list,
        concurrency: int = 10
    ) -> list:
        """批量请求，控制并发"""
        semaphore = asyncio.Semaphore(concurrency)
        
        async def _generate(prompt):
            async with semaphore:
                result = ""
                async for chunk in self.generate_text_stream(prompt):
                    result += chunk
                return result
        
        tasks = [_generate(p) for p in prompts]
        return await asyncio.gather(*tasks)

流式输出的好处是用户不用干等，能看到内容一点点出来，体感上会快很多。再加上连接池复用和异步并发，整体吞吐量提升了大概4倍。

这里还有个细节：腾讯混元的API有QPS限制，我们申请到的额度是20 QPS。所以批量请求的时候一定要控制并发，不然会被限流。我用Semaphore来控制，设了10的并发度，留了一半的余量。

Pika图片生成，这个才是大头

Pika的图片生成是真的慢，平均12秒，有时候能到20秒。而且这玩意儿不支持流式输出，只能干等。

原来的方案是同步等结果返回，这在高并发场景下简直是灾难。

我改成了异步任务+回调的模式：

import uuid
import asyncio
from enum import Enum

class TaskStatus(Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    COMPLETED = "completed"
    FAILED = "failed"

class PikaService:
    def __init__(self):
        self.redis_client = redis.Redis(
            host='redis.internal', 
            port=6379,
            decode_responses=True
        )
        self.task_timeout = 120  # 任务超时2分钟
    
    async def submit_generation(
        self, 
        prompt: str,
        negative_prompt: str = "",
        width: int = 512,
        height: int = 512
    ) -> str:
        """提交生成任务，返回task_id"""
        task_id = str(uuid.uuid4())
        
        # 任务状态写入Redis
        task_info = {
            "status": TaskStatus.PENDING.value,
            "prompt": prompt,
            "created_at": time.time(),
            "result_url": None
        }
        self.redis_client.hset(
            f"pika:task:{task_id}", 
            mapping=task_info
        )
        # 设置过期时间，防止僵尸任务
        self.redis_client.expire(
            f"pika:task:{task_id}", 
            self.task_timeout + 60
        )
        
        # 异步提交到Pika
        asyncio.create_task(
            self._call_pika_api(task_id, prompt, negative_prompt, width, height)
        )
        
        return task_id
    
    async def _call_pika_api(
        self, 
        task_id: str,
        prompt: str,
        negative_prompt: str,
        width: int,
        height: int
    ):
        """实际调用Pika API"""
        try:
            # 更新状态为处理中
            self.redis_client.hset(
                f"pika:task:{task_id}",
                "status", TaskStatus.PROCESSING.value
            )
            
            # 调用Pika API
            async with aiohttp.ClientSession() as session:
                payload = {
                    "prompt": prompt,
                    "negative_prompt": negative_prompt,
                    "width": width,
                    "height": height,
                    "callback_url": f"https://api.our-service.com/pika/callback/{task_id}"
                }
                
                async with session.post(
                    "https://api.pika.art/v1/generate",
                    json=payload,
                    timeout=aiohttp.ClientTimeout(total=10)
                ) as resp:
                    if resp.status != 200:
                        raise Exception(f"Pika API error: {resp.status}")
            
            # 启动轮询检查任务状态
            # 因为Pika的回调有时候不太靠谱，加个兜底
            asyncio.create_task(
                self._poll_task_status(task_id)
            )
            
        except Exception as e:
            logger.error(f"Pika task {task_id} failed: {e}")
            self.redis_client.hset(
                f"pika:task:{task_id}",
                mapping={
                    "status": TaskStatus.FAILED.value,
                    "error": str(e)
                }
            )
    
    async def _poll_task_status(self, task_id: str):
        """轮询兜底，防止回调丢失"""
        max_polls = 60  # 最多轮询60次
        poll_interval = 2  # 每2秒一次
        
        for _ in range(max_polls):
            await asyncio.sleep(poll_interval)
            
            status = self.redis_client.hget(
                f"pika:task:{task_id}", "status"
            )
            
            if status in [TaskStatus.COMPLETED.value, TaskStatus.FAILED.value]:
                return
            
            # 主动查询Pika任务状态
            try:
                pika_status = await self._query_pika_status(task_id)
                if pika_status == "completed":
                    result_url = await self._get_pika_result(task_id)
                    self.redis_client.hset(
                        f"pika:task:{task_id}",
                        mapping={
                            "status": TaskStatus.COMPLETED.value,
                            "result_url": result_url
                        }
                    )
                    return
                elif pika_status == "failed":
                    self.redis_client.hset(
                        f"pika:task:{task_id}",
                        mapping={
                            "status": TaskStatus.FAILED.value,
                            "error": "Pika generation failed"
                        }
                    )
                    return
            except Exception as e:
                logger.warning(f"Poll error for task {task_id}: {e}")
                continue
        
        # 超时了
        self.redis_client.hset(
            f"pika:task:{task_id}",
            mapping={
                "status": TaskStatus.FAILED.value,
                "error": "Task timeout"
            }
        )

前端那边改成轮询或者WebSocket推送，用户提交任务后不用傻等，可以去干点别的，生成完了再通知他。

这个方案上线后，接口的响应时间从平均12秒降到了200ms（只是提交任务的时间）。图片生成的实际耗时还是12秒左右，但用户体验完全不一样了。

整体架构优化

搞完了三个核心模块，我又把整体架构梳理了一遍。原来的架构是这样的：

用户请求 -> Nginx -> API Gateway -> 业务服务 -> 各个AI服务

问题在于所有请求都走同一个网关，AI这种耗时请求会把网关的连接池打满，影响其他正常接口。

我做了个调整，把AI相关的请求拆到了独立的网关实例上：

普通请求 -> Nginx -> API Gateway (普通实例)
AI请求   -> Nginx -> API Gateway (AI专用实例)

同时在K8s层面，给AI服务单独配了HPA：

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: ai-service-hpa
  namespace: production
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: ai-service
  minReplicas: 3
  maxReplicas: 20
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 60
  - type: Pods
    pods:
      metric:
        name: http_requests_in_flight
      target:
        type: AverageValue
        averageValue: "10"
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 30   # 30秒内持续高负载才扩容
      policies:
      - type: Pods
        value: 4
        periodSeconds: 60              # 每分钟最多加4个Pod
    scaleDown:
      stabilizationWindowSeconds: 300  # 缩容要更保守，5分钟稳定期
      policies:
      - type: Pods
        value: 2
        periodSeconds: 120

这里有个经验：AI服务的扩缩容策略要跟普通服务不一样。AI请求耗时长，如果扩容太激进，流量一过就会有一堆空闲Pod浪费资源。缩容也要更保守，不然流量波动的时候频繁扩缩容，反而影响性能。

缓存策略再优化

除了Embedding的缓存，腾讯混元的结果也可以缓存。很多prompt其实是有重复的，比如用户问"杭州有什么好玩的地方"，这种高频问题完全没必要每次都调模型。

class SmartCache:
    def __init__(self):
        self.redis_client = redis.Redis(...)
        # 用simhash做模糊匹配
        self.simhash_client = SimHashClient()
    
    async def get_or_generate(
        self, 
        prompt: str,
        generate_func,
        similarity_threshold: float = 0.9
    ) -> str:
        """智能缓存：精确匹配 + 模糊匹配"""
        # 1. 精确匹配
        exact_key = f"llm:exact:{hashlib.md5(prompt.encode()).hexdigest()}"
        cached = self.redis_client.get(exact_key)
        if cached:
            logger.info(f"Exact cache hit for prompt: {prompt[:50]}...")
            return cached
        
        # 2. 模糊匹配
        prompt_hash = self.simhash_client.compute(prompt)
        similar_key = self._find_similar(prompt_hash, similarity_threshold)
        if similar_key:
            logger.info(f"Similar cache hit for prompt: {prompt[:50]}...")
            return self.redis_client.get(similar_key)
        
        # 3. 没有命中，调用模型生成
        result = await generate_func(prompt)
        
        # 4. 写入缓存，TTL根据prompt长度动态调整
        # 越长的prompt越具体，缓存时间可以更长
        ttl = min(3600 * 24, 3600 + len(prompt) * 10)
        self.redis_client.setex(exact_key, ttl, result)
        
        # 5. 更新simhash索引
        self.simhash_client.add(prompt_hash, exact_key)
        
        return result
    
    def _find_similar(self, prompt_hash, threshold):
        """在simhash索引中查找相似的prompt"""
        candidates = self.simhash_client.search(prompt_hash)
        for candidate_hash, cache_key in candidates:
            similarity = self.simhash_client.similarity(
                prompt_hash, candidate_hash
            )
            if similarity >= threshold:
                return cache_key
        return None

这个模糊缓存上线后，腾讯混元的实际调用量下降了大概35%，省了不少API费用。产品那边听到能省钱，笑得合不拢嘴。

上线后的效果

折腾了差不多一个月，最后的效果汇总一下：

指标	优化前	优化后	提升幅度
文本生成P99延迟	8.2s	2.1s（流式首token 0.8s）	74%
图片生成接口响应	12.3s	0.2s（异步提交）	98%
Embedding平均耗时	1.2s	0.18s	85%
腾讯混元API调用量	100%	65%	省35%费用
系统最大QPS	50	280	460%

上周五做了一轮全链路压测，数据还是挺漂亮的。领导看了报告，难得夸了我一句。

一些心得体会

回过头来看这次优化，有几点感触比较深：

第一，性能优化不能只看接口响应时间。图片生成这种场景，把同步改异步，接口响应时间确实降了，但用户实际等待时间没变。关键是要配合前端的交互优化，让用户"感觉"快。

第二，缓存是万金油，但不能乱用。Embedding的缓存没问题，因为同样的文本生成的向量是一样的。但LLM的缓存就要小心了，得考虑模糊匹配，不然缓存命中率太低，白白浪费Redis内存。

第三，第三方服务的调用一定要做好兜底。Pika的回调就丢过几次，要不是我加了轮询兜底，用户那边就卡死了。跟外部系统打交道，永远要假设它可能会挂。

第四，监控和告警要提前搞。我在这次优化之前就搭好了全链路追踪，不然出了问题都不知道慢在哪。很多团队都是出了问题才想起来加监控，那时候就晚了。

对了，最近在看阿里和网易的机会，如果有杭州的DevOps或者基础架构岗位，欢迎来聊。我觉得这次项目经验还是挺有含金量的，至少面试的时候有东西可以吹。

好了，咖啡喝完了，该继续干活了。最近在看Kueue做GPU任务调度的方案，等搞完了再来分享。各位同行们，共勉。