AI系统架构实战指南：从单体到混合架构的演进

// 单体应用的结构
/app
  /controllers    // 用户、订单、支付
  /models         // 用户、订单、支付
  /services       // 邮件、通知
  /routes.js

# 错误做法：当成普通 HTTP 请求
async def query_llm(prompt: str) -> str:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.openai.com/v1/chat/completions",
            json={"model": "gpt-4", "messages": [...]}
        )
        return response.json()["choices"][0]["message"]["content"]

class LLMCaller:
    def __init__(self, max_concurrent=5, timeout=30):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.timeout = timeout

    async def call(self, prompt: str, retry=3) -> str:
        for attempt in range(retry):
            async with self.semaphore:
                try:
                    async with asyncio.timeout(self.timeout):
                        return await self._do_call(prompt)
                except asyncio.TimeoutError:
                    if attempt == retry - 1:
                        raise
                    await asyncio.sleep(2 ** attempt)  # 指数退避
                except httpx.HTTPStatusError as e:
                    if e.response.status_code == 429:  # Rate limit
                        wait_time = int(e.response.headers.get("Retry-After", 60))
                        await asyncio.sleep(wait_time)
                        continue
                    raise

async def enhanced_query(query: str) -> str:
    # 快速检索
    docs = await vector_search(query, top_k=3)

    # 评估检索质量
    relevance = await assess_relevance(query, docs)

    if relevance < 0.5:
        # 质量不够，扩展检索
        docs = await vector_search(query, top_k=10)
        relevance = await assess_relevance(query, docs)

        if relevance < 0.3:
            # 还是差，走通用回答
            return await generic_llm_response(query)

    return await llm_summarize(query, docs)

async def chat_with_memory(user_id: str, message: str) -> str:
    # 获取最近对话
    recent_history = await redis.get(f"chat:recent:{user_id}")
    messages = json.loads(recent_history) if recent_history else []

    # 超过 10 轮，做摘要
    if len(messages) > 10:
        summary = await llm_summarize_history(messages[:5])
        messages = messages[5:]
        messages.insert(0, {"role": "system", "content": f"对话摘要：{summary}"})

    messages.append({"role": "user", "content": message})
    response = await llm_call(messages)
    messages.append({"role": "assistant", "content": response})

    # 只保留最近 10 轮
    await redis.set(f"chat:recent:{user_id}", json.dumps(messages[-10:]))
    return response

# Supervisor 配置
[program:model-service]
command=/opt/conda/bin/gunicorn -w 4 -b 0.0.0.0:5000 app:app
autorestart=true
stdout_logfile=/var/log/model-service.log

from prometheus_client import start_http_server, Gauge
import pynvml

gpu_memory_used = Gauge('model_gpu_memory_used_mb', 'GPU memory', ['gpu_id'])
gpu_utilization = Gauge('model_gpu_utilization_percent', 'GPU util', ['gpu_id'])

def update_gpu_metrics():
    for i in range(gpu_count):
        handle = pynvml.nvmlDeviceGetHandleByIndex(i)
        mem_info = pynvml.nvmlDeviceGetMemoryInfo(handle)
        util = pynvml.nvmlDeviceGetUtilizationRates(handle)

        gpu_memory_used.labels(gpu_id=i).set(mem_info.used / 1024 / 1024)
        gpu_utilization.labels(gpu_id=i).set(util.gpu)

upstream model_backend {
    server 10.0.1.10:5000 max_fails=3 fail_timeout=30s;
    server 10.0.1.11:5000 max_fails=3 fail_timeout=30s;
    least_conn;  # 模型推理是长连接任务
}

import psycopg2
from psycopg2 import pool

# 全局连接池（关键！）
connection_pool = None

def get_connection():
    global connection_pool
    if connection_pool is None:
        connection_pool = psycopg2.pool.SimpleConnectionPool(
            minconn=1, maxconn=5,
            host=db_host, database=db_name,
            user=db_user, password=db_password
        )
    return connection_pool.getconn()

@app.post("/generate-summary")
async def generate_summary(content: str):
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[{"role": "user", "content": f"摘要：{content}"}]
    )
    return {"summary": response.choices[0].message.content}

@app.post("/process-content")
async def process_content(content: str, background_tasks: BackgroundTasks):
    task_id = f"task_{int(time.time())}"
    redis.hset(f"task:{task_id}", mapping={"status": "pending"})

    # 后台异步处理
    background_tasks.add_task(process_with_ai, task_id, content)

    return {"task_id": task_id, "status": "pending"}

class AIModelSelector:
    def __init__(self):
        self.model_mapping = {
            "high_quality": {"model": "gpt-4", "max_tokens": 2000, "temperature": 0.3},
            "general": {"model": "gpt-3.5-turbo", "max_tokens": 1000, "temperature": 0.7},
            "fast": {"model": "gpt-3.5-turbo", "max_tokens": 500, "temperature": 0.5}
        }

    def select_model(self, task_type: str, priority: str = "normal"):
        if priority == "high":
            return self.model_mapping["high_quality"]
        if task_type in ["content_creation", "code_review"]:
            return self.model_mapping["high_quality"]
        elif task_type in ["summarization", "tag_extraction"]:
            return self.model_mapping["general"]
        else:
            return self.model_mapping["fast"]

class ContentAIWorkflow:
    async def process_content(self, content_id: int, content: dict):
        # AI 初步评估
        ai_assessment = await self._ai_pre_assessment(content)

        # 根据评估走不同流程
        if ai_assessment["risk_level"] == "high":
            await self._escalate_to_human(content_id, ai_assessment)
        elif ai_assessment["risk_level"] == "medium":
            additional_checks = await self._run_additional_checks(content)
            if additional_checks["passed"]:
                await self._approve_content(content_id)
            else:
                await self._escalate_to_human(content_id, additional_checks)
        else:
            await self._approve_content(content_id)

        # 记录反馈，持续优化
        await self._record_feedback(content_id, ai_assessment)

def process_order(user_id, amount):
    inventory.reduce_stock(amount)       # 1. 扣库存
    order = order_service.create_order() # 2. 创建订单
    payment_service.charge(user_id)      # 3. 支付
    # 问题：如果支付失败，库存已经扣了

class CreateOrderSaga:
    def execute(self, user_id, amount):
        try:
            inventory_service.reduce_stock(amount)
            order = order_service.create_order(user_id, amount)
            payment_service.charge(user_id, amount)
            return order
        except Exception as e:
            # 补偿操作
            inventory_service.restore_stock(amount)
            raise e

const consul = require('consul')();

// 服务注册
await consul.agent.service.register({
    name: serviceName,
    address: serviceUrl,
    check: {
        http: `http://${serviceUrl}/health`,
        interval: '10s'
    }
});

// 服务发现
const services = await consul.agent.service.list();

from opentelemetry import trace

tracer = trace.get_tracer(__name__)

@tracer.start_as_current_span("process_order")
def process_order(order_id):
    with tracer.start_as_span("get_order"):
        order = get_order(order_id)
    with tracer.start_as_span("process_payment"):
        process_payment(order)

SHADOW_RATES = {
    "low": 1.0,    # 低风险 100% 影子
    "medium": 0.5, # 中风险 50%
    "high": 0.2    # 高风险 20%
}

# 简单任务用便宜模型
def select_model(task_type):
    if task_type == "simple_qa":
        return "gpt-3.5-turbo"
    elif task_type == "code_generation":
        return "gpt-4"
    else:
        return "claude-3-5-sonnet"

# Prometheus 告警规则
groups:
- name: model_service_alerts
  rules:
  - alert: HighQueueLength
    expr: redis_queue_length{queue="model_tasks"} > 1000
    for: 5m
    labels:
      severity: warning

  - alert: ModelServiceDown
    expr: up{job="model-service"} == 0
    for: 1m
    labels:
      severity: critical

  - alert: HighGPUMemory
    expr: model_gpu_memory_usage_percent > 90
    for: 10m
    labels:
      severity: warning