Add Qwen retry and probe command
This commit is contained in:
Daniel Maddern
parent
0ed706dda1
commit
44c5ba4291
4 changed files with 187 additions and 13 deletions
|
|
@ -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")
|
||||
|
|
@ -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}},
|
||||
},
|
||||
},
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue