从 Demo 到上线:让 AI 真正跑在生产环境里的六堂课

2026-05-21 31 预计阅读时间:1 分钟
来源:infoq.com AI 摘要 原文链接

免责声明:本文为 AI 摘要整理,建议结合原文阅读。摘要可能省略上下文、版本差异或边界条件,不作为官方说明。

预计阅读时间:12 分钟

QCon AI Boston 2026 即将售罄,六场演讲直指一个让工程师头疼的现实——AI 在 demo 里表现惊艳,上了生产却处处翻车。模型延迟暴涨、输出格式飘忽、幻觉在关键业务里冒头、监控盲区让故障排查无从下手……这些问题不是理论探讨,而是每天在上线流程里绊脚的石头。

下面拆解几个最典型的"demo 到生产"断裂点,并给出可以直接拿来改造的工程方案。

断裂点一:输出不稳定,格式随时跑偏

Demo 里你精心调了 prompt,模型每次都乖乖返回 JSON。上了生产,用户输入五花八门,模型突然给你一段散文、一个缺字段的半成品,或者把 JSON 包在一层 markdown 代码块里。这不是偶发事故,是常态。

工程对策:结构化输出 + 校验兜底

不要只靠 prompt 级别的"请返回 JSON"。在代码层做双重保障:

import json
import re
from pydantic import BaseModel, ValidationError

class ExtractionResult(BaseModel):
    summary: str
    key_points: list[str]
    confidence: float

def parse_llm_output(raw: str) -> ExtractionResult:
    # 1. 尝试直接解析
    try:
        return ExtractionResult.model_validate_json(raw)
    except (json.JSONDecodeError, ValidationError):
        pass

    # 2. 从 markdown 代码块里提取
    code_block = re.search(r"```(?:json)?\s*\n(.*?)\n```", raw, re.DOTALL)
    if code_block:
        try:
            return ExtractionResult.model_validate_json(code_block.group(1))
        except (json.JSONDecodeError, ValidationError):
            pass

    # 3. 兜底:用另一个小模型做修复(或抛出业务异常)
    raise ValueError(f"无法解析模型输出: {raw[:200]}")

# 使用示例
raw_output = """```json
{"summary": "会议聚焦生产化AI", "key_points": ["监控", "兜底"], "confidence": 0.85}
```"""
result = parse_llm_output(raw_output)
print(result.summary)

关键思路:解析层永远比 prompt 层可靠。Pydantic 校验能立刻拦截缺字段、类型错、值越界的情况,比在 prompt 里反复强调"必须返回合法 JSON"靠谱得多。

断裂点二:延迟不可控,用户体验崩塌

Demo 时你一个人调 API,延迟 2 秒也无所谓。生产环境里并发上来,模型推理排队、token 生成慢、网络抖动叠加,P99 延迟可能飙到十几秒。用户等不了,前端超时,请求堆积,雪崩效应启动。

工程对策:超时熔断 + 分级降级

import time
from functools import wraps

def llm_call_with_fallback(
    primary_fn,          # 主模型调用
    fallback_fn,         # 降级方案(小模型 / 缓存 / 硬规则)
    timeout_sec: float = 5.0,
    max_retries: int = 1,
):
    """带超时和降级的 LLM 调用包装器"""

    for attempt in range(max_retries + 1):
        try:
            start = time.monotonic()
            result = primary_fn()
            elapsed = time.monotonic() - start
            if elapsed > timeout_sec:
                # 结果拿到了但太慢,记录指标,下次考虑直接降级
                print(f"[WARN] 主模型耗时 {elapsed:.1f}s,超过阈值 {timeout_sec}s")
            return result
        except Exception as e:
            print(f"[ERROR] 主模型调用失败 (attempt {attempt}): {e}")

    # 所有尝试失败或超时,走降级
    print("[FALLBACK] 启用降级方案")
    return fallback_fn()

# --- 实际使用 ---
def call_gpt4x(prompt: str) -> str:
    """调用大模型——可能慢、可能失败"""
    # 这里替换为你的实际 API 调用
    import openai
    resp = openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=300,
    )
    return resp.choices[0].message.content

def call_fast_model_or_cache(prompt: str) -> str:
    """降级:用小模型或命中缓存"""
    import openai
    resp = openai.chat.completions.create(
        model="gpt-4o-mini",  # 更快、更便宜
        messages=[{"role": "user", "content": prompt}],
        max_tokens=200,
    )
    return resp.choices[0].message.content

answer = llm_call_with_fallback(
    primary_fn=lambda: call_gpt4x("总结这段文本的核心观点"),
    fallback_fn=lambda: call_fast_model_or_cache("总结这段文本的核心观点"),
    timeout_sec=4.0,
)

这个模式的核心不是"大模型坏了用小模型",而是让系统在任何情况下都能在可接受时间内给出有用响应。降级方案可以是缓存命中、规则引擎、甚至一句"系统繁忙,稍后重试"——只要你的前端不会白屏卡死。

断裂点三:幻觉在关键业务里冒头,没有护栏

Demo 里模型编造一个无关紧要的事实,你一笑而过。生产里它给用户推荐了一个不存在的药品、编了一段虚假的法律条文、或者把内部 IP 地址写进了对外回复——这是事故。

工程对策:输出护栏 + 事实校验钩子

import re

def output_guardrails(text: str, context: dict) -> str:
    """在模型输出到达用户之前做最后一道检查"""

    blocked_patterns = [
        # 内部信息泄露
        re.compile(r"\b10\.\d+\.\d+\.\d+\b"),  # 内网 IP
        re.compile(r"\b[A-Z]{2,5}-\d{4,6}\b"),  # 内部工单号格式
        # 危险内容
        re.compile(r"(自杀|自残|伤害他人)", re.IGNORECASE),
    ]

    for pattern in blocked_patterns:
        if pattern.search(text):
            return "抱歉,我无法提供这方面的信息。请咨询专业机构。"

    # 事实校验:如果输出包含数值声明,要求与上下文对得上
    if context.get("source_numbers"):
        for num in re.findall(r"\d+\.?\d*", text):
            if float(num) not in context["source_numbers"]:
                # 数值不在来源里,标记可疑但不一定拦截
                text += f"\n[⚠️ 数值 {num} 未在提供的数据中找到,请核实]"

    return text

# 使用
raw_answer = "根据数据,公司营收为 8.7 亿元,服务器地址 10.0.1.44 可查详情。"
safe_answer = output_guardrails(
    raw_answer,
    context={"source_numbers": [8.7, 3.2, 1.5]},  # 8.7 在来源里,10.0.1.44 不该出现
)
print(safe_answer)
# 输出会拦截 IP 地址,并保留 8.7(因为它在 source_numbers 中)

护栏不是替代模型自身的安全训练,而是最后一道工程防线。模式匹配简单粗暴但有效,复杂场景可以接入另一个模型做二次审核——代价是延迟和成本,但关键业务值得。

断裂点四:监控盲区,出了问题全靠用户投诉

Demo 阶段你盯着终端输出,一切可见。生产环境里模型在后台静默运行,输出质量下滑、幻觉率上升、延迟渐增——直到用户投诉你才知道。传统 APM 监控 CPU、内存、QPS,对 LLM 的语义质量完全无感。

工程对策:语义指标采集 + 基础告警

# prometheus_llm_metrics.yaml — 给 LLM 服务加上可观测性
# 在你的推理服务里埋点,暴露这些指标

metrics:
  - name: llm_request_duration_seconds
    type: histogram
    labels: [model, endpoint, status]
    description: "每次 LLM 调用的耗时分布"

  - name: llm_output_token_count
    type: histogram
    labels: [model]
    description: "输出 token 数分布,异常飙升可能意味着幻觉"

  - name: llm_parse_failure_total
    type: counter
    labels: [model, error_type]
    description: "输出解析失败次数——格式跑偏的直接信号"

  - name: llm_guardrail_trigger_total
    type: counter
    labels: [model, rule_name]
    description: "护栏触发次数——内容安全问题的量化指标"

  - name: llm_fallback_trigger_total
    type: counter
    labels: [model, fallback_type]
    description: "降级触发次数——主模型不稳定的核心指标"

# Grafana 告警规则示例(用 yaml 表达逻辑)
alerts:
  - name: LLMParseFailureSpike
    expr: "rate(llm_parse_failure_total[5m]) > 0.1"
    message: "LLM 输出解析失败率超过 10%,格式稳定性可能出问题"
    severity: warning

  - name: LLMFallbackRateHigh
    expr: "rate(llm_fallback_trigger_total[10m]) / rate(llm_request_duration_seconds_count[10m]) > 0.05"
    message: "超过 5% 的请求触发降级,主模型可用性下降"
    severity: critical

在推理服务代码里埋一个简单的计数器就够了:

from prometheus_client import Counter, Histogram, start_http_server

parse_failures = Counter("llm_parse_failure_total", "Parse failures", ["model", "error_type"])
fallback_triggers = Counter("llm_fallback_trigger_total", "Fallback triggers", ["model", "fallback_type"])
request_duration = Histogram("llm_request_duration_seconds", "Request duration", ["model", "endpoint"])

# 在你的调用链里埋点
try:
    with request_duration.labels(model="gpt-4o", endpoint="/extract").time():
        result = call_gpt4x(prompt)
    parsed = parse_llm_output(result)
except ValueError as e:
    parse_failures.labels(model="gpt-4o", error_type="json_invalid").inc()
    fallback_triggers.labels(model="gpt-4o", fallback_type="mini_model").inc()
    parsed = call_fast_model_or_cache(prompt)

start_http_server(8000)  # Prometheus 来这里抓指标

不需要完美的语义评估系统,先把格式失败率、降级率、护栏触发率这三个数字亮出来,你就已经比"全靠用户投诉"强了一个等级。

上线前的自检清单

把上面四条浓缩成可操作的检查项:

检查项 达标标准 不达标的后果
输出解析有 Pydantic 校验 非法格式 100% 被拦截 上游模型格式飘了,下游直接崩溃
每个关键调用有超时 + 降级 P99 延迟有上限 用户白屏等待,请求堆积雪崩
输出经过护栏过滤 内部信息/危险内容被拦截 数据泄露、合规事故
Prometheus 有 LLM 语义指标 解析失败率/降级率可看可告警 问题静默恶化,靠投诉才发现

QCon AI Boston 这六场演讲的价值不在理论,而在一线工程师把这些问题摊开讲,讲他们踩过的坑和实际用的方案。如果你正在把 AI 从 demo 推向生产,这几条断裂点和对应的工程手段,比任何模型参数调优都更值得优先投入。


相关推荐