From 44c5ba4291e2b032dd9ec765be810f03bfe00013 Mon Sep 17 00:00:00 2001 From: Daniel Maddern Date: Sun, 16 Aug 2026 15:52:46 +0700 Subject: [PATCH] Add Qwen retry and probe command --- .../management/commands/qwen_service_probe.py | 82 +++++++++++++++++++ .../commands/seed_spark_resources.py | 2 + model_router/providers.py | 37 ++++++--- tests/test_model_router_providers.py | 79 +++++++++++++++++- 4 files changed, 187 insertions(+), 13 deletions(-) create mode 100644 control_plane/resources/management/commands/qwen_service_probe.py diff --git a/control_plane/resources/management/commands/qwen_service_probe.py b/control_plane/resources/management/commands/qwen_service_probe.py new file mode 100644 index 0000000..c7e8bff --- /dev/null +++ b/control_plane/resources/management/commands/qwen_service_probe.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import json +import time +from concurrent.futures import ThreadPoolExecutor, as_completed + +from django.core.management.base import BaseCommand, CommandError + +from control_plane.resources.models import Resource +from model_router.providers import QwenProvider +from model_router.router import ModelCapability, ModelRequestContract + + +class Command(BaseCommand): + help = "Probe Qwen/local_inference health and bounded concurrent completion capacity." + + def add_arguments(self, parser): + parser.add_argument("--requests", type=int, default=1) + parser.add_argument("--concurrency", type=int, default=1) + parser.add_argument("--prompt", default="Return only valid JSON: {\"ok\": true, \"model\": \"qwen\"}.") + parser.add_argument("--token-budget", type=int, default=256) + + def handle(self, *args, **options): + resource = Resource.objects.filter(provider="local_inference", is_active=True).first() + if resource is None: + raise CommandError("No Qwen/local_inference resource configured. Run seed_spark_resources first.") + provider = QwenProvider(resource) + before_health = provider.health() + total = max(1, int(options["requests"])) + concurrency = max(1, int(options["concurrency"])) + started = time.monotonic() + + def call(index: int) -> dict[str, object]: + call_started = time.monotonic() + try: + response = provider.complete( + ModelRequestContract( + purpose=ModelCapability.REASONING, + prompt=f"{options['prompt']} Request index: {index}", + token_budget=int(options["token_budget"]), + ) + ) + return { + "index": index, + "status": "COMPLETE", + "latency_ms": int((time.monotonic() - call_started) * 1000), + "content_chars": len(response.content), + "model": response.model, + "attempts": response.metadata.get("attempts"), + } + except Exception as exc: + return { + "index": index, + "status": "FAILED", + "latency_ms": int((time.monotonic() - call_started) * 1000), + "failure": str(exc), + } + + results = [] + with ThreadPoolExecutor(max_workers=concurrency) as executor: + futures = {executor.submit(call, index): index for index in range(1, total + 1)} + for future in as_completed(futures): + results.append(future.result()) + results.sort(key=lambda row: row["index"]) + after_health = provider.health() + completed = sum(1 for row in results if row["status"] == "COMPLETE") + failed = total - completed + latencies = [int(row["latency_ms"]) for row in results] + output = { + "before_health": before_health, + "after_health": after_health, + "requests": total, + "concurrency": concurrency, + "completed": completed, + "failed": failed, + "runtime_seconds": round(time.monotonic() - started, 2), + "latency_ms": {"min": min(latencies), "max": max(latencies), "avg": round(sum(latencies) / len(latencies), 1)}, + "results": results, + } + self.stdout.write(json.dumps(output, indent=2)) + if failed: + raise CommandError(f"Qwen probe failed {failed}/{total} requests") diff --git a/control_plane/resources/management/commands/seed_spark_resources.py b/control_plane/resources/management/commands/seed_spark_resources.py index ba739ac..441096d 100644 --- a/control_plane/resources/management/commands/seed_spark_resources.py +++ b/control_plane/resources/management/commands/seed_spark_resources.py @@ -89,6 +89,8 @@ class Command(BaseCommand): "endpoint_url": os.environ.get("ARTIFEX_QWEN_ENDPOINT_URL", "http://192.168.1.162:8002/v1/chat/completions"), "health_url": os.environ.get("ARTIFEX_QWEN_HEALTH_URL", "http://192.168.1.162:8002/health"), "model": os.environ.get("ARTIFEX_QWEN_MODEL", "qwen38"), + "retry_attempts": int(os.environ.get("ARTIFEX_QWEN_RETRY_ATTEMPTS", "3")), + "retry_backoff_seconds": float(os.environ.get("ARTIFEX_QWEN_RETRY_BACKOFF_SECONDS", "2")), "extra_body": {"chat_template_kwargs": {"enable_thinking": False}}, }, }, diff --git a/model_router/providers.py b/model_router/providers.py index 9bbc362..fb29856 100644 --- a/model_router/providers.py +++ b/model_router/providers.py @@ -5,6 +5,7 @@ import os import re import shlex import subprocess +import time import urllib.error import urllib.request from dataclasses import dataclass @@ -113,6 +114,8 @@ class QwenProvider: config = self.resource.config url = str(config.get("endpoint_url", "http://localhost:8000/v1/chat/completions")) timeout = int(config.get("timeout_seconds", 120)) + attempts = max(1, int(config.get("retry_attempts", 1))) + backoff = float(config.get("retry_backoff_seconds", 1.0)) body = { "model": config.get("model", self.resource.name), "messages": [{"role": "user", "content": request.prompt}], @@ -123,17 +126,23 @@ class QwenProvider: body["response_format"] = config["response_format"] if config.get("extra_body"): body.update(config["extra_body"]) - http_request = urllib.request.Request( - url, - data=json.dumps(body).encode("utf-8"), - headers={"Content-Type": "application/json"}, - method="POST", - ) - try: - with urllib.request.urlopen(http_request, timeout=timeout) as response: - data = json.loads(response.read().decode("utf-8")) - except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc: - raise ProviderError(f"Qwen provider failed: {exc}") from exc + failures = [] + for attempt in range(1, attempts + 1): + http_request = urllib.request.Request( + url, + data=json.dumps(body).encode("utf-8"), + headers={"Content-Type": "application/json"}, + method="POST", + ) + try: + with urllib.request.urlopen(http_request, timeout=timeout) as response: + data = json.loads(response.read().decode("utf-8")) + break + except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc: + failures.append(f"attempt {attempt}/{attempts}: {exc}") + if attempt == attempts: + raise ProviderError("Qwen provider failed after retries: " + " | ".join(failures)) from exc + time.sleep(backoff * attempt) choices = data.get("choices", []) content = "" if choices: @@ -144,7 +153,11 @@ class QwenProvider: return ModelResponseContract( model=str(data.get("model", self.resource.name)), content=content, - metadata={"provider": self.provider_name, "usage": data.get("usage", {})}, + metadata={ + "provider": self.provider_name, + "usage": data.get("usage", {}), + "attempts": len(failures) + 1, + }, ) def health(self) -> str: diff --git a/tests/test_model_router_providers.py b/tests/test_model_router_providers.py index 64a3db3..f0f2166 100644 --- a/tests/test_model_router_providers.py +++ b/tests/test_model_router_providers.py @@ -1,8 +1,11 @@ from __future__ import annotations +import io +import urllib.error + from agents.providers import DeterministicCodingProvider, DeterministicSolProvider from control_plane.resources.models import ModelRequest, Resource, ResourceKind -from model_router.providers import SolProvider, providers_from_resources +from model_router.providers import ProviderError, QwenProvider, SolProvider, providers_from_resources from model_router.router import ModelCapability, ModelRequestContract, ModelRouter @@ -50,3 +53,77 @@ def test_model_router_prefers_exact_model_key_resource_for_persisted_requests() request = ModelRequest.objects.get() assert request.model == "Terra" + + +def test_qwen_provider_retries_transient_http_failures(monkeypatch) -> None: + resource = Resource.objects.create( + name="Qwen Test", + kind=ResourceKind.MODEL, + provider="local_inference", + roles=["REASONING"], + config={ + "endpoint_url": "http://qwen.test/v1/chat/completions", + "model": "qwen38", + "retry_attempts": 2, + "retry_backoff_seconds": 0, + }, + ) + calls = {"count": 0} + + class Response: + status = 200 + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def read(self): + return b'{"model":"qwen38","choices":[{"message":{"content":"ok"}}],"usage":{"total_tokens":1}}' + + def fake_urlopen(request, timeout): + calls["count"] += 1 + if calls["count"] == 1: + raise urllib.error.HTTPError(request.full_url, 500, "Internal Server Error", {}, io.BytesIO()) + return Response() + + monkeypatch.setattr("model_router.providers.urllib.request.urlopen", fake_urlopen) + monkeypatch.setattr("model_router.providers.time.sleep", lambda seconds: None) + + response = QwenProvider(resource).complete(ModelRequestContract(purpose=ModelCapability.REASONING, prompt="hello")) + + assert response.content == "ok" + assert response.metadata["attempts"] == 2 + assert calls["count"] == 2 + + +def test_qwen_provider_reports_retry_exhaustion(monkeypatch) -> None: + resource = Resource.objects.create( + name="Qwen Test", + kind=ResourceKind.MODEL, + provider="local_inference", + roles=["REASONING"], + config={ + "endpoint_url": "http://qwen.test/v1/chat/completions", + "retry_attempts": 2, + "retry_backoff_seconds": 0, + }, + ) + + def fake_urlopen(request, timeout): + raise urllib.error.URLError("connection refused") + + monkeypatch.setattr("model_router.providers.urllib.request.urlopen", fake_urlopen) + monkeypatch.setattr("model_router.providers.time.sleep", lambda seconds: None) + + try: + QwenProvider(resource).complete(ModelRequestContract(purpose=ModelCapability.REASONING, prompt="hello")) + except ProviderError as exc: + message = str(exc) + else: + raise AssertionError("expected ProviderError") + + assert "Qwen provider failed after retries" in message + assert "attempt 1/2" in message + assert "attempt 2/2" in message