LLM 应用开发实战：构建生产级 AI 应用

应用架构设计

整体架构

class LLMApplication:
    def __init__(self):
        # API 层
        self.api_server = APIServer()
        
        # 服务层
        self.llm_service = LLMService()
        self.embedding_service = EmbeddingService()
        self.vector_store = VectorStore()
        
        # 数据层
        self.database = Database()
        self.cache = RedisCache()
        
        # 监控层
        self.metrics = MetricsCollector()
​

分层架构

1. API 层

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class QueryRequest(BaseModel):
    query: str
    model: str = "gpt-4"
    temperature: float = 0.7

@app.post("/api/query")
async def query_endpoint(request: QueryRequest):
    try:
        result = await llm_service.query(
            query=request.query,
            model=request.model,
            temperature=request.temperature
        )
        return {"answer": result}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
​

2. 服务层

class LLMService:
    def __init__(self, llm_client, cache):
        self.llm = llm_client
        self.cache = cache
    
    async def query(self, query, model, temperature):
        # 检查缓存
        cache_key = f"{model}:{query}:{temperature}"
        cached = await self.cache.get(cache_key)
        if cached:
            return cached
        
        # 调用 LLM
        response = await self.llm.generate(
            model=model,
            prompt=query,
            temperature=temperature
        )
        
        # 缓存结果
        await self.cache.set(cache_key, response, ttl=3600)
        
        return response
​

数据处理

1. 文档处理

import asyncio
from typing import List

class DocumentProcessor:
    async def process_batch(self, documents: List[str]):
        """批量处理文档"""
        # 分块
        chunks = await self.chunk_documents(documents)
        
        # 嵌入
        embeddings = await self.embedding_service.embed_batch(chunks)
        
        # 存储
        await self.vector_store.add_batch(chunks, embeddings)
        
        return {"processed": len(documents)}
    
    async def chunk_documents(self, documents: List[str]):
        """文档分块"""
        chunks = []
        for doc in documents:
            doc_chunks = self.split_text(doc, chunk_size=1000, overlap=200)
            chunks.extend(doc_chunks)
        return chunks
​

2. 流式处理

async def stream_response(query: str):
    """流式响应"""
    async for chunk in llm_client.stream(query):
        yield chunk

# 使用示例
async def chat_endpoint(request: QueryRequest):
    return StreamingResponse(
        stream_response(request.query),
        media_type="text/plain"
    )
​

性能优化

1. 并发处理

import asyncio

class ConcurrentProcessor:
    def __init__(self, max_concurrent=10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    async def process_with_limit(self, task):
        async with self.semaphore:
            return await task
    
    async def batch_process(self, tasks):
        results = await asyncio.gather(*[
            self.process_with_limit(task) 
            for task in tasks
        ])
        return results
​

2. 连接池

from aiohttp import ClientSession, TCPConnector

class LLMClient:
    def __init__(self):
        self.session = ClientSession(
            connector=TCPConnector(limit=100)
        )
    
    async def close(self):
        await self.session.close()
​

错误处理

1. 重试机制

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 call_llm_with_retry(prompt):
    return await llm_client.generate(prompt)
​

2. 降级策略

class FallbackLLMService:
    def __init__(self, primary, backup):
        self.primary = primary
        self.backup = backup
    
    async def query(self, query, model):
        try:
            return await self.primary.query(query, model)
        except Exception as e:
            logger.error(f"主 LLM 失败：{e}，使用备用")
            return await self.backup.query(query, model)
​

监控与日志

1. 指标收集

from prometheus_client import Counter, Histogram

QUERY_COUNT = Counter('llm_queries_total')
QUERY_LATENCY = Histogram('llm_query_latency_seconds')

class MetricsMiddleware:
    async def process_request(self, query):
        start = time.time()
        
        try:
            result = await llm_service.query(query)
            QUERY_COUNT.labels(status="success").inc()
            return result
        except Exception as e:
            QUERY_COUNT.labels(status="error").inc()
            raise
        finally:
            QUERY_LATENCY.observe(time.time() - start)
​

2. 结构化日志

import structlog

logger = structlog.get_logger()

logger.info("llm_query", 
    query=query,
    model=model,
    latency_ms=latency * 1000,
    status="success"
)
​

安全考虑

1. 输入验证

from pydantic import validator

class QueryRequest(BaseModel):
    query: str
    
    @validator('query')
    def validate_query(cls, v):
        if len(v) > 10000:
            raise ValueError('Query too long')
        if not v.strip():
            raise ValueError('Query cannot be empty')
        return v
​

2. 内容过滤

class ContentFilter:
    def __init__(self):
        self.forbidden_words = ['暴力', '色情', '非法']
    
    def filter(self, text):
        """过滤敏感词"""
        for word in self.forbidden_words:
            if word in text:
                raise ValueError(f"Forbidden content detected")
        return text
​

部署策略

1. 容器化

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000""]

EXPOSE 8000
​

2. Kubernetes 部署

apiVersion: apps/v1
kind: Deployment
metadata:
  name: llm-app
spec:
  replicas: 3
  selector:
    matchLabels:
      app: llm-app
  template:
    metadata:
      labels:
        app: llm-app
    spec:
      containers:
      - name: app
        image: llm-app:latest
        ports:
        - containerPort: 8000
        env:
        - name: MODEL_API_KEY
          valueFrom:
            secretKeyRef:
              name: model-api-key
              key: key
        resources:
          requests:
            cpu: "500m"
            memory: "512Mi"
          limits:
            cpu: "1000m"
            memory: "1Gi"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 5
​

成本优化

1. Token 优化

class TokenOptimizer:
    def optimize_prompt(self, prompt, max_tokens=4000):
        """优化 Prompt 长度"""
        if len(prompt) <= max_tokens:
            return prompt
        
        # 压缩 Prompt
        # 1. 移除冗余词汇
        # 2. 简化表达
        # 3. 使用缩写
        compressed = self.compress_prompt(prompt)
        
        return compressed
    
    def compress_prompt(self, prompt):
        """压缩 Prompt"""
        # 简单实现：移除多余空格和换行
        import re
        return re.sub(r'\s+', ' ', prompt).strip()
​

2. 缓存策略

from cachetools import TTLCache

class SmartCache:
    def __init__(self):
        self.cache = TTLCache(maxsize=1000, ttl=3600)
    
    def get(self, key):
        return self.cache.get(key)
    
    def set(self, key, value, ttl=None):
        self.cache.set(key, value, ttl)
​

测试策略

1. 单元测试

import pytest

@pytest.mark.asyncio
async def test_llm_service():
    service = LLMService(mock_llm_client, mock_cache)
    
    result = await service.query(
        query="测试查询",
        model="gpt-4",
        temperature=0.7
    )
    
    assert result is not None
    assert len(result) > 0
​

2. 集成测试

@pytest.mark.asyncio
async def test_api_endpoint():
    from httpx import AsyncClient
    
    async with AsyncClient(app=app) as client:
        response = await client.post(
            "/api/query",
            json={"query": "测试", "model": "gpt-4", "temperature": 0.7}
        )
        
        assert response.status_code == 200
        data = response.json()
        assert "answer" in data
​

最佳实践总结

1. 架构设计：分层架构、模块化
2. 性能优化：并发、缓存、批处理
3. 错误处理：重试、降级、熔断
4. 监控日志：指标收集、结构化日志
5. 安全考虑：输入验证、内容过滤
6. 测试覆盖：单元测试、集成测试
7. 成本控制：优化 token 使用、缓存策略

从原型到生产，构建可靠的 LLM 应用！