AI编程 性能优化教程｜附源码

在 AI 编程场景中，很多开发者会把注意力集中在“模型能不能跑起来”“接口能不能调用成功”“效果是否足够准确”上，却容易忽略一个非常关键的问题：性能优化。

性能优化不仅仅是让程序跑得更快，它还直接影响：

• 用户体验：响应是否及时；
• 服务器成本：是否需要更多 GPU/CPU 资源；
• 系统稳定性：高并发时是否容易崩溃；
• 产品可扩展性：后续功能是否容易扩展；
• AI 应用落地能力：能否真正投入生产环境。

本文将以一个典型的 AI 编程场景为例，系统讲解如何对 AI 应用进行性能优化，并附带完整源码示例，方便你直接学习和改造。

一、AI 编程中的性能瓶颈通常在哪里？

在传统 Web 开发中，性能瓶颈可能来自数据库、网络、缓存、算法复杂度等方面。而在 AI 编程中，性能瓶颈更加复杂，通常包括以下几个方面。

1. 模型推理速度慢

AI 应用最核心的环节通常是模型推理。例如：

• 文本分类；
• 图像识别；
• 语音转文字；
• 大模型问答；
• 向量检索；
• RAG 知识库问答。

模型越大，推理时间通常越长。

例如调用一个大语言模型接口，可能一次请求需要 1 秒、3 秒，甚至 10 秒以上。如果用户量上来，延迟会非常明显。

2. 重复计算过多

很多 AI 程序存在大量重复计算问题。

例如：

• 同一个问题被用户多次提问；
• 同一段文本反复生成 embedding；
• 同一张图片反复做特征提取；
• 同一个 prompt 每次都重新拼接复杂上下文；
• 知识库每次查询都重新读取和切分文档。

这些操作如果不加缓存，会造成资源浪费。

3. 数据预处理效率低

AI 应用在模型推理之前，往往要进行数据预处理。

例如：

• 文本清洗；
• 分词；
• 图片缩放；
• 音频采样；
• 文档切片；
• JSON 数据解析；
• 向量格式转换。

如果预处理代码写得不合理，即使模型本身很快，整体系统也会变慢。

4. 同步阻塞导致并发能力差

很多新手写 AI 服务时，会采用最简单的同步代码。

例如：

result = call_ai_model(prompt)
return result

在单用户测试时没有问题，但如果同时有 100 个用户请求，请求会堆积，响应时间快速升高。

尤其是大模型接口调用、向量数据库查询、文件读取等操作，本身具有明显的 I/O 等待时间，非常适合使用异步或并发优化。

5. 缺乏缓存与任务队列

AI 应用经常涉及耗时任务：

• 长文本总结；
• 批量图片识别；
• 批量生成内容；
• 文档向量化；
• 离线知识库构建；
• 多轮智能体任务执行。

如果这些任务都在请求线程中同步完成，用户会一直等待，服务器也容易超时。

更合理的做法是使用：

• 缓存；
• 异步任务；
• 消息队列；
• 后台任务；
• 批处理机制。

二、本文示例：构建一个可优化的 AI 文本分析服务

本文将用 Python + FastAPI 写一个简单 AI 文本分析服务。

它的功能是：

1. 接收用户提交的文本；
2. 对文本进行清洗；
3. 调用模拟 AI 模型进行情感分析；
4. 返回分析结果；
5. 加入缓存；
6. 加入异步优化；
7. 加入批处理接口；
8. 提供性能测试方法。

为了方便学习，本文不会依赖真实大模型接口，而是使用一个模拟 AI 推理函数。你可以很容易把它替换为 OpenAI、通义千问、DeepSeek、Claude、Gemini 或本地模型。

三、项目结构设计

建议项目结构如下：

ai-performance-demo/
├── app.py
├── model.py
├── cache.py
├── utils.py
├── requirements.txt
└── README.md

各文件作用如下：

四、基础版本：一个未优化的 AI 服务

我们先写一个最基础的版本。

1. 安装依赖

创建 requirements.txt：

fastapi==0.115.0
uvicorn==0.30.6
pydantic==2.8.2

安装依赖：

pip install -r requirements.txt

2. 编写模拟模型

创建 model.py：

import time
import random

def analyze_sentiment(text: str) -> dict:
    """
    模拟 AI 情感分析模型。
    这里用 time.sleep 模拟模型推理耗时。
    实际项目中，你可以替换为真实 AI 模型或大模型 API。
    """
    time.sleep(1.5)

    positive_words = ["喜欢", "优秀", "满意", "好", "开心", "推荐", "强大"]
    negative_words = ["差", "糟糕", "失望", "不好", "讨厌", "慢", "崩溃"]

    score = 0

    for word in positive_words:
        if word in text:
            score += 1

    for word in negative_words:
        if word in text:
            score -= 1

    if score > 0:
        label = "positive"
    elif score < 0:
        label = "negative"
    else:
        label = random.choice(["neutral", "positive", "negative"])

    return {
        "label": label,
        "score": score,
        "length": len(text)
    }

3. 编写工具函数

创建 utils.py：

import re

def clean_text(text: str) -> str:
    """
    文本清洗：
    1. 去除多余空白；
    2. 去除特殊不可见字符；
    3. 控制文本长度。
    """
    text = text.strip()
    text = re.sub(r"\s+", " ", text)
    text = text.replace("\u200b", "")

    max_length = 2000
    if len(text) > max_length:
        text = text[:max_length]

    return text

4. 编写 FastAPI 服务

创建 app.py：

from fastapi import FastAPI
from pydantic import BaseModel

from model import analyze_sentiment
from utils import clean_text

app = FastAPI(title="AI Performance Demo")

class AnalyzeRequest(BaseModel):
    text: str

@app.post("/analyze")
def analyze(req: AnalyzeRequest):
    text = clean_text(req.text)
    result = analyze_sentiment(text)

    return {
        "text": text,
        "result": result
    }

启动服务：

uvicorn app:app --reload

测试接口：

curl -X POST "http://127.0.0.1:8000/analyze" \
-H "Content-Type: application/json" \
-d '{"text":"这个 AI 编程工具非常好，我很满意，强烈推荐。"}'

返回示例：

{
  "text": "这个 AI 编程工具非常好，我很满意，强烈推荐。",
  "result": {
    "label": "positive",
    "score": 3,
    "length": 25
  }
}

这个版本可以正常运行，但有明显问题：每次请求都要等待约 1.5 秒。如果 100 个用户请求相同文本，也会重复执行 100 次模型推理。

五、优化一：使用缓存减少重复推理

缓存是 AI 性能优化中最直接、最有效的手段之一。

适合缓存的内容包括：

• 相同 prompt 的回答；
• 相同文本的 embedding；
• 相同图片的识别结果；
• 相同文档的切片结果；
• 相同知识库查询结果；
• 大模型接口返回结果。

这里我们使用简单内存缓存演示。

1. 编写缓存模块

创建 cache.py：

import time
import hashlib
from typing import Any, Optional

class SimpleTTLCache:
    """
    一个简单的 TTL 内存缓存。
    ttl_seconds 表示缓存有效期。
    """

    def __init__(self, ttl_seconds: int = 300):
        self.ttl_seconds = ttl_seconds
        self.store = {}

    def _now(self) -> float:
        return time.time()

    def make_key(self, text: str) -> str:
        return hashlib.sha256(text.encode("utf-8")).hexdigest()

    def get(self, key: str) -> Optional[Any]:
        item = self.store.get(key)

        if not item:
            return None

        value, expire_at = item

        if self._now() > expire_at:
            del self.store[key]
            return None

        return value

    def set(self, key: str, value: Any):
        expire_at = self._now() + self.ttl_seconds
        self.store[key] = (value, expire_at)

    def clear(self):
        self.store.clear()

2. 在接口中使用缓存

修改 app.py：

from fastapi import FastAPI
from pydantic import BaseModel

from model import analyze_sentiment
from utils import clean_text
from cache import SimpleTTLCache

app = FastAPI(title="AI Performance Demo")

cache = SimpleTTLCache(ttl_seconds=600)

class AnalyzeRequest(BaseModel):
    text: str

@app.post("/analyze")
def analyze(req: AnalyzeRequest):
    text = clean_text(req.text)
    cache_key = cache.make_key(text)

    cached_result = cache.get(cache_key)
    if cached_result is not None:
        return {
            "text": text,
            "result": cached_result,
            "cache_hit": True
        }

    result = analyze_sentiment(text)
    cache.set(cache_key, result)

    return {
        "text": text,
        "result": result,
        "cache_hit": False
    }

第一次请求：

{
  "cache_hit": false
}

第二次请求相同文本：

{
  "cache_hit": true
}

此时第二次请求几乎会立即返回。

3. 缓存优化注意事项

缓存虽然简单，但在真实项目中需要注意：

1）缓存 Key 要稳定

如果用户输入中存在多余空格、换行、全角半角差异，可能导致缓存无法命中。

所以要先做文本标准化，再生成缓存 Key。

2）缓存不能无限增长

内存缓存如果没有过期策略，可能导致内存不断升高。

常见策略：

• TTL 过期；
• LRU 最近最少使用淘汰；
• 最大容量限制；
• 定期清理。

3）生产环境建议使用 Redis

内存缓存只适合单进程演示。

生产环境中更推荐：

• Redis；
• Memcached；
• 数据库存储；
• 向量数据库缓存；
• 对象存储缓存。

六、优化二：异步接口提升并发能力

FastAPI 支持异步接口。如果你的 AI 调用是网络 I/O 型，比如请求第三方大模型 API，使用异步可以显著提升并发能力。

我们先模拟一个异步模型调用。

1. 增加异步模型函数

修改 model.py：

import time
import random
import asyncio

def analyze_sentiment(text: str) -> dict:
    """
    同步模拟模型。
    """
    time.sleep(1.5)

    return _analyze(text)

async def analyze_sentiment_async(text: str) -> dict:
    """
    异步模拟模型。
    用 asyncio.sleep 模拟网络 I/O 等待。
    """
    await asyncio.sleep(1.5)

    return _analyze(text)

def _analyze(text: str) -> dict:
    positive_words = ["喜欢", "优秀", "满意", "好", "开心", "推荐", "强大"]
    negative_words = ["差", "糟糕", "失望", "不好", "讨厌", "慢", "崩溃"]

    score = 0

    for word in positive_words:
        if word in text:
            score += 1

    for word in negative_words:
        if word in text:
            score -= 1

    if score > 0:
        label = "positive"
    elif score < 0:
        label = "negative"
    else:
        label = random.choice(["neutral", "positive", "negative"])

    return {
        "label": label,
        "score": score,
        "length": len(text)
    }

2. 编写异步接口

修改 app.py：

from fastapi import FastAPI
from pydantic import BaseModel

from model import analyze_sentiment, analyze_sentiment_async
from utils import clean_text
from cache import SimpleTTLCache

app = FastAPI(title="AI Performance Demo")

cache = SimpleTTLCache(ttl_seconds=600)

class AnalyzeRequest(BaseModel):
    text: str

@app.post("/analyze")
def analyze(req: AnalyzeRequest):
    text = clean_text(req.text)
    cache_key = cache.make_key(text)

    cached_result = cache.get(cache_key)
    if cached_result is not None:
        return {
            "text": text,
            "result": cached_result,
            "cache_hit": True,
            "mode": "sync"
        }

    result = analyze_sentiment(text)
    cache.set(cache_key, result)

    return {
        "text": text,
        "result": result,
        "cache_hit": False,
        "mode": "sync"
    }

@app.post("/analyze_async")
async def analyze_async(req: AnalyzeRequest):
    text = clean_text(req.text)
    cache_key = cache.make_key(text)

    cached_result = cache.get(cache_key)
    if cached_result is not None:
        return {
            "text": text,
            "result": cached_result,
            "cache_hit": True,
            "mode": "async"
        }

    result = await analyze_sentiment_async(text)
    cache.set(cache_key, result)

    return {
        "text": text,
        "result": result,
        "cache_hit": False,
        "mode": "async"
    }

3. 同步与异步的区别

同步代码适合：

• CPU 密集型计算；
• 本地小模型推理；
• 简单脚本；
• 并发量不高的服务。

异步代码适合：

• 调用大模型 API；
• 请求第三方服务；
• 查询数据库；
• 查询向量数据库；
• 文件上传下载；
• 网络 I/O 等待较多的场景。

需要注意的是：异步并不会让单次模型推理变快，它主要提升并发吞吐能力。

例如单次请求仍然是 1.5 秒，但系统可以同时处理更多请求，而不是一个一个阻塞等待。

七、优化三：批处理提升吞吐量

AI 模型通常更适合批处理。

例如本来你要处理 100 条文本，如果逐条调用模型，需要调用 100 次。如果模型支持 batch，一次传入 100 条，通常会更高效。

我们模拟一个批量分析接口。

1. 增加批量模型函数

修改 model.py：

import time
import random
import asyncio

def analyze_sentiment(text: str) -> dict:
    time.sleep(1.5)
    return _analyze(text)

async def analyze_sentiment_async(text: str) -> dict:
    await asyncio.sleep(1.5)
    return _analyze(text)

def analyze_sentiment_batch(texts: list[str]) -> list[dict]:
    """
    模拟批处理。
    假设单条处理需要 1.5 秒；
    批处理固定开销 1 秒，每条额外增加 0.1 秒。
    """
    time.sleep(1.0 + 0.1 * len(texts))
    return [_analyze(text) for text in texts]

def _analyze(text: str) -> dict:
    positive_words = ["喜欢", "优秀", "满意", "好", "开心", "推荐", "强大"]
    negative_words = ["差", "糟糕", "失望", "不好", "讨厌", "慢", "崩溃"]

    score = 0

    for word in positive_words:
        if word in text:
            score += 1

    for word in negative_words:
        if word in text:
            score -= 1

    if score > 0:
        label = "positive"
    elif score < 0:
        label = "negative"
    else:
        label = random.choice(["neutral", "positive", "negative"])

    return {
        "label": label,
        "score": score,
        "length": len(text)
    }

2. 增加批量接口

修改 app.py：

from fastapi import FastAPI
from pydantic import BaseModel, Field

from model import analyze_sentiment, analyze_sentiment_async, analyze_sentiment_batch
from utils import clean_text
from cache import SimpleTTLCache

app = FastAPI(title="AI Performance Demo")

cache = SimpleTTLCache(ttl_seconds=600)

class AnalyzeRequest(BaseModel):
    text: str

class BatchAnalyzeRequest(BaseModel):
    texts: list[str] = Field(..., max_length=100)

@app.post("/analyze")
def analyze(req: AnalyzeRequest):
    text = clean_text(req.text)
    cache_key = cache.make_key(text)

    cached_result = cache.get(cache_key)
    if cached_result is not None:
        return {
            "text": text,
            "result": cached_result,
            "cache_hit": True,
            "mode": "sync"
        }

    result = analyze_sentiment(text)
    cache.set(cache_key, result)

    return {
        "text": text,
        "result": result,
        "cache_hit": False,
        "mode": "sync"
    }

@app.post("/analyze_async")
async def analyze_async(req: AnalyzeRequest):
    text = clean_text(req.text)
    cache_key = cache.make_key(text)

    cached_result = cache.get(cache_key)
    if cached_result is not None:
        return {
            "text": text,
            "result": cached_result,
            "cache_hit": True,
            "mode": "async"
        }

    result = await analyze_sentiment_async(text)
    cache.set(cache_key, result)

    return {
        "text": text,
        "result": result,
        "cache_hit": False,
        "mode": "async"
    }

@app.post("/analyze_batch")
def analyze_batch(req: BatchAnalyzeRequest):
    cleaned_texts = [clean_text(text) for text in req.texts]

    results = [None] * len(cleaned_texts)
    miss_texts = []
    miss_indexes = []

    for index, text in enumerate(cleaned_texts):
        cache_key = cache.make_key(text)
        cached_result = cache.get(cache_key)

        if cached_result is not None:
            results[index] = {
                "text": text,
                "result": cached_result,
                "cache_hit": True
            }
        else:
            miss_texts.append(text)
            miss_indexes.append(index)

    if miss_texts:
        batch_results = analyze_sentiment_batch(miss_texts)

        for text, index, result in zip(miss_texts, miss_indexes, batch_results):
            cache_key = cache.make_key(text)
            cache.set(cache_key, result)

            results[index] = {
                "text": text,
                "result": result,
                "cache_hit": False
            }

    return {
        "count": len(results),
        "items": results
    }

3. 批处理效果分析

假设有 10 条文本：

逐条处理

1.5 秒 × 10 = 15 秒

批量处理

1.0 秒 + 0.1 秒 × 10 = 2 秒

性能差异非常明显。

在真实 AI 项目中，批处理常用于：

• 批量生成 embedding；
• 批量文本分类；
• 批量 OCR；
• 批量图像识别；
• 批量内容审核；
• 批量推荐排序；
• 离线数据处理。

八、优化四：避免无意义的大文本输入

很多 AI 应用性能差，是因为输入内容过大。

例如：

• 用户上传了几万字文档；
• Prompt 拼接了太多历史对话；
• RAG 检索返回了过多片段；
• 图片未压缩就直接送入模型；
• 音频未切片就整体识别。

模型输入越长，推理成本越高。尤其是大语言模型，输入 token 越多，延迟和费用都会增加。

文本长度控制策略

可以采用以下策略：

1. 设置最大输入长度；
2. 对长文本做摘要；
3. 对文档做分段；
4. 只保留高相关内容；
5. 对历史对话做压缩；
6. RAG 检索时控制 top_k；
7. 移除无意义格式字符。

在本文源码中，clean_text 已经做了最大长度限制：

max_length = 2000
if len(text) > max_length:
    text = text[:max_length]

生产环境中不建议简单截断所有文本，而是根据业务做智能处理。例如：

• 新闻摘要：保留标题、导语和关键段落；
• 客服问答：保留最近几轮高相关对话；
• 合同审查：先切分，再按章节处理；
• 知识库问答：使用向量检索筛选相关片段。

九、优化五：使用性能计时器定位瓶颈

性能优化不能靠猜，必须靠数据。

你需要知道每一步耗时：

• 请求解析耗时；
• 文本清洗耗时；
• 缓存查询耗时；
• 模型推理耗时；
• 数据库查询耗时；
• 返回结果序列化耗时。

我们可以增加一个简单计时器。

1. 增加计时工具

修改 utils.py：

import re
import time
from contextlib import contextmanager

def clean_text(text: str) -> str:
    text = text.strip()
    text = re.sub(r"\s+", " ", text)
    text = text.replace("\u200b", "")

    max_length = 2000
    if len(text) > max_length:
        text = text[:max_length]

    return text

@contextmanager
def timer(name: str):
    start = time.perf_counter()
    yield
    end = time.perf_counter()
    print(f"[TIMER] {name}: {(end - start) * 1000:.2f} ms")

2. 在接口中使用计时器

示例：

@app.post("/analyze")
def analyze(req: AnalyzeRequest):
    with timer("clean_text"):
        text = clean_text(req.text)

    with timer("cache_get"):
        cache_key = cache.make_key(text)
        cached_result = cache.get(cache_key)

    if cached_result is not None:
        return {
            "text": text,
            "result": cached_result,
            "cache_hit": True,
            "mode": "sync"
        }

    with timer("model_inference"):
        result = analyze_sentiment(text)

    with timer("cache_set"):
        cache.set(cache_key, result)

    return {
        "text": text,
        "result": result,
        "cache_hit": False,
        "mode": "sync"
    }

控制台会输出：

[TIMER] clean_text: 0.12 ms
[TIMER] cache_get: 0.05 ms
[TIMER] model_inference: 1501.23 ms
[TIMER] cache_set: 0.03 ms

这时你就能清楚知道瓶颈在模型推理，而不是文本清洗或缓存。

十、优化六：生产环境中的更多高级优化方案

上面的示例适合入门，但如果进入生产环境，还需要考虑更多优化手段。

1. 模型量化

如果你使用本地模型，可以通过量化减少模型体积和推理成本。

常见量化方式：

• FP32；
• FP16；
• INT8；
• INT4。

一般来说，精度越低，速度越快、显存占用越低，但可能带来一定效果损失。

适用场景：

• 本地大模型部署；
• 边缘设备推理；
• GPU 显存不足；
• 高并发推理服务。

2. 模型蒸馏

模型蒸馏是用一个大模型训练一个更小的模型。

例如：

• 用大模型生成标注数据；
• 用小模型学习大模型输出；
• 在线上使用小模型推理。

这样可以显著降低成本。

适用场景：

• 文本分类；
• 意图识别；
• 情感分析；
• 内容审核；
• 推荐排序；
• 高频低复杂度任务。

3. 向量检索优化

如果你的 AI 应用包含 RAG 知识库，向量检索性能非常重要。

优化方向包括：

• embedding 批量生成；
• 向量维度压缩；
• 使用 HNSW 索引；
• 控制 top_k；
• 增加关键词过滤；
• 增加 rerank 但限制候选数量；
• 缓存高频查询结果；
• 文档切片大小合理化。

4. Prompt 优化

很多大模型应用慢，是因为 prompt 太长。

常见问题：

• 系统提示词过长；
• 历史对话无限追加；
• RAG 上下文过多；
• Few-shot 示例过多；
• 输出格式要求重复冗长。

优化方式：

• 压缩系统提示词；
• 控制历史轮数；
• 使用摘要记忆；
• 只传必要上下文；
• 使用结构化输出；
• 减少无用自然语言描述。

5. 流式输出

对于大模型问答类产品，流式输出不能减少总推理时间，但可以显著改善用户体验。

用户不需要等完整答案生成完，而是可以先看到部分内容。

适合场景：

• AI 聊天；
• AI 写作；
• 代码生成；
• 长文本总结；
• 智能客服。

6. 任务队列

对于耗时较长的 AI 任务，不应该让用户一直阻塞等待。

可以使用：

• Celery；
• RQ；
• Dramatiq；
• Kafka；
• RabbitMQ；
• Redis Stream；
• FastAPI BackgroundTasks。

典型流程：

1. 用户提交任务；
2. 服务立即返回任务 ID；
3. 后台 worker 执行 AI 推理；
4. 用户轮询任务状态；
5. 完成后获取结果。

十一、完整源码汇总

下面给出一个整合后的完整版本。

requirements.txt

fastapi==0.115.0
uvicorn==0.30.6
pydantic==2.8.2

cache.py

import time
import hashlib
from typing import Any, Optional

class SimpleTTLCache:
    def __init__(self, ttl_seconds: int = 300):
        self.ttl_seconds = ttl_seconds
        self.store = {}

    def _now(self) -> float:
        return time.time()

    def make_key(self, text: str) -> str:
        return hashlib.sha256(text.encode("utf-8")).hexdigest()

    def get(self, key: str) -> Optional[Any]:
        item = self.store.get(key)

        if not item:
            return None

        value, expire_at = item

        if self._now() > expire_at:
            del self.store[key]
            return None

        return value

    def set(self, key: str, value: Any):
        expire_at = self._now() + self.ttl_seconds
        self.store[key] = (value, expire_at)

    def clear(self):
        self.store.clear()

utils.py

import re
import time
from contextlib import contextmanager

def clean_text(text: str) -> str:
    text = text.strip()
    text = re.sub(r"\s+", " ", text)
    text = text.replace("\u200b", "")

    max_length = 2000
    if len(text) > max_length:
        text = text[:max_length]

    return text

@contextmanager
def timer(name: str):
    start = time.perf_counter()
    yield
    end = time.perf_counter()
    print(f"[TIMER] {name}: {(end - start) * 1000:.2f} ms")

model.py

import time
import random
import asyncio

def analyze_sentiment(text: str) -> dict:
    time.sleep(1.5)
    return _analyze(text)

async def analyze_sentiment_async(text: str) -> dict:
    await asyncio.sleep(1.5)
    return _analyze(text)

def analyze_sentiment_batch(texts: list[str]) -> list[dict]:
    time.sleep(1.0 + 0.1 * len(texts))
    return [_analyze(text) for text in texts]

def _analyze(text: str) -> dict:
    positive_words = ["喜欢", "优秀", "满意", "好", "开心", "推荐", "强大"]
    negative_words = ["差", "糟糕", "失望", "不好", "讨厌", "慢", "崩溃"]

    score = 0

    for word in positive_words:
        if word in text:
            score += 1

    for word in negative_words:
        if word in text:
            score -= 1

    if score > 0:
        label = "positive"
    elif score < 0:
        label = "negative"
    else:
        label = random.choice(["neutral", "positive", "negative"])

    return {
        "label": label,
        "score": score,
        "length": len(text)
    }

app.py

from fastapi import FastAPI
from pydantic import BaseModel, Field

from model import analyze_sentiment, analyze_sentiment_async, analyze_sentiment_batch
from utils import clean_text, timer
from cache import SimpleTTLCache

app = FastAPI(title="AI Performance Demo")

cache = SimpleTTLCache(ttl_seconds=600)

class AnalyzeRequest(BaseModel):
    text: str

class BatchAnalyzeRequest(BaseModel):
    texts: list[str] = Field(..., max_length=100)

@app.post("/analyze")
def analyze(req: AnalyzeRequest):
    with timer("clean_text"):
        text = clean_text(req.text)

    with timer("cache_get"):
        cache_key = cache.make_key(text)
        cached_result = cache.get(cache_key)

    if cached_result is not None:
        return {
            "text": text,
            "result": cached_result,
            "cache_hit": True,
            "mode": "sync"
        }

    with timer("model_inference"):
        result = analyze_sentiment(text)

    with timer("cache_set"):
        cache.set(cache_key, result)

    return {
        "text": text,
        "result": result,
        "cache_hit": False,
        "mode": "sync"
    }

@app.post("/analyze_async")
async def analyze_async(req: AnalyzeRequest):
    text = clean_text(req.text)
    cache_key = cache.make_key(text)

    cached_result = cache.get(cache_key)
    if cached_result is not None:
        return {
            "text": text,
            "result": cached_result,
            "cache_hit": True,
            "mode": "async"
        }

    result = await analyze_sentiment_async(text)
    cache.set(cache_key, result)

    return {
        "text": text,
        "result": result,
        "cache_hit": False,
        "mode": "async"
    }

@app.post("/analyze_batch")
def analyze_batch(req: BatchAnalyzeRequest):
    cleaned_texts = [clean_text(text) for text in req.texts]

    results = [None] * len(cleaned_texts)
    miss_texts = []
    miss_indexes = []

    for index, text in enumerate(cleaned_texts):
        cache_key = cache.make_key(text)
        cached_result = cache.get(cache_key)

        if cached_result is not None:
            results[index] = {
                "text": text,
                "result": cached_result,
                "cache_hit": True
            }
        else:
            miss_texts.append(text)
            miss_indexes.append(index)

    if miss_texts:
        batch_results = analyze_sentiment_batch(miss_texts)

        for text, index, result in zip(miss_texts, miss_indexes, batch_results):
            cache_key = cache.make_key(text)
            cache.set(cache_key, result)

            results[index] = {
                "text": text,
                "result": result,
                "cache_hit": False
            }

    return {
        "count": len(results),
        "items": results
    }

@app.post("/cache/clear")
def clear_cache():
    cache.clear()
    return {
        "message": "cache cleared"
    }

十二、性能测试方法

可以使用 curl 简单测试，也可以使用压测工具。

1. 使用 curl 测试

启动服务：

uvicorn app:app --reload

请求接口：

curl -X POST "http://127.0.0.1:8000/analyze" \
-H "Content-Type: application/json" \
-d '{"text":"这个产品很好，我很满意，推荐大家使用。"}'

再次请求相同文本：

curl -X POST "http://127.0.0.1:8000/analyze" \
-H "Content-Type: application/json" \
-d '{"text":"这个产品很好，我很满意，推荐大家使用。"}'

你会发现第二次速度明显更快。

2. 使用批量接口测试

curl -X POST "http://127.0.0.1:8000/analyze_batch" \
-H "Content-Type: application/json" \
-d '{
  "texts": [
    "这个产品很好，我很满意。",
    "系统太慢了，经常崩溃。",
    "功能一般，没有特别感觉。",
    "体验优秀，非常推荐。",
    "这次升级让我很失望。"
  ]
}'

3. 使用 ab 压测

如果你安装了 ApacheBench，可以这样测试：

ab -n 100 -c 10 -p data.json -T application/json http://127.0.0.1:8000/analyze

其中 data.json 内容如下：

{
  "text": "这个产品很好，我很满意，推荐大家使用。"
}

参数含义：

十三、AI 性能优化实践清单

最后整理一份 AI 编程性能优化清单，方便你在实际项目中逐项检查。

1. 输入优化

• 是否限制了最大输入长度；
• 是否去除了无意义空白和特殊字符；
• 是否避免传入重复上下文；
• 是否对长文档做了切片；
• 是否控制 RAG 检索数量；
• 是否压缩历史对话。

2. 缓存优化

• 高频请求是否加入缓存；
• embedding 是否缓存；
• prompt 结果是否缓存；
• 知识库检索结果是否缓存；
• 缓存是否设置过期时间；
• 是否有缓存淘汰策略；
• 生产环境是否使用 Redis。

3. 并发优化

• I/O 型调用是否使用异步；
• 是否避免阻塞事件循环；
• 是否合理设置 worker 数量；
• 是否区分 CPU 密集型和 I/O 密集型任务；
• 是否使用连接池；
• 是否设置超时和重试机制。

4. 模型优化

• 是否选择了合适大小的模型；
• 是否可以使用小模型替代大模型；
• 是否可以量化；
• 是否可以蒸馏；
• 是否可以使用批处理；
• 是否开启 GPU 加速；
• 是否有模型预热机制。

5. 系统架构优化

• 长任务是否放入后台队列；
• 是否拆分在线服务和离线任务；
• 是否有熔断和限流；
• 是否有监控和告警；
• 是否记录慢请求日志；
• 是否支持水平扩容；
• 是否避免单点故障。

十四、总结

AI 编程的性能优化不是某一个技巧，而是一整套工程能力。

从本文示例可以看到，一个简单的 AI 文本分析服务，通过以下方法就能明显提升性能：

1. 缓存：避免重复模型推理；
2. 异步：提升 I/O 型任务并发能力；
3. 批处理：提升整体吞吐量；
4. 输入控制：减少无意义计算；
5. 计时分析：用数据定位瓶颈；
6. 架构拆分：把长任务交给后台处理。

如果你正在开发 AI 应用，不要只关注“模型效果”，也要关注“工程性能”。一个真正可上线的 AI 系统，必须同时满足准确性、稳定性、速度和成本控制。

你可以直接复制本文源码运行，也可以将模拟模型替换为真实大模型 API、本地模型或向量数据库，进一步扩展成自己的 AI 服务。