部署、API 与模型网关

New-Item -ItemType Directory -Force l8-deployment
cd l8-deployment

mkdir -p l8-deployment
cd l8-deployment

"""用请求字典演示 API 网关的处理顺序与错误映射。"""
READY = True
LIMIT = 2

def local_model(text, calls):
    # 本地固定答案适配器：函数执行时才记录调用，不接远端模型。
    calls.append("local-model")
    return {"answer": "报销上限200元", "model": "local-model"}

def gateway(request, *, ready=True, active=0, downstream_ms=200):
    path = request.get("path")
    # 健康探针独立于业务请求；存活表示进程能回答，就绪表示可接业务。
    if path == "/live":
        return 200, {"live": True}, []
    if path == "/ready":
        return (200 if ready else 503), {"ready": ready}, []
    if path != "/ask":
        return 404, {"error": "route"}, []
    # 身份来自网关可信验证结果，示例不把模型输出当身份。
    if request.get("principal") != "alice":
        return 401, {"error": "identity"}, []
    text = request.get("text")
    if not isinstance(text, str) or not text.strip() or len(text) > 100:
        return 400, {"error": "input"}, []
    if not ready:
        return 503, {"error": "not_ready"}, []
    if active >= LIMIT:
        return 429, {"error": "concurrency"}, []
    # 调用次数可观测，超时后不自动重试，避免扩大下游压力。
    calls = []
    body = local_model(text, calls)
    if downstream_ms > 1000:
        return 504, {"error": "deadline"}, calls
    return 200, body, calls

def main():
    samples = [
        ("存活", {"path": "/live"}, {}),
        ("就绪", {"path": "/ready"}, {}),
        ("问答", {"path": "/ask", "principal": "alice", "text": "报销上限?"}, {}),
        ("缺少身份", {"path": "/ask", "text": "报销上限?"}, {}),
        ("并发已满", {"path": "/ask", "principal": "alice", "text": "报销上限?"}, {"active": 2}),
        ("下游超时", {"path": "/ask", "principal": "alice", "text": "报销上限?"}, {"downstream_ms": 1200}),
    ]
    for label, request, options in samples:
        status, body, calls = gateway(request, ready=READY, **options)
        print(f"{label}: {status} / {body} / 调用{len(calls)}")

if __name__ == "__main__":
    main()

"""检查在调用前拒绝、阈值边界与健康探针的不同语义。"""
import unittest
from gateway import gateway, local_model

ASK = {"path": "/ask", "principal": "alice", "text": "报销上限?"}

class GatewayTests(unittest.TestCase):
    def test_adapter_records_execution(self):
        calls = []
        body = local_model("报销上限?", calls)
        self.assertEqual(calls, ["local-model"])
        self.assertEqual(body["answer"], "报销上限200元")

    def test_normal_and_invalid(self):
        self.assertEqual(gateway(ASK)[0], 200)
        self.assertEqual(gateway({**ASK, "text": " "}),
                         (400, {"error": "input"}, []))

    def test_identity_before_call(self):
        self.assertEqual(gateway({**ASK, "principal": "bob"})[2], [])

    def test_readiness_and_liveness(self):
        self.assertEqual(gateway({"path": "/live"}, ready=False)[0], 200)
        self.assertEqual(gateway({"path": "/ready"}, ready=False)[0], 503)
        self.assertEqual(gateway(ASK, ready=False)[2], [])

    def test_slot_boundary(self):
        self.assertEqual(gateway(ASK, active=1)[0], 200)
        self.assertEqual(gateway(ASK, active=2)[0], 429)

    def test_deadline_boundary(self):
        self.assertEqual(gateway(ASK, downstream_ms=1000)[0], 200)
        self.assertEqual(gateway(ASK, downstream_ms=1001)[0], 504)

    def test_unknown_route(self):
        self.assertEqual(gateway({"path": "/missing"})[0], 404)

if __name__ == "__main__":
    unittest.main()

ls
python -X utf8 gateway.py
python -X utf8 -m unittest -v test_gateway.py

存活: 200 / {'live': True} / 调用0
就绪: 200 / {'ready': True} / 调用0
问答: 200 / {'answer': '报销上限200元', 'model': 'local-model'} / 调用1
缺少身份: 401 / {'error': 'identity'} / 调用0
并发已满: 429 / {'error': 'concurrency'} / 调用0
下游超时: 504 / {'error': 'deadline'} / 调用1

python -X utf8 gateway.py