本地推理与模型服务：先看清 HTTP 契约，再决定是否自己部署

"""启动本机接口，用固定响应练习请求发送与回答检查。"""

import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer


def inspect_response(payload):
    """从choices列表中取出回答，检查message及正文的格式。"""
    if not isinstance(payload, dict):
        return "invalid_response"
    choices = payload.get("choices")
    if not isinstance(choices, list) or not choices:
        return "invalid_response"
    first = choices[0]
    if not isinstance(first, dict) or not isinstance(first.get("message"), dict):
        return "invalid_response"
    content = first["message"].get("content")
    if not isinstance(content, str) or not content.strip():
        return "invalid_response"
    return "candidate_only"


class Handler(BaseHTTPRequestHandler):
    def log_message(self, *_args):
        pass  # Avoid echoing potentially sensitive request data in this fixture.

    def send_json(self, status, body):
        data = json.dumps(body, 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(data)))
        self.end_headers()
        self.wfile.write(data)

    def do_POST(self):
        if self.path != "/v1/chat/completions":
            self.send_json(404, {"error": "unknown_path"})
            return
        try:
            length = int(self.headers.get("Content-Length", "0"))
            if length < 1 or length > 16384:
                raise ValueError("request_size")
            body = json.loads(self.rfile.read(length).decode("utf-8"))
        except (ValueError, UnicodeError, json.JSONDecodeError):
            self.send_json(400, {"error": "invalid_json_or_size"})
            return
        if not isinstance(body, dict) or body.get("model") != "fixture-model":
            self.send_json(404, {"error": "model_not_found"})
            return
        messages = body.get("messages")
        if not isinstance(messages, list) or not messages:
            self.send_json(400, {"error": "messages_required"})
            return
        if any(not isinstance(m, dict) or m.get("role") not in ("user", "system")
               or not isinstance(m.get("content"), str) for m in messages):
            self.send_json(400, {"error": "invalid_message"})
            return
        self.send_json(200, {
            "model": "fixture-model",
            "choices": [{"index": 0, "message": {"role": "assistant", "content": "需要人工核实"},
                         "finish_reason": "stop"}],
            "usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0},
            "fixture": True,
        })


if __name__ == "__main__":
    server = ThreadingHTTPServer(("127.0.0.1", 8765), Handler)
    print("offline fixture listening at http://127.0.0.1:8765")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()

# 核对本地推理与模型服务的正常结果和错误分支；断言失败时会显示对应的测试。
import json
import threading
import unittest
from http.server import ThreadingHTTPServer
from urllib.error import HTTPError
from urllib.request import Request, urlopen

from service_contract import Handler, inspect_response


class ContractTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
        cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
        cls.thread.start()
        cls.url = "http://127.0.0.1:%d/v1/chat/completions" % cls.server.server_port

    @classmethod
    def tearDownClass(cls):
        cls.server.shutdown()
        cls.server.server_close()
        cls.thread.join(timeout=3)

    def post(self, body):
        data = json.dumps(body).encode("utf-8")
        request = Request(self.url, data=data, headers={"Content-Type": "application/json"})
        with urlopen(request, timeout=2) as response:
            return response.status, json.load(response)

    def test_valid_response_is_only_a_candidate(self):
        status, body = self.post({"model": "fixture-model", "messages": [{"role": "user", "content": "hi"}]})
        self.assertEqual(status, 200)
        self.assertEqual(inspect_response(body), "candidate_only")

    def test_missing_messages_rejected(self):
        with self.assertRaises(HTTPError) as error:
            self.post({"model": "fixture-model", "messages": []})
        self.assertEqual(error.exception.code, 400)

    def test_wrong_model_rejected(self):
        with self.assertRaises(HTTPError) as error:
            self.post({"model": "other", "messages": [{"role": "user", "content": "hi"}]})
        self.assertEqual(error.exception.code, 404)

    def test_missing_choices_not_accepted(self):
        self.assertEqual(inspect_response({"choices": []}), "invalid_response")

    def test_empty_content_not_accepted(self):
        self.assertEqual(inspect_response({"choices": [{"message": {"content": ""}}]}), "invalid_response")


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

curl -sS -w '\nHTTP %{http_code}\n' \
  -H 'Content-Type: application/json' \
  -d '{"model":"fixture-model","messages":[{"role":"user","content":"你好"}]}' \
  http://127.0.0.1:8765/v1/chat/completions

# 正常请求：预期 HTTP 200
curl -sS -w '\nHTTP %{http_code}\n' -H 'Content-Type: application/json' -d '{"model":"fixture-model","messages":[{"role":"user","content":"hello"}]}' http://127.0.0.1:8765/v1/chat/completions

# 模型名称错误：预期 HTTP 404
curl -sS -w '\nHTTP %{http_code}\n' -H 'Content-Type: application/json' -d '{"model":"other","messages":[{"role":"user","content":"hello"}]}' http://127.0.0.1:8765/v1/chat/completions

# 消息为空：预期 HTTP 400
curl -sS -w '\nHTTP %{http_code}\n' -H 'Content-Type: application/json' -d '{"model":"fixture-model","messages":[]}' http://127.0.0.1:8765/v1/chat/completions

curl -sS -w '\nHTTP %{http_code}\n' -H 'Content-Type: application/json' -d '{"model":"fixture-model","messages":[{"role":"user","content":"hello"}]}' http://127.0.0.1:8765/v1/missing