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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

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()

1

AI_MODEL=demo-model python app.py

1
2
3
4

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":"什么是上下文？"}'