AI+云原生应用开发：从设计到部署运维的全链路实战与提效jzit

[Git Repo] → [CODING CI 流水线] 
   → 单元测试 + 镜像构建（Docker） 
   → 推送至 TCR（腾讯云容器镜像仓库）
   → [CODING CD] 应用部署到 TKE 集群
   → [可观测性] 接入腾讯云 Prometheus + CLS（日志服务）
   → [AI 运维] 使用 CodeBuddy 进行故障诊断和性能剖析

openapi: 3.1.0
info:
  title: Fashion Recommender API
  version: v1
paths:
  /api/v1/recommend:
    post:
      summary: 获取个性化推荐
      requestBody:
        required: true
        content:
          application/json:
            schema:
              type: object
              properties:
                user_id:
                  type: string
                category:
                  type: string
                limit:
                  type: integer
                  default: 10
      responses:
        '200':
          content:
            application/json:
              schema:
                type: array
                items:
                  type: object
                  properties:
                    item_id: string
                    score: number

apiVersion: apps/v1
kind: Deployment
metadata:
  name: recommender
  namespace: prod
spec:
  replicas: 3
  strategy:
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  selector:
    matchLabels:
      app: recommender
  template:
    metadata:
      labels:
        app: recommender
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "8000"
    spec:
      containers:
      - name: app
        image: ccr.ccs.tencentyun.com/myproject/recommender:latest
        ports:
        - containerPort: 8000
          name: http
        resources:
          requests:
            cpu: 500m
            memory: 512Mi
          limits:
            cpu: 1000m
            memory: 1Gi
        livenessProbe:
          httpGet:
            path: /healthz
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 5
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 5
        env:
        - name: MILVUS_HOST
          valueFrom:
            configMapKeyRef:
              name: recommender-config
              key: milvus_host
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: recommender-hpa
  namespace: prod
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: recommender
  minReplicas: 3
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Pods
    pods:
      metric:
        name: requests_per_second
      target:
        type: AverageValue
        averageValue: 500

# app/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
import asyncio
from milvus import MilvusClient  # 假设已安装

app = FastAPI(title="Recommender")

# 初始化 Milvus 客户端（从环境变量读取）
client = MilvusClient(uri=os.getenv("MILVUS_URI"))

tracer = trace.get_tracer(__name__)

class RecommendRequest(BaseModel):
    user_id: str
    category: str
    limit: int = 10

@app.post("/api/v1/recommend")
async def recommend(req: RecommendRequest):
    with tracer.start_as_current_span("recommend_query") as span:
        span.set_attribute("user_id", req.user_id)
        # 1. 获取用户向量（从缓存或特征库）
        user_vector = await get_user_embedding(req.user_id)
        # 2. 执行 Milvus 相似性搜索
        search_params = {"metric_type": "IP", "params": {"nprobe": 10}}
        results = client.search(
            collection_name=req.category,
            data=[user_vector],
            limit=req.limit,
            search_params=search_params,
            output_fields=["item_id"]
        )
        # 3. 构造返回
        items = [{"item_id": hit["entity"]["item_id"], "score": hit["distance"]} for hit in results[0]]
        span.set_attribute("result_count", len(items))
        return items

@app.get("/healthz")
async def healthz():
    return {"status": "ok"}

@app.get("/ready")
async def ready():
    # 检查 Milvus 连接
    if client.health():
        return {"status": "ready"}
    raise HTTPException(status_code=503, detail="Milvus unavailable")

# 启动时挂载 OTel 中间件
FastAPIInstrumentor.instrument_app(app)

import pytest
from httpx import AsyncClient
from unittest.mock import patch, AsyncMock
from app.main import app

@pytest.fixture
async def client():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac

@patch("app.main.client.search")
@patch("app.main.get_user_embedding", new_callable=AsyncMock)
async def test_recommend_success(mock_embed, mock_search, client):
    mock_embed.return_value = [0.1] * 128
    mock_search.return_value = [[{"entity": {"item_id": "A1"}, "distance": 0.99}]]
    resp = await client.post("/api/v1/recommend", json={"user_id": "u1", "category": "dress"})
    assert resp.status_code == 200
    assert len(resp.json()) == 1

FROM python:3.11-slim AS builder
WORKDIR /build
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt -t /install

FROM python:3.11-slim
WORKDIR /app
COPY --from=builder /install /usr/local/lib/python3.11/site-packages
COPY app/ ./app/
ENV PYTHONPATH=/app
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

version: '1.0'
stages:
  - name: test
    steps:
      - script: pip install -r requirements.txt && pytest --cov=app
  - name: build
    steps:
      - script: docker build -t $TCR_REGISTRY/recommender:$CI_COMMIT_SHORT_SHA .
      - script: docker push $TCR_REGISTRY/recommender:$CI_COMMIT_SHORT_SHA
  - name: deploy
    steps:
      - script: |
          helm upgrade --install recommender ./charts \
            --set image.tag=$CI_COMMIT_SHORT_SHA \
            --set milvus.host=$MILVUS_HOST
        env:
          KUBECONFIG: $TKE_KUBECONFIG

rules:
- seriesQuery: 'http_server_requests_seconds_count{namespace="prod",pod!=""}'
  resources:
    overrides:
      namespace: {resource: "namespace"}
      pod: {resource: "pod"}
  name:
    matches: "http_server_requests_seconds_count"
    as: "requests_per_second"
  metricsQuery: 'sum(rate(<<.Series>>{<<.LabelMatchers>>}[1m])) by (<<.GroupBy>>)'

import logging
from pythonjsonlogger import jsonlogger

logger = logging.getLogger()
handler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter('%(asctime)s %(levelname)s %(name)s %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)

from prometheus_client import Counter, Histogram, generate_latest

REQ_COUNT = Counter('http_requests_total', 'Total requests', ['method', 'endpoint'])
REQ_LATENCY = Histogram('http_request_duration_seconds', 'Latency', ['endpoint'])

@app.middleware("http")
async def metrics_middleware(request, call_next):
    REQ_COUNT.labels(method=request.method, endpoint=request.url.path).inc()
    with REQ_LATENCY.labels(endpoint=request.url.path).time():
        return await call_next(request)

from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

provider = TracerProvider()
processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="http://apm-collector:4317"))
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)