从原型到生产：AI 应用上线完整检查清单

# 差：AI 生成的测试，没有验证任何有意义的东西
def test_calculate_price():
    result = calculate_price(100)
    assert result is not None  # 这什么都没告诉我们

# 好：验证实际业务规则的测试
def test_calculate_price_with_discount():
    # 订单超过 $50 打八折
    assert calculate_price(100, discount_tier="gold") == 80.0

def test_calculate_price_rejects_negative():
    with pytest.raises(ValueError, match="Price cannot be negative"):
        calculate_price(-10)

def test_calculate_price_applies_tax():
    # 税费应在折扣后计算
    result = calculate_price(100, discount_tier="gold", tax_rate=0.1)
    assert result == 88.0  # (100 * 0.8) * 1.1

# 测试 API 层与服务层的正确交互
async def test_create_order_integration():
    # 准备：初始化测试数据库
    user = await create_test_user()
    product = await create_test_product(price=50.0)

    # 执行：调用实际的 API 端点
    response = await client.post("/api/orders", json={
        "user_id": user.id,
        "product_id": product.id,
        "quantity": 2
    })

    # 断言：检查整个链路是否正常工作
    assert response.status_code == 201
    order = response.json()
    assert order["total"] == 100.0
    assert order["status"] == "pending"

    # 验证副作用
    db_order = await get_order(order["id"])
    assert db_order is not None
    assert db_order.user_id == user.id

// 测试完整的用户流程
test('user can complete checkout', async ({ page }) => {
  await page.goto('/products');
  await page.click('[data-testid="add-to-cart-btn"]');
  await page.click('[data-testid="checkout-btn"]');
  await page.fill('#email', '[email protected]');
  await page.fill('#card-number', '4242424242424242');
  await page.click('[data-testid="pay-btn"]');
  await expect(page.locator('.order-confirmation')).toBeVisible();
});

# .github/workflows/test.yml
name: Test Suite
on: [push, pull_request]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Run unit tests
        run: pytest tests/unit/ --cov=src --cov-fail-under=80
      - name: Run integration tests
        run: pytest tests/integration/
      - name: Run E2E tests
        run: npx playwright test

# AI 经常生成这样的代码——非常危险
@app.post("/api/query")
async def query(request: Request):
    body = await request.json()
    result = db.execute(f"SELECT * FROM users WHERE name = '{body['name']}'")
    return result

# 带有正确验证的生产版本
from pydantic import BaseModel, validator

class QueryRequest(BaseModel):
    name: str

    @validator('name')
    def validate_name(cls, v):
        if len(v) > 100:
            raise ValueError('Name too long')
        # 过滤 SQL 注入尝试
        if any(char in v for char in [';', '--', "'", '"']):
            raise ValueError('Invalid characters in name')
        return v.strip()

@app.post("/api/query")
async def query(request: QueryRequest):
    result = db.execute(
        "SELECT * FROM users WHERE name = :name",
        {"name": request.name}
    )
    return result

# 扫描依赖项中的已知漏洞
npm audit
pip-audit
snyk test

# 保持依赖项更新
npm update
pip install --upgrade -r requirements.txt

# N+1 问题：1 次查询获取订单 + N 次查询获取用户
orders = db.query(Order).all()
for order in orders:
    user = db.query(User).filter(User.id == order.user_id).first()
    order.user_name = user.name  # 每个订单触发一次查询

# 修复：预加载，只需 2 次查询
orders = db.query(Order).options(joinedload(Order.user)).all()
for order in orders:
    order.user_name = order.user.name  # 不再有额外查询

import redis
from functools import wraps

cache = redis.Redis(host='localhost', port=6379)

def cached(ttl=300):
    """将函数结果缓存到 Redis，有效期 `ttl` 秒。"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            cache_key = f"{func.__name__}:{hash(str(args) + str(kwargs))}"
            cached_result = cache.get(cache_key)
            if cached_result:
                return json.loads(cached_result)
            result = await func(*args, **kwargs)
            cache.setex(cache_key, ttl, json.dumps(result))
            return result
        return wrapper
    return decorator

@cached(ttl=600)
async def get_product_catalog():
    """缓存 10 分钟，因为商品目录很少变化。"""
    return await db.query(Product).filter(Product.active == True).all()

from celery import Celery

celery_app = Celery('tasks', broker='redis://localhost:6379')

# 不再在请求中同步发邮件
@app.post("/api/orders")
async def create_order(order: OrderCreate):
    db_order = await save_order(order)

    # 将邮件发送交给后台 Worker
    send_order_confirmation.delay(db_order.id, order.email)

    return {"order_id": db_order.id, "status": "created"}

@celery_app.task
def send_order_confirmation(order_id: str, email: str):
    """在后台 Worker 中运行，不阻塞 API。"""
    order = get_order(order_id)
    send_email(to=email, subject="Order Confirmed", body=render_template(order))

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@app.post("/api/login")
@limiter.limit("5/minute")  # 防止暴力破解
async def login(request: Request, credentials: LoginRequest):
    return await authenticate(credentials)

@app.get("/api/data")
@limiter.limit("100/minute")  # 防止 API 滥用
async def get_data(request: Request):
    return await fetch_data()

import structlog

logger = structlog.get_logger()

@app.post("/api/orders")
async def create_order(order: OrderCreate, user: User = Depends(get_current_user)):
    logger.info(
        "order_created",
        user_id=user.id,
        order_total=order.total,
        product_count=len(order.items),
        payment_method=order.payment_method
    )

    try:
        result = await process_order(order)
        logger.info("order_processed", order_id=result.id, duration_ms=result.processing_time)
        return result
    except PaymentError as e:
        logger.error(
            "payment_failed",
            user_id=user.id,
            error_code=e.code,
            error_message=str(e)
        )
        raise HTTPException(status_code=402, detail="Payment failed")

{
  "event": "order_created",
  "user_id": "usr_123",
  "order_total": 99.50,
  "product_count": 3,
  "timestamp": "2026-03-13T10:30:00Z",
  "level": "info"
}

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

# 初始化追踪
provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer(__name__)

@app.post("/api/orders")
async def create_order(order: OrderCreate):
    with tracer.start_as_current_span("create_order") as span:
        span.set_attribute("order.total", order.total)

        with tracer.start_as_current_span("validate_inventory"):
            await check_inventory(order.items)

        with tracer.start_as_current_span("process_payment"):
            payment = await charge_card(order.payment)

        with tracer.start_as_current_span("save_order"):
            result = await save_to_database(order, payment)

        return result

告警触发 → 值班工程师被叫醒 → 手动排查
→ 根因分析 → 手动修复 → 编写事后复盘

告警触发 → AI Agent 自动收集上下文
         → AI 分析可能的根因
         → AI 推荐修复方案并给出置信度
         → 人工审批（或 AI 自动执行低风险修复）
         → AI 生成复盘报告草稿

# 完整的 CI/CD 流水线
name: Deploy
on:
  push:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: npm ci
      - run: npm test -- --coverage
      - run: npm run lint
      - run: npx playwright test

  security:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: npm audit --audit-level=high
      - run: npx semgrep --config=auto

  deploy:
    needs: [test, security]
    runs-on: ubuntu-latest
    steps:
      - name: Deploy to staging
        run: deploy --env staging
      - name: Run smoke tests
        run: npm run test:smoke -- --target=staging
      - name: Deploy to production (canary)
        run: deploy --env production --canary 10%
      - name: Monitor canary metrics
        run: check-metrics --duration 15m --threshold error_rate<0.01
      - name: Full rollout
        run: deploy --env production --canary 100%

1. 需求分析
   ├── 编写清晰的规格说明 / PRD
   ├── 定义验收标准
   └── 明确安全和性能需求

2. 架构设计
   ├── 选择技术栈（考虑 AI 工具支持度）
   ├── 设计系统架构
   ├── 建立项目上下文（CLAUDE.md、设计文档）
   └── 配置工具集成（MCP 服务器、API）

3. 快速原型 [AI：80%+ 自动化]
   ├── 用 AI 生成原型
   ├── 验证核心功能
   └── 收集反馈，完善规格

4. 功能开发 [AI：50-70% 自动化]
   ├── 将需求拆分为子任务
   ├── 分配给 AI Agent 执行
   ├── 检查点审查 + 人工完善
   └── 代码评审

5. 质量保证 [AI：40-60% 自动化]
   ├── 单元 + 集成 + 端到端测试覆盖
   ├── 安全扫描（SAST + DAST）
   ├── 性能基准测试
   └── 无障碍审计

6. 部署 [主要靠工具链自动化]
   ├── CI/CD 流水线搭建
   ├── 金丝雀 / 蓝绿部署
   ├── 监控与告警配置
   └── 回滚方案

7. 运维 [AI：30-50% 自动化]
   ├── 可观测性（日志 + 指标 + 追踪）
   ├── 自动化事件响应
   ├── 定期安全审计
   └── 持续优化

8. 迭代 [回到第 1 步]
   ├── 从生产反馈中提取新需求
   └── 更新上下文和文档