AI 落地实战：从 POC 到生产的四道关卡与工程化实现

原创

IT大佬 jzit-top

发布于 2026-10-08 15:23:44

350

一、POC 成功不等于落地成功

AI 项目有一个高频现象：POC 阶段效果惊艳，上线后一地鸡毛。原因不是模型退化，而是 POC 和生产面对的是两套完全不同的约束。

维度	POC	生产
数据	精选样本	真实分布，含脏数据和边界
调用量	每天几十次	每天数万次
失败处理	人工兜底	必须自动降级
成本	不计	直接决定毛利
权限	单用户	多租户隔离
合规	忽略	审核、脱敏、留痕

POC 验证的是"技术上能不能做"，生产验证的是"工程上能不能稳"。中间隔着四道关卡：数据关、失败关、成本关、合规关。本文逐关拆解。

二、第一关：数据关

POC 用精选样本，生产用真实输入。真实输入的特点是：长度不可控、格式不规范、包含敏感信息、存在恶意构造。

第一关的目标是让输入可控。

import re
from dataclasses import dataclass

PII = [
    re.compile(r"\d{11}"),                    # 手机号
    re.compile(r"\d{17}[\dXx]"),              # 身份证
    re.compile(r"[\w.+-]+@[\w-]+\.[\w.]+"),   # 邮箱
]

@dataclass
class SanitizedInput:
    text: str
    truncated: bool
    masked_count: int

def sanitize(raw: str, max_len: int = 4000) -> SanitizedInput:
    if not isinstance(raw, str):
        raise ValueError("输入必须为字符串")

    text = raw.strip()
    truncated = len(text) > max_len
    if truncated:
        text = text[:max_len]

    masked = 0
    for pat in PII:
        text, n = pat.subn("[REDACTED]", text)
        masked += n

    return SanitizedInput(text=text, truncated=truncated,
                          masked_count=masked)

三个必须做的处理：长度截断（防止超长输入打爆 token 预算）、脱敏（个人信息不入日志、不入外部模型）、类型校验（非字符串输入直接拒绝）。

日志中永远写脱敏后的文本，不写原始输入。

三、第二关：失败关

POC 阶段失败可以人工重试。生产环境必须自动处理。失败关的目标是让系统在部分组件故障时仍能提供服务。

import asyncio, logging, time
from dataclasses import dataclass

log = logging.getLogger("resilience")

@dataclass
class CallResult:
    text: str
    source: str          # primary / fallback / cache
    latency: float
    degraded: bool = False

async def call_with_fallback(prompt: str, primary, fallback,
                             timeout: float = 8.0) -> CallResult:
    start = time.time()
    try:
        text = await asyncio.wait_for(primary(prompt), timeout)
        return CallResult(text, "primary", time.time() - start)
    except asyncio.TimeoutError:
        log.warning("primary timeout, falling back")
    except Exception as e:
        log.warning("primary failed: %s", e)

    try:
        text = await asyncio.wait_for(fallback(prompt), timeout * 2)
        return CallResult(text, "fallback", time.time() - start,
                          degraded=True)
    except Exception:
        log.exception("fallback also failed")
        raise RuntimeError("服务暂时不可用") from None

三个设计要点：

超时是硬约束。没有超时的调用，会拖垮整个服务。
降级要标记。degraded=True 让下游知道当前是降级状态，便于告警和统计。
降级不是无限兜底。备用也失败时，明确返回"服务不可用"，而不是返回空字符串。

四、第三关：成本关

POC 不算账，生产必须算。成本关的目标是让单次调用成本可控且可预测。

import hashlib
from dataclasses import dataclass, field
from collections import OrderedDict

@dataclass
class CostTracker:
    input_price_per_1k: float
    output_price_per_1k: float
    total_cost: float = 0.0
    calls: int = 0

    def record(self, in_tokens: int, out_tokens: int) -> float:
        cost = (in_tokens / 1000 * self.input_price_per_1k
                + out_tokens / 1000 * self.output_price_per_1k)
        self.total_cost += cost
        self.calls += 1
        return round(cost, 6)

    def avg_cost(self) -> float:
        return round(self.total_cost / self.calls, 6) if self.calls else 0.0

class LRUCache:
    def __init__(self, capacity: int = 5000):
        self.capacity = capacity
        self._data: OrderedDict[str, str] = OrderedDict()

    def key(self, prompt: str, params: str = "") -> str:
        raw = f"{prompt}|{params}"
        return hashlib.sha256(raw.encode()).hexdigest()

    def get(self, k: str) -> str | None:
        if k not in self._data:
            return None
        self._data.move_to_end(k)
        return self._data[k]

    def put(self, k: str, v: str) -> None:
        self._data[k] = v
        self._data.move_to_end(k)
        if len(self._data) > self.capacity:
            self._data.popitem(last=False)

缓存是降本最有效的手段。相同问题命中缓存，成本降为零。配合成本追踪，可以实时看到单次平均成本和总成本，超阈值告警。

五、第四关：合规关

生产环境必须处理合规问题。合规关的目标是让系统可审计、可追溯、可举证。

import uuid, logging, time

log = logging.getLogger("audit")

class AuditLogger:
    def __init__(self):
        self.records: list[dict] = []

    def log_call(self, user_id: str, scene: str,
                 model: str, prompt_fp: str,
                 tokens: int, cost: float,
                 audited: bool) -> str:
        trace_id = uuid.uuid4().hex[:12]
        self.records.append({
            "trace_id": trace_id,
            "ts": int(time.time()),
            "user": hashlib.sha256(user_id.encode()).hexdigest()[:12],
            "scene": scene,
            "model": model,
            "prompt_fp": prompt_fp,
            "tokens": tokens,
            "cost": cost,
            "audited": audited,
        })
        return trace_id

def audit_output(text: str, banned: set[str]) -> None:
    if any(w in text for w in banned):
        raise ValueError("输出未通过审核")

审计日志要回答五个问题：谁调的、什么时候、用了哪个模型和提示词版本、花了多少、是否通过审核。 用户标识要哈希，不存明文。

六、四关联动：完整调用链

async def serve(raw_input: str, user_id: str, ctx) -> dict:
    sanitized = sanitize(raw_input)

    cache_key = ctx.cache.key(sanitized.text, ctx.params_fp)
    if cached := ctx.cache.get(cache_key):
        return {"text": cached, "source": "cache"}

    result = await call_with_fallback(
        sanitized.text, ctx.primary, ctx.fallback
    )

    ctx.audit.log_call(
        user_id, ctx.scene, ctx.model, ctx.prompt_fp,
        ctx.last_tokens, ctx.last_cost, audited=True,
    )
    audit_output(result.text, ctx.banned)
    ctx.cache.put(cache_key, result.text)

    return {
        "text": result.text,
        "source": result.source,
        "degraded": result.degraded,
    }

四关在这条链路里各司其职：数据关做清洗，成本关做缓存，失败关做降级，合规关做审计。

七、总结

AI 落地的难点不在模型，而在四道关卡：

关卡	目标	关键动作
数据关	输入可控	截断、脱敏、类型校验
失败关	系统可降级	超时、主备切换、降级标记
成本关	成本可预测	缓存、成本追踪、阈值告警
合规关	行为可审计	trace、脱敏、审核、留痕

代码可以简单，但截断、超时、缓存、审计这四件事不能省。POC 验证的是技术可行性，生产验证的是工程可控性——两者的距离，就是这四道关卡。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

AI 双师课堂