Stand-In Server से अपने Integration की Testing

अपना कोड परखें, मॉडल नहीं

हर test में असली API बुलाना धीमा है, पैसा ख़र्चता है, CI में keys माँगता है और अनिश्चित है। इसके बजाय मॉडल के आसपास का integration fake server या mock client से परखें: SDK का base_url ऐसे स्थानीय server की ओर रखें जो तैयार उत्तर लौटाता है। तब आप ठीक उस पर assert कर सकते हैं जो आपका कोड भेजता है (headers, body, मॉडल नाम, max_tokens), retries और fallbacks जाँचने को 429, 5xx और timeouts अनुकरण कर सकते हैं, streams और tool-use आदान-प्रदान replay कर सकते हैं, और सब कुछ CI में मुफ़्त चला सकते हैं। गुणवत्ता मापने के लिए असली मॉडल के विरुद्ध अलग छोटा suite (निर्धारित समय पर या releases से पहले चलने वाला evaluation set) रखें, क्योंकि fakes नहीं बता सकते कि उत्तर अच्छे हैं या नहीं। इस कोर्स के सारे आउटपुट इसी तरह बने।

Stand-in server (mock.py नाम से सहेजें)

सिर्फ़ Python standard library से बना छोटा server। वह Anthropic Messages और OpenAI Chat Completions रूपों के लिए तैयार उत्तर लौटाता है, माँग पर 429 दे सकता है, server-sent events stream करता है, और हर अनुरोध STATE["log"] में दर्ज करता है। यह मॉडल नहीं है और वह सब validate नहीं करता जो असली सेवा करती।

"""A tiny local stand-in for the Anthropic Messages API and the OpenAI Chat Completions API.
It returns canned replies, so SDK behaviour (headers, retries, errors, streaming, tools) can be
tested offline. It is NOT a model."""
import json, threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

STATE = {"fail_next": 0, "log": []}

def start(port=0):
    class H(BaseHTTPRequestHandler):
        def log_message(self, *a): pass
        def _send(self, code, obj, headers=None):
            body = json.dumps(obj).encode()
            self.send_response(code)
            self.send_header("content-type", "application/json")
            for k, v in (headers or {}).items(): self.send_header(k, v)
            self.send_header("content-length", str(len(body)))
            self.end_headers(); self.wfile.write(body)
        def do_POST(self):
            n = int(self.headers.get("content-length", 0))
            body = json.loads(self.rfile.read(n) or b"{}")
            STATE["log"].append({"path": self.path, "headers": {k.lower(): v for k, v in self.headers.items()}, "body": body})
            if STATE["fail_next"] > 0:
                STATE["fail_next"] -= 1
                return self._send(429, {"type": "error", "error": {"type": "rate_limit_error", "message": "slow down"}},
                                  {"retry-after": "0"})
            if self.path == "/v1/messages":
                return self.anthropic(body)
            if self.path == "/v1/chat/completions":
                return self.openai(body)
            self._send(404, {"error": {"message": "not found"}})
        def anthropic(self, b):
            if not b.get("max_tokens"):
                return self._send(400, {"type": "error", "error": {"type": "invalid_request_error", "message": "max_tokens: Field required"}})
            if b.get("stream"):
                self.send_response(200); self.send_header("content-type", "text/event-stream"); self.end_headers()
                def ev(name, data): self.wfile.write(f"event: {name}\ndata: {json.dumps(data)}\n\n".encode())
                ev("message_start", {"type": "message_start", "message": {"id": "msg_1", "type": "message", "role": "assistant", "model": b["model"], "content": [], "stop_reason": None, "usage": {"input_tokens": 12, "output_tokens": 1}}})
                ev("content_block_start", {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}})
                for piece in ["Hel", "lo ", "there"]:
                    ev("content_block_delta", {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": piece}})
                ev("content_block_stop", {"type": "content_block_stop", "index": 0})
                ev("message_delta", {"type": "message_delta", "delta": {"stop_reason": "end_turn"}, "usage": {"output_tokens": 3}})
                ev("message_stop", {"type": "message_stop"})
                return
            if b.get("tools"):
                last = b["messages"][-1]
                if last["role"] == "user" and isinstance(last["content"], str):
                    return self._send(200, {"id": "msg_2", "type": "message", "role": "assistant", "model": b["model"],
                        "content": [{"type": "tool_use", "id": "toolu_1", "name": b["tools"][0]["name"], "input": {"order_id": "481516"}}],
                        "stop_reason": "tool_use", "usage": {"input_tokens": 40, "output_tokens": 20}})
                return self._send(200, {"id": "msg_3", "type": "message", "role": "assistant", "model": b["model"],
                    "content": [{"type": "text", "text": "Your order 481516 has shipped."}],
                    "stop_reason": "end_turn", "usage": {"input_tokens": 70, "output_tokens": 9}})
            self._send(200, {"id": "msg_0", "type": "message", "role": "assistant", "model": b["model"],
                "content": [{"type": "text", "text": "Paris is the capital of France."}],
                "stop_reason": "end_turn", "usage": {"input_tokens": 14, "output_tokens": 8}})
        def openai(self, b):
            if b.get("stream"):
                self.send_response(200); self.send_header("content-type", "text/event-stream"); self.end_headers()
                for piece in ["Hel", "lo ", "there"]:
                    c = {"id": "c1", "object": "chat.completion.chunk", "created": 1, "model": b["model"], "choices": [{"index": 0, "delta": {"content": piece}, "finish_reason": None}]}
                    self.wfile.write(f"data: {json.dumps(c)}\n\n".encode())
                end = {"id": "c1", "object": "chat.completion.chunk", "created": 1, "model": b["model"], "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}
                self.wfile.write(f"data: {json.dumps(end)}\n\ndata: [DONE]\n\n".encode()); return
            if b.get("tools") and b["messages"][-1]["role"] == "user":
                msg = {"role": "assistant", "content": None, "tool_calls": [{"id": "call_1", "type": "function",
                       "function": {"name": b["tools"][0]["function"]["name"], "arguments": json.dumps({"order_id": "481516"})}}]}
                fr = "tool_calls"
            elif b.get("tools"):
                msg, fr = {"role": "assistant", "content": "Your order 481516 has shipped."}, "stop"
            else:
                msg, fr = {"role": "assistant", "content": "Paris is the capital of France."}, "stop"
            self._send(200, {"id": "chatcmpl-1", "object": "chat.completion", "created": 1, "model": b["model"],
                "choices": [{"index": 0, "message": msg, "finish_reason": fr}],
                "usage": {"prompt_tokens": 14, "completion_tokens": 8, "total_tokens": 22}})
    srv = ThreadingHTTPServer(("127.0.0.1", port), H)
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    return srv

Retries का test (उदाहरण)

Stand-in server उपयोग करता है और SDK ने जो भेजा उस पर assert करता है। pytest शैली में लिखा; यहाँ test के रूप में चलाया नहीं गया।

import mock, anthropic

def test_retries_once_on_429_then_succeeds():
    srv = mock.start(); base = f"http://127.0.0.1:{srv.server_address[1]}"
    mock.STATE["log"].clear(); mock.STATE["fail_next"] = 1
    client = anthropic.Anthropic(api_key="test", base_url=base, max_retries=2)
    msg = client.messages.create(model="m", max_tokens=10, messages=[{"role": "user", "content": "hi"}])
    assert msg.stop_reason == "end_turn"
    assert len(mock.STATE["log"]) == 2            # first request got 429, second succeeded
    assert mock.STATE["log"][-1]["body"]["max_tokens"] == 10

अप्रिय रास्ते परखें

Stand-in server को विफल, अटकने और ग़लत डेटा भेजने दें, और जाँचें कि आपका कोड शालीनता से घटता है।

त्वरित जाँच: Unit tests में असली API की जगह fake server से क्यों परखें?

• Tests में असली APIs मना हैं
• Fakes उत्तर की गुणवत्ता मापते हैं
• Tests तेज़, मुफ़्त, निश्चित हैं और keys नहीं चाहिए
• Fakes मॉडल प्रशिक्षित करते हैं