端侧 AI 推理迁移：用 cgroups 隔离资源并保留 RPC 降级

将原本运行在云端的 AI 模型下沉至终端设备（如嵌入式 Linux、边缘网关或客户端），是降低网络传输延迟与保护数据隐私的常见选型。但在存量系统升级过程中，若未经评估直接把云端 API 或既有规则引擎全盘替换为端侧小模型，在高并发或长上下文场景下，容易触发内存超限（OOM）、CPU 资源抢占，进而影响主业务进程的实时响应。

端侧 AI 推理的部署除了模型格式转换（如 GGUF、ONNX 或 TensorRT-LLM）和镜像打包，还需要在操作系统层面规划资源隔离、流量限流与熔断降级。

---

存量系统迁移中的核心资源安全冲突

存量系统往往已经围绕既有负载调过线程、内存和权限策略。接入端侧推理引擎后，需要重新检查以下三类资源与安全边界：

1. 内存消耗与 OOM 风险

端侧推理会占用模型权重、激活值和 KV Cache 等内存。具体分配方式取决于运行时、硬件和驱动；长输入或并发请求仍可能推高内存使用，并触发 OOM 处置，影响关键业务进程。

2. 实时性调度干扰

小模型推理属于典型的 CPU/NPU 密集型任务。若推理线程与主业务进程共享相同的 CPU 优先级与调度策略，高负载推理会抢占 CPU 核心，使原本毫秒级响应的系统调用（Syscall）出现阻塞。

3. 内存隔离与权限控制

部分端侧推理库为追求执行效率，大量使用 Unsafe C/C++ 代码与共享内存机制。若将推理逻辑直接嵌入主业务进程空间，一旦推理库发生缓冲区溢出，可能导致主进程的内存上下文被非法访问。

---

存量系统的四阶段平滑迁移路径

为保障操作系统层面的业务连续性，迁移过程应遵循“旁路观察 ➔ 双跑比对 ➔ 受限主导 ➔ 全量切换”的分阶段推进策略。

flowchart TD
    subgraph 传统主进程 [安全隔离区]
        A[业务请求] --> B[传统规则引擎/云端 API]
        B --> C[响应返回]
    end

    subgraph 端侧 AI 独立子进程 [受限沙箱区 (cgroups)]
        D[阴影流量 / 影子推理]
        E[端侧 AI 推理引擎]
        D --> E
    end

    A -- Phase 1: 异步旁路复制 --> D
    B -- Phase 2: 结果一致性校验器 --> F{差异分析}
    
    subgraph 降级熔断防线
    G[主进程限流 / 门限闸门] -->|超过团队设定的时延或资源预算| B
        G -->|系统负载正常| E
    end

阶段一：旁路影子运行 (Shadow Phase)

保持主业务逻辑 100% 独立运行。在后台启动独立沙箱进程运行端侧 AI 推理，主进程通过 Unix Domain Socket 等 IPC 机制将请求副本异步发送至推理进程。该阶段仅记录推理耗时、内存峰值与输出质量，不参与实际业务决策。

阶段二：双跑比对与一致性校验 (Dual-Run Phase)

主进程同时触发旧有规则引擎与端侧 AI 引擎，以旧引擎结果作为最终返回值。后台服务比对两者的输出一致性与响应时延，当端侧 AI 置信度低于预设门限时，自动记录样本以备离线分析。

阶段三：受限主导与自动降级 (Active-Degrade Phase)

端侧 AI 接入主流程担当核心响应方。在主进程中建立硬限制机制：当推理耗时超过超时门限（如 200ms），或系统 CPU/内存使用率达到预警阈值时，自动降级回退至轻量级规则引擎。

阶段四：全量切换与旧链路收容 (Full Cutover Phase)

在经过多轮长周期稳定性压测且性能与安全指标全面达标后，切断旧有旁路，清理残留的影子代码与临时比对逻辑。

---

Linux cgroups v2 资源隔离示例

端侧推理进程通常应配置资源边界。cgroups v2 可限制 CPU 和内存，taskset 可限制进程可运行的 CPU 集；两者是不同机制，应按部署环境分别验证。

以下 Shell 脚本展示了如何构建端侧 AI 进程的资源隔离沙箱：

#!/usr/bin/env bash
set -euo pipefail

# 定义 cgroup 沙箱名称
CGROUP_NAME="ai_inference_sandbox"
CGROUP_PATH="/sys/fs/cgroup/${CGROUP_NAME}"

echo "==> 创建 cgroups v2 隔离组: ${CGROUP_NAME}"
if [ ! -d "${CGROUP_PATH}" ]; then
  mkdir -p "${CGROUP_PATH}"
fi

# 1. 限制 CPU 使用上限（配额设置为上限 2 个 CPU 核心: 200000 / 100000）
echo "200000 100000" > "${CGROUP_PATH}/cpu.max"

# 2. 限制内存使用：硬上限 1.5GB，超限在沙箱内触发 OOM，不影响宿主主进程
echo "1610612736" > "${CGROUP_PATH}/memory.max"
# 设置内存高位水位预警 (1.2GB)
echo "1288490188" > "${CGROUP_PATH}/memory.high"

# 3. 禁用 Swap 交换区擦写
echo "0" > "${CGROUP_PATH}/memory.swap.max"

echo "==> cgroups 规则配置完成："
echo "    CPU Limit: 2 Cores"
echo "    Memory Max: 1.5 GB"

# 4. 启动端侧 AI 推理服务，并将 PID 加入统一层级的 cgroup
# 假设主业务占用 Core 0,1
echo "==> 启动推理服务子进程..."
taskset -c 2,3 \
    /usr/bin/python3 /opt/ai_edge/inference_service.py \
    --model /opt/models/qwen_1.8b_q4.gguf \
    --port 9090 &

INFERENCE_PID=$!
echo "${INFERENCE_PID}" > "${CGROUP_PATH}/cgroup.procs"
echo "==> 推理进程已启动, PID: ${INFERENCE_PID}"

上述配置展示了资源边界的基本写法。实际部署前还需确认父 cgroup 已启用相应 controller、进程具备写入权限，并在目标发行版上验证 OOM 行为和主业务的影响范围。

---

客户端防御性 RPC 降级 Client 实现

主业务进程调取端侧推理服务时，需配置超时控制与熔断保护，避免主线程产生阻塞。

以下为 Python asyncio 实现的防御性客户端调用模块：

import asyncio
import logging

logger = logging.getLogger("SystemBridge")

class SafeEdgeInferenceClient:
    def __init__(self, socket_path: str, fallback_engine):
        self.socket_path = socket_path
        self.fallback_engine = fallback_engine
        self.consecutive_failures = 0
        self.circuit_open = False
        self.max_failures = 3

    async def predict_with_fallback(self, payload: dict, timeout_sec: float = 0.2) -> dict:
        """
        带熔断与降级机制的推理调用入口
        """
        if self.circuit_open:
            logger.warning("[Degrade] Circuit breaker OPEN. Invoking fallback engine.")
            return self.fallback_engine.execute(payload)

        try:
            # 配置严格的响应超时门限
            result = await asyncio.wait_for(
                self._call_ipc_inference(payload), 
                timeout=timeout_sec
            )
            # 调用成功重置连续失败计数
            self.consecutive_failures = 0
            return result

        except Exception as e:
            self.consecutive_failures += 1
            logger.error(f"[IPC Failure] Exception: {type(e).__name__}, Count: {self.consecutive_failures}")

            # 触发熔断保护门限
            if self.consecutive_failures >= self.max_failures:
                self.circuit_open = True
                logger.error("[Circuit Breaker] Tripped! Switching to fallback engine.")

            # 无缝降级至经典规则引擎
            return self.fallback_engine.execute(payload)

    async def _call_ipc_inference(self, payload: dict) -> dict:
        reader, writer = await asyncio.open_unix_connection(self.socket_path)
        writer.write(str(payload).encode('utf-8'))
        await writer.drain()

        data = await reader.read(4096)
        writer.close()
        await writer.wait_closed()
        return {"status": "success", "data": data.decode('utf-8')}

端侧推理上线前，重点应放在资源边界、超时和降级路径上。它们需要结合设备规格、模型和真实请求分布压测，而不是照搬示例中的数值。