AI应用开发：从原型到生产的工程实践

将AI能力落地到产品中，需要解决许多工程化问题。

模型选型

选型维度

维度	考虑因素
能力	任务匹配度、输出质量
成本	Token价格、调用频率
延迟	响应时间、用户体验
稳定	可用性SLA、限流策略

主流模型对比

GPT-4: 能力最强，成本较高
GPT-3.5: 性价比高，适合大多数场景
Claude: 长文本处理优秀
开源模型: 可私有部署，数据安全

API设计

封装层设计

class AIService:
    def __init__(self, model: str, api_key: str):
        self.client = OpenAI(api_key=api_key)
        self.model = model
    
    async def chat(self, messages: list, **kwargs):
        try:
            response = await self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                **kwargs
            )
            return response.choices[0].message.content
        except RateLimitError:
            # 处理限流
            await asyncio.sleep(1)
            return await self.chat(messages, **kwargs)
        except APIError as e:
            # 记录错误并降级
            logger.error(f"API error: {e}")
            return self.fallback_response()

重试策略

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry=retry_if_exception_type(RateLimitError)
)
async def call_with_retry(self, messages):
    return await self.chat(messages)

成本优化

Token优化

def optimize_prompt(prompt: str) -> str:
    # 移除冗余空格
    prompt = " ".join(prompt.split())
    # 压缩重复内容
    prompt = re.sub(r'(.)\1{3,}', r'\1\1\1', prompt)
    return prompt

缓存策略

class CachedAIService:
    def __init__(self, ttl: int = 3600):
        self.cache = TTLCache(maxsize=1000, ttl=ttl)
    
    async def chat(self, prompt: str):
        cache_key = hashlib.md5(prompt.encode()).hexdigest()
        
        if cache_key in self.cache:
            return self.cache[cache_key]
        
        response = await self.ai_service.chat(prompt)
        self.cache[cache_key] = response
        return response

成本监控

def track_tokens(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        result = await func(*args, **kwargs)
        
        # 记录Token使用
        usage = result.usage
        cost = calculate_cost(
            usage.prompt_tokens,
            usage.completion_tokens,
            model=kwargs.get('model', 'gpt-3.5-turbo')
        )
        
        metrics.record('ai.tokens.prompt', usage.prompt_tokens)
        metrics.record('ai.tokens.completion', usage.completion_tokens)
        metrics.record('ai.cost', cost)
        
        return result
    return wrapper

监控告警

关键指标

metrics:
  - ai.request.count: 请求总数
  - ai.request.latency: 响应延迟
  - ai.tokens.total: Token消耗
  - ai.cost.total: 成本累计
  - ai.error.rate: 错误率

告警规则

alerts:
  - name: high_error_rate
    condition: ai.error.rate > 0.05
    action: notify_team
    
  - name: cost_spike
    condition: ai.cost.daily > threshold * 1.5
    action: notify_team

最佳实践

1. 渐进式上线：先小流量验证
2. 降级方案：AI服务不可用时的备选
3. 用户反馈：收集并持续优化
4. 成本预算：设置每日/每月限额