AI 应用的性能优化与监控怎么做？从响应速度到资源管理

应用上线后，怎么知道它跑得好不好？慢了怎么优化？

本篇回答三个问题：
1. 性能瓶颈怎么定位？
2. 监控体系怎么搭建？
3. 常见的优化手段有哪些？

性能瓶颈怎么定位？

慢在哪？先量化

不要凭感觉优化。先用数据定位瓶颈：

# backend/app/services/monitor.py
import time
import logging
from functools import wraps

logger = logging.getLogger(__name__)


def timed(logger=None):
    """函数耗时统计装饰器。"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            start = time.time()
            try:
                result = await func(*args, **kwargs)
                return result
            finally:
                elapsed = time.time() - start
                if elapsed > 0.5:  # 超过 500ms 记录慢请求
                    log = logger or logging.getLogger(func.__module__)
                    log.warning(f"Slow call: {func.__name__} took {elapsed:.2f}s")
        return wrapper
    return decorator

慢查询追踪

数据库查询是最常见的瓶颈。用 SQLAlchemy 的日志定位慢查询：

# backend/app/database.py
import logging
logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)

# 只记录超过 1 秒的查询
import time
from sqlalchemy import event
from sqlalchemy.engine import Engine


@event.listens_for(Engine, "before_cursor_execute")
def before_cursor(conn, cursor, statement, parameters, context, executemany):
    conn.info["query_start"] = time.time()


@event.listens_for(Engine, "after_cursor_execute")
def after_cursor(conn, cursor, statement, parameters, context, executemany):
    total = time.time() - conn.info["query_start"]
    if total > 1:
        logging.getLogger("slow_query").warning(f"Slow query ({total:.2f}s): {statement[:200]}")

监控体系怎么搭建？

关键指标

API 层面：
  ├─ 请求延迟：P50, P95, P99
  ├─ 请求量：QPS（每秒查询数）
  └─ 错误率：5xx 比例

AI 层面：
  ├─ LLM 调用延迟
  ├─ Token 消耗量
  └─ 向量检索延迟

系统层面：
  ├─ CPU/内存/GPU 使用率
  ├─ 磁盘 IO
  └─ 网络带宽

用 Prometheus + Grafana 搭建监控

# backend/app/services/metrics.py
from prometheus_client import Counter, Histogram, Gauge, generate_latest
from fastapi import Response
import time

# 定义指标
LLM_LATENCY = Histogram("llm_request_duration_seconds", "LLM API call latency", buckets=(0.1, 0.5, 1.0, 2.0, 5.0, 10.0))
LLM_TOKENS = Counter("llm_tokens_total", "Total tokens consumed", ["model"])
API_REQUESTS = Counter("api_requests_total", "Total API requests", ["method", "endpoint", "status"])
API_LATENCY = Histogram("api_request_duration_seconds", "API request latency")


@app.get("/metrics")
async def metrics():
    """Prometheus 指标接口。"""
    return Response(generate_latest(), media_type="text/plain")

LLM 延迟监控采集

在 LLM 调用处加上监控：

from app.services.metrics import LLM_LATENCY, LLM_TOKENS

with LLM_LATENCY.time():
    response = await client.chat.completions.create(...)

LLM_TOKENS.labels(model="deepseek-chat").inc(response.usage.total_tokens)

健康检查接口

@app.get("/api/health")
async def health_check():
    """健康检查：返回各依赖服务状态。"""
    import redis
    status = {"app": "ok", "database": "unknown", "redis": "unknown"}

    # 检查数据库
    try:
        with models.get_db() as conn:
            conn.execute("SELECT 1")
        status["database"] = "ok"
    except Exception:
        status["database"] = "error"

    # 检查 Redis
    try:
        r = redis.from_url(settings.REDIS_URL)
        r.ping()
        status["redis"] = "ok"
    except Exception:
        status["redis"] = "error"

    overall = "ok" if all(v == "ok" for v in status.values()) else "degraded"
    return {"status": overall, "checks": status}

Docker 容器监控

# 查看容器资源使用
docker stats

# 查看特定容器日志
docker logs --tail 50 --follow know-backend

# 容器级别监控（CPU/内存）
docker stats know-backend --no-stream --format "table {{.Name}}\t{{.CPUPerc}}\t{{.MemUsage}}"

常见的优化手段有哪些？

1. 数据库查询优化

# ❌ 慢：N+1 查询
for kb in knowledge_bases:
    docs = await db.execute(select(Document).where(Document.knowledge_base_id == kb.id))

# ✅ 快：批量查询
# 用 JOIN 一次查出
result = await db.execute(
    select(KnowledgeBase, Document).join(Document).where(...)
)

2. LLM 调用缓存

# backend/app/services/cache.py
import hashlib
import json
from app.config import settings


def llm_cache_key(messages: list) -> str:
    """生成 LLM 调用的缓存 key。"""
    content = json.dumps(messages, ensure_ascii=False, sort_keys=True)
    return f"llm:{hashlib.md5(content.encode()).hexdigest()}"


async def get_cached_llm(messages: list, ttl: int = 3600):
    """带缓存的 LLM 调用。"""
    import redis
    r = redis.from_url(settings.REDIS_URL)
    key = llm_cache_key(messages)

    # 查缓存
    cached = r.get(key)
    if cached:
        return cached.decode()

    # 调用 LLM
    result = await call_llm(messages)

    # 写入缓存
    r.setex(key, ttl, result)
    return result

3. 静态文件缓存

Nginx 配置中已经做了：

location /static/ {
    expires 7d;
    add_header Cache-Control "public, immutable";
}

4. 数据库连接池

# SQLAlchemy 连接池配置
engine = create_async_engine(
    settings.DATABASE_URL,
    pool_size=10,          # 连接池大小
    max_overflow=20,       # 最大溢出连接数
    pool_pre_ping=True,    # 使用前检查连接是否有效
    pool_recycle=3600,     # 1 小时后回收连接
)

5. 前端资源优化

// Vite 配置已包含：
// - 代码分割（Code Splitting）
// - Tree Shaking
// - CSS 压缩
// - 图片压缩

// 额外的优化：组件懒加载
const KnowledgeBaseDetail = lazy(() => import("./pages/KnowledgeBaseDetail"));

优化 checklist

□ 数据库查询是否有 N+1 问题？
□ 慢查询是否加了索引？
□ LLM 响应是否做了缓存？
□ 静态资源是否设置了缓存头？
□ 数据库连接池是否配置了合理大小？
□ 是否有健康检查接口？
□ 容器资源限制是否设置？
□ 关键路径是否加了耗时监控？
□ 是否有错误告警？

总结

性能优化和监控的关键：

方向	做什么	效果
定位瓶颈	慢查询追踪 + 耗时统计	知道慢在哪
LLM 缓存	缓存相同请求的结果	减少 50-80% 的 LLM 调用
数据库优化	索引 + JOIN 替代 N+1	查询速度提升 10-100x
健康检查	定期检测服务状态	及时发现故障
Prometheus	指标采集 + Grafana 可视化	实时了解系统状态

本文是 《AI 全栈开发实战——做一个真正的产品》 系列的第 12 篇。
本文由 Zyentor（智元界）原创发布

AI 全栈开发实战（12）：性能优化与监控——从慢查询定位到 Prometheus 监控

AI 应用的性能优化与监控怎么做？从响应速度到资源管理

性能瓶颈怎么定位？

慢在哪？先量化

慢查询追踪

监控体系怎么搭建？

关键指标

用 Prometheus + Grafana 搭建监控

LLM 延迟监控采集

健康检查接口

Docker 容器监控

常见的优化手段有哪些？

1. 数据库查询优化

2. LLM 调用缓存

3. 静态文件缓存

4. 数据库连接池

5. 前端资源优化

优化 checklist

总结

相关推荐

AI 服务的缓存策略与性能优化

Redis 缓存策略在 AI 应用中的实战

用 Grafana + Prometheus 监控 AI 服务

Python 日志系统在 AI 服务中的最佳实践

Prometheus 客户端在 Python AI 服务中的集成

📖 更多原创