从 ChatGPT 原型到生产系统：迁移避坑指南

原型 vs 生产：到底差在哪

在 ChatGPT 界面里跑通一个 demo，和生产环境服务千万用户，中间隔着一条鸿沟。原型阶段你不需要关心的事情，在生产中每一个都会变成事故：

维度	原型阶段	生产阶段
延迟	手动等待，几秒可接受	P99 < 3s，超时降级
可用性	失败就重试	99.9% SLA，多活容灾
成本	不关心	每月 API 账单可预测
安全	Prompt 直接硬编码	密钥管理、输入消毒、输出过滤
可观测性	print 大法	结构化日志、指标、链路追踪
一致性	每次回答不同	温度控制、缓存、确定性输出

核心认知：原型验证的是"能不能做"，生产回答的是"能不能稳定做、能不能赚钱做、能不能安全做"。

API 选型：不只是选个模型

模型选择决策树

def choose_model(task: str, latency_req: float, budget: float) -> str:
    if task in ["简单分类", "抽取"] and latency_req < 0.5:
        return "gpt-4o-mini"  # 快且便宜
    if task in ["复杂推理", "代码生成"] and budget > 0.05:
        return "gpt-4o"       # 能力优先
    if task in ["长文档总结"] and budget < 0.02:
        return "gpt-4o-mini"  # 上下文够长，成本可控
    return "gpt-4o-mini"      # 默认保守

成本估算公式

单次调用成本 = (input_tokens × input_price) + (output_tokens × output_price)

月度预算 = 日均调用量 × 平均(input_tokens + output_tokens) × 30 × 单价

# GPT-4o (2026 pricing)
# Input:  $2.50 / 1M tokens
# Output: $10.00 / 1M tokens
# 一个典型客服对话: ~500 input + ~200 output
# 单次成本 ≈ $0.00325
# 日均 10000 次 → 月成本 ≈ $975

多模型路由策略

生产环境不要只用一个模型。根据请求复杂度路由到不同模型：

import hashlib

class ModelRouter:
    def __init__(self):
        self.routes = {
            "simple": {"model": "gpt-4o-mini", "max_tokens": 500},
            "standard": {"model": "gpt-4o-mini", "max_tokens": 1000},
            "complex": {"model": "gpt-4o", "max_tokens": 2000},
        }

    def classify(self, prompt: str) -> str:
        # 基于规则快速分类，不要用 LLM 分类 LLM 请求
        if len(prompt) < 100 and "?" in prompt:
            return "simple"
        if any(kw in prompt for kw in ["分析", "设计", "推理", "代码"]):
            return "complex"
        return "standard"

    def route(self, prompt: str):
        tier = self.classify(prompt)
        return self.routes[tier]

Prompt 工程化：从对话到系统

Prompt 的三层架构

┌─────────────────────────────────┐
│ System Prompt (固定，版本管理)    │
│  - 角色定义、全局规则、安全约束    │
├─────────────────────────────────┤
│ Task Template (模板，参数化)      │
│  - 任务指令、输出格式、Few-shot    │
├─────────────────────────────────┤
│ User Input (动态，需消毒)         │
│  - 用户实际输入，经清洗和截断      │
└─────────────────────────────────┘

Prompt 模板示例

from string import Template

SYSTEM_PROMPT = """你是一个客服助手。遵守以下规则：
1. 只回答与产品相关的问题
2. 涉及隐私信息时引导用户联系人工客服
3. 如果不确定，说"我需要确认一下"，不要编造
4. 回答不超过 200 字
"""

TASK_TEMPLATE = Template("""用户问题: $user_question
用户上下文: $context
请按照以下格式回答:
{"answer": "你的回答", "confidence": 0.0-1.0, "need_human": bool}
""")

def build_prompt(user_question: str, context: dict) -> list:
    # 输入消毒：移除可能的 Prompt 注入
    sanitized = user_question.replace("忽略以上指令", "").replace("ignore previous", "")
    if len(sanitized) > 2000:
        sanitized = sanitized[:2000] + "...[截断]"

    return [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": TASK_TEMPLATE.substitute(
            user_question=sanitized,
            context=str(context)
        )}
    ]

错误处理：生产系统的生命线

错误分类与处理策略

import time
from enum import Enum

class ErrorType(Enum):
    RATE_LIMIT = "rate_limit"        # 429: 退避重试
    SERVER_ERROR = "server_error"    # 5xx: 切换备用
    CONTEXT_LENGTH = "context_len"   # 400: 截断重试
    CONTENT_FILTER = "content_filter" # 400: 降级回复
    TIMEOUT = "timeout"              # 自定义: 降级或重试

class LLMErrorHandler:
    def __init__(self):
        self.retry_config = {
            ErrorType.RATE_LIMIT: {"max_retries": 3, "backoff": "exponential"},
            ErrorType.SERVER_ERROR: {"max_retries": 2, "backoff": "linear"},
            ErrorType.TIMEOUT: {"max_retries": 1, "backoff": "fixed"},
        }

    async def handle(self, error, call_fn, *args, **kwargs):
        etype = self._classify(error)
        config = self.retry_config.get(etype)

        if not config:
            return self._fallback_response(etype)

        for attempt in range(config["max_retries"]):
            wait = self._calc_backoff(config["backoff"], attempt)
            await asyncio.sleep(wait)
            try:
                return await call_fn(*args, **kwargs)
            except Exception as e:
                if attempt == config["max_retries"] - 1:
                    return self._fallback_response(etype)

    def _calc_backoff(self, strategy, attempt):
        if strategy == "exponential":
            return min(2 ** attempt, 60)  # 上限 60s
        if strategy == "linear":
            return (attempt + 1) * 2
        return 5  # fixed

    def _fallback_response(self, etype):
        return {
            "answer": "服务暂时繁忙，请稍后重试",
            "confidence": 0.0,
            "need_human": True,
            "error": etype.value,
        }

熔断器模式

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failures = 0
        self.threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.last_failure_time = None
        self.state = "closed"  # closed, open, half-open

    def can_call(self):
        if self.state == "closed":
            return True
        if self.state == "open":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "half-open"
                return True
            return False
        return True  # half-open: 允许一次试探

    def record_success(self):
        self.failures = 0
        self.state = "closed"

    def record_failure(self):
        self.failures += 1
        self.last_failure_time = time.time()
        if self.failures >= self.threshold:
            self.state = "open"

测试策略：测试不可确定性系统

LLM 输出是非确定性的，传统单元测试不适用。需要分层测试：

测试层	方法	覆盖目标
单元测试	Mock LLM 响应，测试解析逻辑	解析、消毒、路由
集成测试	真实 API，固定 seed	端到端流程
回归测试	快照 + 语义相似度	防止退化
评估测试	LLM-as-Judge + 人工标注	质量基线

# 回归测试示例
import pytest
from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')

class TestRegression:
    @pytest.fixture
    def baseline(self):
        return load_baseline_results("v1.2.0")

    def test_answer_similarity(self, baseline):
        test_cases = baseline["cases"]
        for case in test_cases:
            current = llm_call(case["input"])
            sim = cosine_similarity(
                model.encode(current),
                model.encode(case["expected"])
            )
            assert sim > 0.85, f"回归: {case['name']} 相似度 {sim:.2f}"

团队协作：Prompt 不是一个人的事

角色分工

Prompt 工程师：编写和优化 Prompt，维护版本
后端工程师：API 集成、错误处理、缓存
QA：评估数据集、回归测试、人工标注
产品经理：定义成功标准、审查输出质量

Prompt 评审流程

1. Prompt 工程师提交 PR (Prompt 变更 + 评估结果)
2. CI 自动运行评估套件 (200+ 测试用例)
3. QA 人工抽检 (20 条)
4. 产品经理审查输出质量
5. 合并后灰度发布 (10% → 50% → 100%)

常见陷阱清单

硬编码 API Key → 用密钥管理服务（AWS Secrets Manager / HashiCorp Vault）
不做输入截断 → 用户输入无上限，必须截断到模型上下文限制的 50% 以内
忽略 token 计数 → 用 tiktoken 在调用前估算，避免超限
缓存只缓存成功 → 适当缓存部分失败，避免雪崩
日志记录完整 Prompt → 可能泄露用户隐私，脱敏后再记录
没有降级方案 → LLM 不可用时必须有规则引擎兜底
忽视流式响应 → 用户等待感强，优先用 streaming
温度参数一刀切 → 不同任务用不同 temperature（客服 0.3，创意 0.9）

迁移检查清单

API Key 已迁移到密钥管理服务
实现了多模型路由策略
错误处理覆盖所有错误类型
熔断器和退避重试已上线
日志脱敏 + 结构化
成本监控告警已配置
评估流水线已集成 CI
降级方案已验证
Prompt 版本管理已建立
灰度发布机制已就绪

总结

从原型到生产不是"部署一下"那么简单。核心工作在于：把 ChatGPT 界面里的"对话"变成一个有错误处理、有成本控制、有质量保障、有降级方案的工程系统。把 Prompt 当代码管理，把 LLM 当不稳定依赖处理，把每次迭代当科学实验来做。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。

原型 vs 生产：到底差在哪#

API 选型：不只是选个模型#

模型选择决策树#

成本估算公式#

多模型路由策略#

Prompt 工程化：从对话到系统#

Prompt 的三层架构#

Prompt 模板示例#

错误处理：生产系统的生命线#

错误分类与处理策略#

熔断器模式#

测试策略：测试不可确定性系统#

团队协作：Prompt 不是一个人的事#

角色分工#

Prompt 评审流程#

常见陷阱清单#

迁移检查清单#

总结#

加入讨论#

📌 相关推荐