AI 生活化产品的架构全景：从单机原型到分布式系统的演进路径规划

AI 生活化产品的架构全景：从单机原型到分布式系统的演进路径规划

一、单机原型的技术债务与规模化瓶颈

AI 生活化产品的初期原型通常是单机架构：一个 Python 服务同时处理 API 路由、LLM 调用、向量检索和数据库读写。原型阶段的核心目标是验证产品逻辑，架构问题被有意推迟。当用户量从 100 增长到 10000 时，单机架构暴露三类瓶颈：LLM 调用延迟从 1 秒升至 8 秒（并发请求排队），向量检索从 50ms 升至 2 秒（单机内存不够容纳完整索引），数据库连接从 10 个升至 200 个（连接池耗尽）。更严重的是技术债务：所有功能耦合在一个服务中，修改 Prompt 模板需要重启整个服务，数据库迁移影响 LLM 调用。架构演进的目标不是一次性重构，而是渐进式拆分：按功能边界逐步拆出独立服务，每步拆分都不影响现有功能运行。

二、架构演进的四阶段路径与依赖关系

架构演进从单机到分布式的四阶段路径，每个阶段的拆分依赖前一个阶段的完成：

三、架构演进各阶段的代码骨架与迁移策略

# AI 生活化产品架构演进 — 各阶段核心组件骨架
# 阶段一：数据层拆分 — 透明代理模式（Strangler Fig）

class RepositoryFactory:
"""数据层透明代理工厂

设计意图：应用层代码不直接连接数据库，
通过工厂获取代理，代理内部决定路由到
单机数据库还是独立数据库服务。
拆分期间：代理同时写入新旧数据库（双写），
读取优先从新数据库读取。
拆分完成后：代理只连接新数据库。
"""

def __init__(self, config: dict):
self._config = config
self._phase = config.get("migration_phase", "dual_write")
# 双写阶段：同时连接新旧数据库
self._old_db = self._create_connection(config["old_db"])
self._new_db = self._create_connection(config["new_db"])

def get_repository(self, repo_type: str):
"""获取数据代理 — 按类型返回不同代理"""
if repo_type == "user":
return DualWriteUserProxy(self._old_db, self._new_db, self._phase)
elif repo_type == "conversation":
return DualWriteConversationProxy(
self._old_db, self._new_db, self._phase
)
elif repo_type == "vector":
return VectorSearchProxy(self._config, self._phase)
raise ValueError(f"Unknown repo type: {repo_type}")

def _create_connection(self, db_config: dict):
"""创建数据库连接（实际实现替换为真实客户端）"""
return db_config # 模拟连接对象

class DualWriteUserProxy:
"""双写用户数据代理

拆分期间：写入同时写入新旧数据库，
读取优先从新数据库读取，失败时回退旧数据库。
拆分完成后：只操作新数据库。
"""

def __init__(self, old_db, new_db, phase: str):
self._old_db = old_db
self._new_db = new_db
self._phase = phase

async def save(self, user_data: dict) -> dict:
"""保存用户数据 — 双写或单写"""
if self._phase == "dual_write":
# 双写：同时写入新旧数据库，忽略旧库写入失败
try:
await self._write_db(self._new_db, user_data)
except Exception:
pass # 新库写入失败不影响旧库
await self._write_db(self._old_db, user_data)
return user_data

# 拆分完成：只写入新数据库
return await self._write_db(self._new_db, user_data)

async def load(self, user_id: str) -> dict:
"""加载用户数据 — 优先新库"""
# 优先从新数据库读取
result = await self._read_db(self._new_db, user_id)
if result:
return result

# 新库无数据时回退旧库（拆分过渡期）
if self._phase == "dual_write":
return await self._read_db(self._old_db, user_id)

raise ValueError(f"User {user_id} not found")

# 阶段二：推理层拆分 — 推理服务独立部署

class InferenceServiceConfig:
"""推理服务配置

设计意图：推理服务从主服务拆出后，
通过 HTTP/RPC 调用，主服务不再直接调用 LLM。
推理服务内部管理优先队列和批量合并。
"""

# 推理服务的部署配置
INFERENCE_SERVICE_URL = "http://inference-service:8080"

# 优先级定义（与之前文章一致）
PRIORITY_LEVELS = {
"interactive": 1, # 用户交互触发的推理
"batch": 2, # 批量任务（日记分析、简报生成）
"background": 3, # 后台任务（模型微调、数据清洗）
}

# 槽位分配：交互 60%、批量 30%、后台 10%
SLOT_DISTRIBUTION = {
"interactive": 0.6,
"batch": 0.3,
"background": 0.1,
}

class InferenceServiceClient:
"""推理服务客户端

主服务通过此客户端调用独立推理服务，
不再直接管理 LLM 连接和优先队列。
"""

def __init__(self, config: InferenceServiceConfig):
self._url = config.INFERENCE_SERVICE_URL
self._priority_levels = config.PRIORITY_LEVELS

async def inference(self, prompt: str, context: str,
priority: str = "interactive") -> dict:
"""调用推理服务

主服务只需传入 prompt、context 和优先级，
推理服务内部处理队列调度和限速重试。
"""
request = {
"prompt": prompt,
"context": context,
"priority": self._priority_levels[priority],
}
# HTTP 调用推理服务（实际实现替换为真实 HTTP 客户端）
return {"response": "推理结果", "latency_ms": 1500}

async def batch_inference(self, prompts: List[str],
priority: str = "batch") -> List[dict]:
"""批量推理 — 合并请求减少 LLM 调用次数"""
request = {
"prompts": prompts,
"priority": self._priority_levels[priority],
"mode": "batch",
}
return [{"response": "批量推理结果"} for _ in prompts]

# 阶段三：网关层引入 — 弹性网关配置

class ElasticGatewayConfig:
"""弹性网关配置

设计意图：网关层统一处理限速、路由和缓存，
下游服务（主服务、推理服务、数据服务）不再
直接面对用户请求，全部通过网关转发。
"""

# 限速配置：按用户等级动态调整
RATE_LIMITS = {
"free": {"rpm": 30, "concurrent": 3},
"basic": {"rpm": 100, "concurrent": 5},
"premium": {"rpm": 500, "concurrent": 10},
}

# 路由配置：按请求类型路由到不同服务
ROUTE_TABLE = {
"/api/chat": "inference-service:8080",
"/api/analyze": "inference-service:8080",
"/api/user": "main-service:3000",
"/api/search": "vector-service:6333",
}

# 缓存配置：双层缓存策略
CACHE_STRATEGY = {
"local_ttl": 60, # 本地缓存 60 秒
"remote_ttl": 300, # 远程缓存 300 秒
"preheat_keys": [ # 预加热的热点数据
"popular_prompts",
"daily_recommendations",
"emotion_templates",
],
}

# 阶段四：观测层闭环 — 三层监控与全链路追踪

class ArchitectureObservabilityConfig:
"""架构可观测性配置

设计意图：分布式架构的故障定位依赖全链路追踪，
三层告警覆盖基础设施、函数性能和质量巡检。
"""

# 三层告警配置
ALERT_TIERS = {
"P0_infrastructure": {
# 基础设施层：CPU/内存/磁盘/网络
"cpu_threshold": 80, # CPU 使用率 > 80%
"memory_threshold": 85, # 内存使用率 > 85%
"disk_threshold": 90, # 磁盘使用率 > 90%
"check_interval": 60, # 每 60 秒检查
},
"P1_function_performance": {
# 函数性能层：API延迟/错误率/吞吐量
"latency_p95": 3000, # P95 延迟 > 3 秒
"error_rate": 0.05, # 错误率 > 5%
"throughput_min": 100, # 最小吞吐量 < 100 RPM
"check_interval": 30, # 每 30 秒检查
},
"P2_quality_patrol": {
# 质量巡检层：LLM输出质量/缓存命中率/降级状态
"llm_quality_score": 0.8, # LLM 输出质量 < 0.8
"cache_hit_rate": 0.7, # 缓存命中率 < 70%
"degradation_level": "heavy", # 降级层级 ≥ 重度
"check_interval": 1800, # 每 30 分钟巡检
},
}

# 全链路追踪配置（OpenTelemetry）
TRACING_CONFIG = {
"service_name": "ai-life-app",
"exporter": "otlp",
"endpoint": "http://otel-collector:4317",
"sampling_rate": 0.1, # 10% 请求采样（生产环境）
"context_propagation": "w3c_trace_context",
}

四、架构演进的渐进式迁移策略与回滚保障

架构演进不是一次性重构，而是四阶段渐进拆分。每个阶段的迁移策略遵循 Strangler Fig 模式：新旧系统并行运行，逐步将流量从旧系统迁移到新系统。阶段一数据层拆分的关键是双写策略：写入同时写入新旧数据库，读取优先新库回退旧库。双写持续 7 天后验证新库数据完整性，确认无误后停止旧库写入。阶段二推理层拆分的关键是服务发现：主服务通过配置中心的 URL 调用推理服务，推理服务不可用时回退到主服务内置的 LLM 调用（本地回退）。阶段三网关层引入的关键是流量切换：网关先以 10% 流量转发，90% 流量仍走主服务直连路由，确认网关无异常后逐步提升到 100%。阶段四观测层闭环的关键是告警阈值校准：初期用宽松阈值避免误报，运行 7 天后根据实际数据收紧阈值。每个阶段的回滚保障是：拆分前保留旧系统的完整功能，新系统异常时一键回退到旧系统路由，数据不丢失、服务不中断。

# 架构演进的渐进式迁移与回滚保障
class MigrationPhaseManager:
"""迁移阶段管理器

设计意图：管理四阶段迁移的进度和回滚，
每个阶段有独立的健康检查和回滚触发条件。
"""

PHASES = [
"data_layer", # 阶段一：数据层拆分
"inference_layer", # 阶段二：推理层拆分
"gateway_layer", # 阶段三：网关层引入
"observability", # 阶段四：观测层闭环
]

# 每阶段迁移的流量分配策略
TRAFFIC_RAMP = {
"data_layer": {"dual_write": 100, "new_only": 100},
"inference_layer": {
"phase_1": 10, # 10% 流量走推理服务
"phase_2": 50, # 50% 流量走推理服务
"phase_3": 100, # 100% 流量走推理服务
},
"gateway_layer": {
"phase_1": 10, # 10% 流量走网关
"phase_2": 50, # 50% 流量走网关
"phase_3": 100, # 100% 流量走网关
},
}

# 每阶段的回滚触发条件
ROLLBACK_CONDITIONS = {
"data_layer": {
"new_db_error_rate": 0.01, # 新库错误率 > 1%
"dual_write_lag_ms": 500, # 双写延迟 > 500ms
},
"inference_layer": {
"inference_latency_p95": 5000, # 推理 P95 延迟 > 5s
"inference_error_rate": 0.05, # 推理错误率 > 5%
},
"gateway_layer": {
"gateway_error_rate": 0.02, # 网关错误率 > 2%
"gateway_latency_p95": 1000, # 网关 P95 延迟 > 1s
},
}

def __init__(self):
self._current_phase = 0
self._phase_status: Dict[str, str] = {}

def check_rollback(self, phase: str, metrics: dict) -> bool:
"""检查是否需要回滚"""
conditions = self.ROLLBACK_CONDITIONS.get(phase, {})
for metric_name, threshold in conditions.items():
actual = metrics.get(metric_name, 0)
if actual > threshold:
return True # 触发回滚
return False

def rollback(self, phase: str):
"""执行回滚 — 切回旧系统路由"""
print(f"[回滚] {phase}: 流量切回旧系统，"
f"新系统保留但不接收流量")
self._phase_status[phase] = "rolled_back"

def advance_phase(self):
"""推进到下一阶段 — 仅在当前阶段健康后推进"""
if self._current_phase < len(self.PHASES) – 1:
current = self.PHASES[self._current_phase]
if self._phase_status.get(current) == "healthy":
self._current_phase += 1
next_phase = self.PHASES[self._current_phase]
print(f"[推进] 从 {current} → {next_phase}")
else:
print(f"[等待] {current} 尚未完成健康检查")

五、总结

AI 生活化产品的架构演进遵循四阶段渐进拆分路径：数据层拆分（PostgreSQL/Qdrant/Redis 独立）→ 推理层拆分（LLM 调用独立服务+优先队列+批量合并）→ 网关层引入（弹性限速+智能路由+双层缓存）→ 观测层闭环（三层告警+全链路追踪+自动降级）。每个阶段的拆分依赖前阶段完成，避免并行拆分导致的交叉依赖问题。迁移策略遵循 Strangler Fig 模式：数据层用双写保障新旧并行，推理层用服务发现+本地回退保障推理不中断，网关层用 10%→50%→100% 流量逐步切换，观测层用宽松阈值避免初期误报。回滚保障的关键是每个阶段保留旧系统的完整功能，新系统异常时一键回退，数据不丢失服务不中断。架构演进不是一次性重构而是渐进式生长，每步拆分都让系统更模块化也更可观测，最终从单机原型成长为可水平扩展的分布式系统。