AI 应用从本地原型到生产部署：先定义可运行的服务边界

前面的文章已经介绍了模型调用、错误处理、评估、日志和性能优化。到了部署阶段，难点通常不在“把代码放到服务器”，而在于让程序能够被稳定启动、配置、检查和停止。本篇只聚焦一个核心知识点：如何为 AI 应用建立最小的生产部署边界。示例使用 Python 标准库模拟模型调用，不需要真实密钥，也不依赖 Web 框架；它可以直接运行，帮助我们先理解部署契约，再替换为实际的模型客户端。

本地能运行不等于可以部署

本地原型往往把配置写在代码里，调用失败时直接打印异常，程序退出也没有明确规则。这样的程序可以验证想法，却很难交给进程管理器或容器运行。生产部署至少要把以下职责说清楚：

配置从哪里来：模型名、超时和密钥通过环境变量注入，代码只提供安全的非敏感默认值。
进程如何工作：启动后持续运行，收到停止信号时完成收尾，不把一个请求的失败变成整个服务的无声崩溃。
是否健康：健康检查要能区分“进程还活着”和“已经可以接收请求”。
如何观察：日志输出到标准输出，包含请求结果和耗时，交给平台统一采集。

这四项是部署边界，而不是某个云平台的专属功能。容器、虚拟机或进程管理器都可以据此启动同一个程序。真正的模型 SDK 只负责调用模型，部署层仍然负责进程、配置和运行状态。

一个最小的可运行服务

下面的程序提供一个简单的 HTTP 服务。/healthz 返回进程状态，/ask 接收 JSON 中的 question，并用本地函数模拟模型调用。为了让示例可验证，模拟函数会把问题转换为固定格式；接入真实服务时，只替换 call_model，不要把密钥写进源码。

import json
import logging
import os
import signal
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
HOST = os.getenv("AI_HOST", "127.0.0.1")
PORT = int(os.getenv("AI_PORT", "8080"))
MODEL_NAME = os.getenv("AI_MODEL", "demo-model")


def call_model(question: str) -> str:
    """用确定性结果模拟一次模型调用。"""
    return f"[{MODEL_NAME}] 已收到：{question}"


class Handler(BaseHTTPRequestHandler):
    def send_json(self, status: int, payload: dict) -> None:
        body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self) -> None:
        if self.path == "/healthz":
            self.send_json(200, {"status": "ok"})
        else:
            self.send_json(404, {"error": "not found"})

    def do_POST(self) -> None:
        if self.path != "/ask":
            self.send_json(404, {"error": "not found"})
            return

        started = time.perf_counter()
        try:
            length = int(self.headers.get("Content-Length", "0"))
            data = json.loads(self.rfile.read(length))
            question = data.get("question")
            if not isinstance(question, str) or not question.strip():
                raise ValueError("question 必须是非空字符串")
            answer = call_model(question.strip())
            self.send_json(200, {"answer": answer})
        except (ValueError, json.JSONDecodeError) as exc:
            self.send_json(400, {"error": str(exc)})
        except Exception:
            logging.exception("request failed")
            self.send_json(502, {"error": "model request failed"})
        finally:
            elapsed = time.perf_counter() - started
            logging.info("path=%s elapsed=%.3fs", self.path, elapsed)

    def log_message(self, format: str, *args: object) -> None:
        logging.info("http " + format, *args)


def main() -> None:
    server = ThreadingHTTPServer((HOST, PORT), Handler)

    def stop(signum: int, _frame: object) -> None:
        logging.info("received signal=%s, shutting down", signum)
        server.shutdown()

    signal.signal(signal.SIGTERM, stop)
    signal.signal(signal.SIGINT, stop)
    logging.info("listening on http://%s:%s", HOST, PORT)
    try:
        server.serve_forever()
    finally:
        server.server_close()
        logging.info("server stopped")


if __name__ == "__main__":
    main()

保存为 app.py 后运行：

1	AI_MODEL=demo-model python app.py

另开终端验证：

curl -i http://127.0.0.1:8080/healthz
curl -i -X POST http://127.0.0.1:8080/ask \
  -H 'Content-Type: application/json' \
  -d '{"question":"什么是上下文？"}'

程序收到 Ctrl-C 或终止信号后会退出 serve_forever，执行 server_close。这里的健康检查只表示进程已经能响应；真实项目还可以在启动阶段检查必要配置，并增加依赖服务检查，但不要让健康检查本身触发一次昂贵的模型调用。

接入真实模型时，边界应该放在哪里

把 call_model 替换为 SDK 调用时，至少保留三层边界。第一层是配置边界：使用类似 os.environ["PROVIDER_API_KEY"] 的方式读取密钥，部署平台负责注入，日志中绝不打印密钥和完整用户输入。第二层是请求边界：为网络调用设置连接和读取超时，限制输入长度，并将供应商异常转换为应用自己的错误类型。第三层是输出边界：即使模型返回了内容，也要经过解析、校验和必要的安全过滤后再交给调用方。

不要把“服务启动成功”理解成“模型调用一定成功”。启动时可以校验密钥变量是否存在，但网络、配额、上游限流和模型临时故障仍然可能在运行中发生。因此 /healthz 与请求错误处理应当分开：健康检查反映服务进程状态，请求处理则按超时、重试和降级策略返回结果。

部署清单：从脚本到进程

实际部署时，可以按下面顺序检查：

固定运行方式：明确 Python 版本、启动命令和工作目录，依赖文件锁定版本，避免“在我的机器上可以”。
配置外置：区分非敏感配置和密钥；密钥使用环境变量或平台的密钥管理服务，不提交到 Git。
限制资源：为并发数、请求体大小、单次输入长度和超时设上限。AI 请求通常比普通接口更慢，不能默认无限等待。
标准输出日志：至少记录请求类型、耗时、状态和可关联的请求 ID；不要记录原始密钥、完整隐私文本或模型内部思维内容。
可停止：处理 SIGTERM，停止接收新任务并等待正在执行的任务收尾。若调用可能很久，还需要设置最大等待时间。
健康检查：平台应能通过固定路径判断进程是否响应；重启策略要避免故障时无限快速重启。
分离后台任务：批量摘要、索引和定时处理不要阻塞在线请求，必要时使用独立 worker 和队列。

初次上线不必一开始就引入复杂平台。先让同一份程序在本地、测试环境和生产环境通过环境变量获得不同配置，再由进程管理器或容器负责拉起、停止和重启。部署工具可以变化，但应用的输入、输出和退出行为应保持稳定。

常见问题

为什么不把密钥直接写进配置文件？ 配置文件很容易被提交、复制或进入镜像层。环境变量也不是万能的，但配合平台密钥管理和最小权限，泄露范围更容易控制。

健康检查返回 200 就代表模型服务正常吗？ 不代表。它通常只验证当前进程能响应。若把上游模型也纳入检查，可能造成额外费用和级联故障，应根据系统目标拆成轻量存活检查和受控的依赖检查。

为什么一个请求失败不能让进程退出？ 临时网络错误只影响当前请求，进程退出会把一个局部问题扩大成全站不可用。只有无法恢复的启动配置错误或进程级故障，才应交给外部管理器重启。

ThreadingHTTPServer 能直接承载生产流量吗？ 这里的重点是展示部署边界，不是推荐某个 Web 服务器。正式环境应选择经过验证的服务栈，并在前面配置 TLS、认证、限流和网关；这些都不应由这个最小示例承担。

小结

从原型走向生产，第一步不是增加更多模型能力，而是定义稳定的运行契约：配置外置、请求有边界、日志可观察、健康状态可检查、进程能够优雅停止。本文的标准库示例刻意省略了真实供应商细节，因此可以先验证启动、健康检查、错误响应和信号处理。接入具体模型后，只需把模型调用放回明确的函数边界，并继续保留超时、校验和密钥管理。这样，AI 能力只是服务中的一个可替换部件，而不是部署系统的全部。

搜索时光笔记