forked from retoor/devplacepy
feat: add api key auth, devii agent, openai gateway, and admin service management
This commit introduces a comprehensive set of new features including API key authentication with CLI management commands (get, reset, backfill), a Devii agentic assistant with WebSocket terminal and session bootstrap, an OpenAI-compatible LLM gateway service, and an admin service management panel. It also adds Playwright browser automation for bot support, configures internal gateway URLs, refactors content editing/deletion to support JSON API responses, and updates documentation across AGENTS.md, README.md, and the developer docs site.
This commit is contained in:
@@ -0,0 +1,3 @@
|
||||
from devplacepy.services.openai_gateway.service import GatewayService
|
||||
|
||||
__all__ = ["GatewayService"]
|
||||
@@ -0,0 +1,369 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Optional
|
||||
|
||||
from devplacepy.database import db, get_table
|
||||
from devplacepy.services.openai_gateway.reliability import percentile
|
||||
from devplacepy.services.openai_gateway.usage import (
|
||||
GATEWAY_CONCURRENCY,
|
||||
GATEWAY_LEDGER,
|
||||
PER_MILLION,
|
||||
Pricing,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
MAX_WINDOW_HOURS = 168
|
||||
|
||||
|
||||
def _now() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def _iso(moment: datetime) -> str:
|
||||
return moment.isoformat()
|
||||
|
||||
|
||||
def _pset(values: list[float]) -> dict:
|
||||
if not values:
|
||||
return {"avg": 0.0, "p50": 0.0, "p90": 0.0, "p95": 0.0, "p99": 0.0, "max": 0.0, "count": 0}
|
||||
ordered = sorted(values)
|
||||
return {
|
||||
"avg": round(sum(ordered) / len(ordered), 3),
|
||||
"p50": round(percentile(ordered, 0.50), 3),
|
||||
"p90": round(percentile(ordered, 0.90), 3),
|
||||
"p95": round(percentile(ordered, 0.95), 3),
|
||||
"p99": round(percentile(ordered, 0.99), 3),
|
||||
"max": round(ordered[-1], 3),
|
||||
"count": len(ordered),
|
||||
}
|
||||
|
||||
|
||||
def _positive(rows: list[dict], field: str) -> list[float]:
|
||||
return [float(r[field]) for r in rows if r.get(field) and float(r[field]) > 0]
|
||||
|
||||
|
||||
def _top_group(rows: list[dict], key_fn, top_n: int) -> list[dict]:
|
||||
agg: dict = {}
|
||||
for r in rows:
|
||||
key = key_fn(r)
|
||||
bucket = agg.setdefault(key, {"key": key, "requests": 0, "success": 0,
|
||||
"cost_usd": 0.0, "total_tokens": 0})
|
||||
bucket["requests"] += 1
|
||||
bucket["success"] += int(r.get("success") or 0)
|
||||
bucket["cost_usd"] += float(r.get("cost_usd") or 0)
|
||||
bucket["total_tokens"] += int(r.get("total_tokens") or 0)
|
||||
out = sorted(agg.values(), key=lambda b: b["requests"], reverse=True)
|
||||
for bucket in out:
|
||||
bucket["cost_usd"] = round(bucket["cost_usd"], 6)
|
||||
return out[:top_n] if top_n else out
|
||||
|
||||
|
||||
def _ledger_rows(cutoff: str) -> list[dict]:
|
||||
return list(db.query(
|
||||
f"SELECT * FROM {GATEWAY_LEDGER} WHERE created_at >= :cutoff ORDER BY created_at",
|
||||
cutoff=cutoff,
|
||||
))
|
||||
|
||||
|
||||
def _concurrency(cutoff: str) -> dict:
|
||||
if GATEWAY_CONCURRENCY not in db.tables:
|
||||
return {"peak": 0, "avg": 0.0, "p95": 0.0, "samples": 0}
|
||||
values = [int(r["in_flight"] or 0) for r in db.query(
|
||||
f"SELECT in_flight FROM {GATEWAY_CONCURRENCY} WHERE created_at >= :cutoff",
|
||||
cutoff=cutoff,
|
||||
)]
|
||||
if not values:
|
||||
return {"peak": 0, "avg": 0.0, "p95": 0.0, "samples": 0}
|
||||
ordered = sorted(values)
|
||||
return {
|
||||
"peak": ordered[-1],
|
||||
"avg": round(sum(ordered) / len(ordered), 2),
|
||||
"p95": round(percentile(ordered, 0.95), 2),
|
||||
"samples": len(ordered),
|
||||
}
|
||||
|
||||
|
||||
def empty_payload(hours: int = 48) -> dict:
|
||||
return {
|
||||
"window_hours": hours,
|
||||
"generated_at": _iso(_now()),
|
||||
"requests": 0,
|
||||
"volume": {}, "tokens": {}, "latency": {}, "errors": {}, "cost": {}, "behavior": {},
|
||||
"hourly": [],
|
||||
"notes": {"ttft": "not available: gateway forwards non-streaming upstream"},
|
||||
}
|
||||
|
||||
|
||||
def build_analytics(hours: int = 48, top_n: int = 10, pricing: Optional[Pricing] = None) -> dict:
|
||||
if GATEWAY_LEDGER not in db.tables:
|
||||
return empty_payload(hours)
|
||||
hours = max(1, min(hours, MAX_WINDOW_HOURS))
|
||||
now = _now()
|
||||
cutoff = _iso(now - timedelta(hours=hours))
|
||||
rows = _ledger_rows(cutoff)
|
||||
if not rows:
|
||||
return empty_payload(hours)
|
||||
|
||||
requests = len(rows)
|
||||
success = sum(int(r.get("success") or 0) for r in rows)
|
||||
failed = requests - success
|
||||
|
||||
hour_start = now.strftime("%Y-%m-%dT%H")
|
||||
day_cutoff = _iso(now - timedelta(hours=24))
|
||||
|
||||
minute_counts: dict = {}
|
||||
for r in rows:
|
||||
minute_counts[r["created_at"][:16]] = minute_counts.get(r["created_at"][:16], 0) + 1
|
||||
peak_req_per_min = max(minute_counts.values()) if minute_counts else 0
|
||||
|
||||
first_hour: dict = {}
|
||||
for r in rows:
|
||||
owner = f"{r.get('owner_kind')}:{r.get('owner_id')}"
|
||||
bucket = r["created_at"][:13]
|
||||
if owner not in first_hour or bucket < first_hour[owner]:
|
||||
first_hour[owner] = bucket
|
||||
|
||||
total_cost = sum(float(r.get("cost_usd") or 0) for r in rows)
|
||||
input_cost = sum(float(r.get("input_cost_usd") or 0) for r in rows)
|
||||
output_cost = sum(float(r.get("output_cost_usd") or 0) for r in rows)
|
||||
cost_this_hour = sum(float(r.get("cost_usd") or 0) for r in rows if r["created_at"][:13] == hour_start)
|
||||
cost_24h = sum(float(r.get("cost_usd") or 0) for r in rows if r["created_at"] >= day_cutoff)
|
||||
|
||||
prompt_total = sum(int(r.get("prompt_tokens") or 0) for r in rows)
|
||||
completion_total = sum(int(r.get("completion_tokens") or 0) for r in rows)
|
||||
total_tokens = sum(int(r.get("total_tokens") or 0) for r in rows)
|
||||
cache_hit_total = sum(int(r.get("cache_hit_tokens") or 0) for r in rows)
|
||||
cache_miss_total = sum(int(r.get("cache_miss_tokens") or 0) for r in rows)
|
||||
reasoning_total = sum(int(r.get("reasoning_tokens") or 0) for r in rows)
|
||||
|
||||
caching_savings = 0.0
|
||||
if pricing is not None:
|
||||
rate_delta = pricing.chat_cache_miss_per_m - pricing.chat_cache_hit_per_m
|
||||
chat_cache_hits = sum(int(r.get("cache_hit_tokens") or 0) for r in rows if r.get("backend") == "chat")
|
||||
caching_savings = chat_cache_hits / PER_MILLION * rate_delta
|
||||
|
||||
util_values = [float(r["context_utilization"]) for r in rows if r.get("context_utilization") is not None]
|
||||
|
||||
owner_counts: dict = {}
|
||||
for r in rows:
|
||||
owner = f"{r.get('owner_kind')}:{r.get('owner_id')}"
|
||||
owner_counts[owner] = owner_counts.get(owner, 0) + 1
|
||||
per_owner = sorted(owner_counts.values())
|
||||
|
||||
ua_counts: dict = {}
|
||||
for r in rows:
|
||||
ua = r.get("user_agent") or "unknown"
|
||||
ua_counts[ua] = ua_counts.get(ua, 0) + 1
|
||||
user_agents = sorted(
|
||||
[{"key": k, "requests": v} for k, v in ua_counts.items()],
|
||||
key=lambda x: x["requests"], reverse=True,
|
||||
)[:top_n]
|
||||
|
||||
error_categories: dict = {}
|
||||
for r in rows:
|
||||
category = r.get("error_category")
|
||||
if category:
|
||||
error_categories[category] = error_categories.get(category, 0) + 1
|
||||
count_4xx = sum(1 for r in rows if 400 <= int(r.get("status_code") or 0) < 500)
|
||||
count_5xx = sum(1 for r in rows if int(r.get("status_code") or 0) >= 500)
|
||||
|
||||
temperatures = [float(r["temperature"]) for r in rows if r.get("temperature") is not None]
|
||||
top_ps = [float(r["top_p"]) for r in rows if r.get("top_p") is not None]
|
||||
|
||||
hourly = _hourly(rows, first_hour)
|
||||
|
||||
volume = {
|
||||
"requests": requests,
|
||||
"success": success,
|
||||
"failed": failed,
|
||||
"requests_per_hour": round(requests / hours, 2),
|
||||
"peak_requests_per_minute": peak_req_per_min,
|
||||
"by_model": _top_group(rows, lambda r: r.get("model") or "unknown", top_n),
|
||||
"by_endpoint": _top_group(rows, lambda r: r.get("endpoint") or "unknown", top_n),
|
||||
"by_backend": _top_group(rows, lambda r: r.get("backend") or "unknown", 0),
|
||||
"by_caller": _top_group(rows, lambda r: f"{r.get('owner_kind')}:{r.get('owner_id')}", top_n),
|
||||
"concurrency": _concurrency(cutoff),
|
||||
}
|
||||
|
||||
tokens = {
|
||||
"prompt_total": prompt_total,
|
||||
"completion_total": completion_total,
|
||||
"total": total_tokens,
|
||||
"cache_hit_total": cache_hit_total,
|
||||
"cache_miss_total": cache_miss_total,
|
||||
"reasoning_total": reasoning_total,
|
||||
"cache_hit_rate": round(cache_hit_total / prompt_total, 4) if prompt_total else 0.0,
|
||||
"input_output_ratio": round(prompt_total / completion_total, 3) if completion_total else 0.0,
|
||||
"tokens_per_hour": round(total_tokens / hours, 1),
|
||||
"context_utilization_avg": round(sum(util_values) / len(util_values), 4) if util_values else None,
|
||||
"prompt": _pset(_positive(rows, "prompt_tokens")),
|
||||
"completion": _pset(_positive(rows, "completion_tokens")),
|
||||
"total_per_request": _pset(_positive(rows, "total_tokens")),
|
||||
}
|
||||
|
||||
avg_upstream = sum(_positive(rows, "upstream_latency_ms")) / max(len(_positive(rows, "upstream_latency_ms")), 1)
|
||||
avg_total = sum(_positive(rows, "total_latency_ms")) / max(len(_positive(rows, "total_latency_ms")), 1)
|
||||
latency = {
|
||||
"upstream_ms": _pset(_positive(rows, "upstream_latency_ms")),
|
||||
"gateway_overhead_ms": _pset(_positive(rows, "gateway_overhead_ms")),
|
||||
"queue_wait_ms": _pset(_positive(rows, "queue_wait_ms")),
|
||||
"connect_ms": _pset(_positive(rows, "connect_ms")),
|
||||
"total_ms": _pset(_positive(rows, "total_latency_ms")),
|
||||
"tokens_per_second": _pset(_positive(rows, "tokens_per_second")),
|
||||
"gateway_overhead_share": round((avg_total - avg_upstream) / avg_total, 4) if avg_total else 0.0,
|
||||
"upstream_availability_pct": round(success / requests * 100, 2) if requests else 0.0,
|
||||
"ttft_ms": None,
|
||||
"inter_token_ms": None,
|
||||
}
|
||||
|
||||
errors = {
|
||||
"total": failed,
|
||||
"error_rate_pct": round(failed / requests * 100, 2) if requests else 0.0,
|
||||
"by_category": error_categories,
|
||||
"count_4xx": count_4xx,
|
||||
"count_5xx": count_5xx,
|
||||
"timeouts": error_categories.get("timeout", 0),
|
||||
"retries_attempted": sum(int(r.get("retries_attempted") or 0) for r in rows),
|
||||
"retries_succeeded": sum(int(r.get("retry_succeeded") or 0) for r in rows),
|
||||
"circuit_open_events": sum(int(r.get("circuit_open") or 0) for r in rows),
|
||||
}
|
||||
|
||||
cost = {
|
||||
"total_usd": round(total_cost, 6),
|
||||
"input_usd": round(input_cost, 6),
|
||||
"output_usd": round(output_cost, 6),
|
||||
"this_hour_usd": round(cost_this_hour, 6),
|
||||
"last_24h_usd": round(cost_24h, 6),
|
||||
"projected_monthly_usd": round(cost_24h * 30, 2),
|
||||
"effective_per_1k_tokens_usd": round(total_cost / total_tokens * 1000, 6) if total_tokens else 0.0,
|
||||
"caching_savings_usd": round(caching_savings, 6),
|
||||
"by_model": [{"key": b["key"], "cost_usd": b["cost_usd"]} for b in _top_group(rows, lambda r: r.get("model") or "unknown", top_n)],
|
||||
"by_caller": [{"key": b["key"], "cost_usd": b["cost_usd"]} for b in _top_group(rows, lambda r: f"{r.get('owner_kind')}:{r.get('owner_id')}", top_n)],
|
||||
}
|
||||
|
||||
behavior = {
|
||||
"unique_callers": len(owner_counts),
|
||||
"requests_per_caller": {
|
||||
"avg": round(sum(per_owner) / len(per_owner), 2) if per_owner else 0.0,
|
||||
"p95": round(percentile(per_owner, 0.95), 2) if per_owner else 0.0,
|
||||
"max": per_owner[-1] if per_owner else 0,
|
||||
},
|
||||
"avg_temperature": round(sum(temperatures) / len(temperatures), 3) if temperatures else None,
|
||||
"avg_top_p": round(sum(top_ps) / len(top_ps), 3) if top_ps else None,
|
||||
"tool_call_requests": sum(int(r.get("has_tools") or 0) for r in rows),
|
||||
"streaming_requests": sum(int(r.get("stream_requested") or 0) for r in rows),
|
||||
"non_streaming_requests": requests - sum(int(r.get("stream_requested") or 0) for r in rows),
|
||||
"user_agents": user_agents,
|
||||
}
|
||||
|
||||
return {
|
||||
"window_hours": hours,
|
||||
"generated_at": _iso(now),
|
||||
"requests": requests,
|
||||
"volume": volume,
|
||||
"tokens": tokens,
|
||||
"latency": latency,
|
||||
"errors": errors,
|
||||
"cost": cost,
|
||||
"behavior": behavior,
|
||||
"hourly": hourly,
|
||||
"notes": {"ttft": "not available: gateway forwards non-streaming upstream"},
|
||||
}
|
||||
|
||||
|
||||
def _hourly(rows: list[dict], first_hour: dict) -> list[dict]:
|
||||
buckets: dict = {}
|
||||
for r in rows:
|
||||
hour = r["created_at"][:13]
|
||||
owner = f"{r.get('owner_kind')}:{r.get('owner_id')}"
|
||||
bucket = buckets.setdefault(hour, {
|
||||
"hour": hour, "requests": 0, "success": 0, "failed": 0,
|
||||
"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0,
|
||||
"cost_usd": 0.0, "latency_sum": 0.0, "latency_n": 0,
|
||||
"owners": set(), "new_owners": 0,
|
||||
})
|
||||
bucket["requests"] += 1
|
||||
if int(r.get("success") or 0):
|
||||
bucket["success"] += 1
|
||||
else:
|
||||
bucket["failed"] += 1
|
||||
bucket["prompt_tokens"] += int(r.get("prompt_tokens") or 0)
|
||||
bucket["completion_tokens"] += int(r.get("completion_tokens") or 0)
|
||||
bucket["total_tokens"] += int(r.get("total_tokens") or 0)
|
||||
bucket["cost_usd"] += float(r.get("cost_usd") or 0)
|
||||
latency = float(r.get("upstream_latency_ms") or 0)
|
||||
if latency > 0:
|
||||
bucket["latency_sum"] += latency
|
||||
bucket["latency_n"] += 1
|
||||
bucket["owners"].add(owner)
|
||||
|
||||
for hour, bucket in buckets.items():
|
||||
bucket["new_owners"] = sum(1 for owner in bucket["owners"] if first_hour.get(owner) == hour)
|
||||
bucket["active_owners"] = len(bucket["owners"])
|
||||
bucket["returning_owners"] = bucket["active_owners"] - bucket["new_owners"]
|
||||
bucket["avg_latency_ms"] = round(bucket["latency_sum"] / bucket["latency_n"], 1) if bucket["latency_n"] else 0.0
|
||||
bucket["cost_usd"] = round(bucket["cost_usd"], 6)
|
||||
del bucket["owners"], bucket["latency_sum"], bucket["latency_n"]
|
||||
|
||||
return sorted(buckets.values(), key=lambda b: b["hour"], reverse=True)
|
||||
|
||||
|
||||
def summary_metrics() -> dict:
|
||||
zero = {"requests": 0, "success_pct": 0.0, "error_pct": 0.0, "cost_hour": 0.0,
|
||||
"cost_24h": 0.0, "tokens_24h": 0, "avg_latency_ms": 0.0, "avg_tps": 0.0,
|
||||
"peak_concurrency": 0, "top_model": "-", "top_caller": "-"}
|
||||
if GATEWAY_LEDGER not in db.tables:
|
||||
return zero
|
||||
now = _now()
|
||||
cutoff = _iso(now - timedelta(hours=24))
|
||||
hour_start = now.strftime("%Y-%m-%dT%H")
|
||||
agg = list(db.query(
|
||||
f"SELECT COUNT(*) AS requests, COALESCE(SUM(success),0) AS ok, "
|
||||
f"COALESCE(SUM(cost_usd),0) AS cost, COALESCE(SUM(total_tokens),0) AS tokens, "
|
||||
f"COALESCE(AVG(upstream_latency_ms),0) AS avg_lat, "
|
||||
f"COALESCE(AVG(NULLIF(tokens_per_second,0)),0) AS avg_tps "
|
||||
f"FROM {GATEWAY_LEDGER} WHERE created_at >= :cutoff",
|
||||
cutoff=cutoff,
|
||||
))
|
||||
row = agg[0] if agg else {}
|
||||
requests = int(row.get("requests") or 0)
|
||||
if not requests:
|
||||
return zero
|
||||
ok = int(row.get("ok") or 0)
|
||||
hour_rows = list(db.query(
|
||||
f"SELECT COALESCE(SUM(cost_usd),0) AS c FROM {GATEWAY_LEDGER} WHERE created_at >= :h",
|
||||
h=hour_start,
|
||||
))
|
||||
top_model = list(db.query(
|
||||
f"SELECT model, COUNT(*) AS n FROM {GATEWAY_LEDGER} WHERE created_at >= :cutoff "
|
||||
f"GROUP BY model ORDER BY n DESC LIMIT 1", cutoff=cutoff,
|
||||
))
|
||||
top_caller = list(db.query(
|
||||
f"SELECT owner_kind, owner_id, COUNT(*) AS n FROM {GATEWAY_LEDGER} WHERE created_at >= :cutoff "
|
||||
f"GROUP BY owner_kind, owner_id ORDER BY n DESC LIMIT 1", cutoff=cutoff,
|
||||
))
|
||||
peak = 0
|
||||
if GATEWAY_CONCURRENCY in db.tables:
|
||||
peak_rows = list(db.query(
|
||||
f"SELECT COALESCE(MAX(in_flight),0) AS m FROM {GATEWAY_CONCURRENCY} WHERE created_at >= :cutoff",
|
||||
cutoff=cutoff,
|
||||
))
|
||||
peak = int(peak_rows[0]["m"]) if peak_rows else 0
|
||||
return {
|
||||
"requests": requests,
|
||||
"success_pct": round(ok / requests * 100, 1),
|
||||
"error_pct": round((requests - ok) / requests * 100, 1),
|
||||
"cost_hour": round(float(hour_rows[0]["c"]) if hour_rows else 0.0, 6),
|
||||
"cost_24h": round(float(row.get("cost") or 0), 6),
|
||||
"tokens_24h": int(row.get("tokens") or 0),
|
||||
"avg_latency_ms": round(float(row.get("avg_lat") or 0), 1),
|
||||
"avg_tps": round(float(row.get("avg_tps") or 0), 2),
|
||||
"peak_concurrency": peak,
|
||||
"top_model": top_model[0]["model"] if top_model and top_model[0].get("model") else "-",
|
||||
"top_caller": f"{top_caller[0]['owner_kind']}:{top_caller[0]['owner_id']}" if top_caller else "-",
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
UPSTREAM_URL_DEFAULT = "https://api.deepseek.com/chat/completions"
|
||||
MODEL_DEFAULT = "deepseek-chat"
|
||||
TIMEOUT_DEFAULT = 180
|
||||
INSTANCES_DEFAULT = 4
|
||||
|
||||
VISION_URL_DEFAULT = "https://openrouter.ai/api/v1/chat/completions"
|
||||
VISION_MODEL_DEFAULT = "google/gemma-3-12b-it"
|
||||
VISION_CACHE_SIZE_DEFAULT = 256
|
||||
|
||||
VISION_INSTRUCTION = (
|
||||
"Describe this image in detail. Note objects, people, scene, any visible "
|
||||
"text, layout, colors, and anything else that could be relevant for "
|
||||
"answering questions about it. Be specific but concise."
|
||||
)
|
||||
|
||||
PRICE_CACHE_HIT_PER_M_DEFAULT = 0.0028
|
||||
PRICE_CACHE_MISS_PER_M_DEFAULT = 0.14
|
||||
PRICE_OUTPUT_PER_M_DEFAULT = 0.28
|
||||
VISION_PRICE_INPUT_PER_M_DEFAULT = 0.0
|
||||
VISION_PRICE_OUTPUT_PER_M_DEFAULT = 0.0
|
||||
|
||||
USAGE_RETENTION_HOURS_DEFAULT = 720
|
||||
|
||||
MAX_RETRIES_DEFAULT = 2
|
||||
RETRY_BACKOFF_MS_DEFAULT = 250
|
||||
CIRCUIT_THRESHOLD_DEFAULT = 5
|
||||
CIRCUIT_COOLDOWN_SECONDS_DEFAULT = 30
|
||||
|
||||
MODEL_CONTEXT_MAP_DEFAULT = {
|
||||
"deepseek-chat": 65536,
|
||||
"deepseek-reasoner": 65536,
|
||||
"google/gemma-3-12b-it": 8192,
|
||||
}
|
||||
@@ -0,0 +1,311 @@
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
import uuid
|
||||
from typing import Optional
|
||||
|
||||
import httpx
|
||||
from fastapi.responses import JSONResponse, Response, StreamingResponse
|
||||
|
||||
from devplacepy.services.openai_gateway import config
|
||||
from devplacepy.services.openai_gateway.reliability import CircuitBreaker, retry_send
|
||||
from devplacepy.services.openai_gateway.usage import (
|
||||
GatewayUsageLedger,
|
||||
classify_error,
|
||||
extract_params,
|
||||
parse_context_map,
|
||||
pricing_from_cfg,
|
||||
)
|
||||
from devplacepy.services.openai_gateway.vision import VisionAugmenter, VisionCache
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _fake_stream(data: dict, model: str):
|
||||
chunk_id = data.get("id", f"chatcmpl-{uuid.uuid4().hex[:12]}")
|
||||
created = data.get("created", int(time.time()))
|
||||
out_model = data.get("model", model)
|
||||
|
||||
try:
|
||||
msg = data["choices"][0]["message"]
|
||||
except (KeyError, IndexError):
|
||||
msg = {"content": ""}
|
||||
tool_calls = msg.get("tool_calls")
|
||||
content = msg.get("content") or ""
|
||||
reasoning_content = msg.get("reasoning_content") or ""
|
||||
|
||||
def _chunk(delta: dict, finish: Optional[str] = None) -> str:
|
||||
return "data: " + json.dumps({
|
||||
"id": chunk_id,
|
||||
"object": "chat.completion.chunk",
|
||||
"created": created,
|
||||
"model": out_model,
|
||||
"choices": [{"index": 0, "delta": delta, "finish_reason": finish}],
|
||||
}) + "\n\n"
|
||||
|
||||
async def gen():
|
||||
yield _chunk({"role": "assistant"})
|
||||
if reasoning_content:
|
||||
for i in range(0, len(reasoning_content), 50):
|
||||
yield _chunk({"reasoning_content": reasoning_content[i:i + 50]})
|
||||
if tool_calls:
|
||||
yield _chunk({"tool_calls": tool_calls})
|
||||
elif content:
|
||||
for i in range(0, len(content), 50):
|
||||
yield _chunk({"content": content[i:i + 50]})
|
||||
yield _chunk({}, finish="tool_calls" if tool_calls else "stop")
|
||||
yield "data: [DONE]\n\n"
|
||||
|
||||
return gen()
|
||||
|
||||
|
||||
def _connect_tracer(holder: dict):
|
||||
started: dict = {}
|
||||
|
||||
async def trace(name: str, info: dict) -> None:
|
||||
if name.endswith("connect_tcp.started") or name.endswith("start_tls.started"):
|
||||
started[name] = time.monotonic()
|
||||
elif name.endswith("connect_tcp.complete") or name.endswith("start_tls.complete"):
|
||||
begin = started.get(name.replace(".complete", ".started"))
|
||||
if begin is not None:
|
||||
holder["ms"] += (time.monotonic() - begin) * 1000
|
||||
|
||||
return trace
|
||||
|
||||
|
||||
class GatewayRuntime:
|
||||
def __init__(self):
|
||||
self._client: Optional[httpx.AsyncClient] = None
|
||||
self._sem: Optional[asyncio.Semaphore] = None
|
||||
self._instances = 0
|
||||
self._timeout = 0
|
||||
self._vision_cache: Optional[VisionCache] = None
|
||||
self._vision_cache_size = -1
|
||||
self._ledger = GatewayUsageLedger()
|
||||
self._breaker = CircuitBreaker(config.CIRCUIT_THRESHOLD_DEFAULT, config.CIRCUIT_COOLDOWN_SECONDS_DEFAULT)
|
||||
self.requests = 0
|
||||
self.errors = 0
|
||||
self.in_flight = 0
|
||||
self.peak_in_flight = 0
|
||||
self.vision_calls = 0
|
||||
self.last_status = 0
|
||||
self.last_latency_ms = 0
|
||||
|
||||
def _ensure(self, cfg: dict):
|
||||
instances = max(1, cfg["gateway_instances"])
|
||||
timeout = max(1, cfg["gateway_timeout"])
|
||||
if self._client is None or instances != self._instances or timeout != self._timeout:
|
||||
old = self._client
|
||||
limits = httpx.Limits(max_connections=instances, max_keepalive_connections=instances)
|
||||
self._client = httpx.AsyncClient(timeout=float(timeout), limits=limits)
|
||||
self._sem = asyncio.Semaphore(instances)
|
||||
self._instances = instances
|
||||
self._timeout = timeout
|
||||
if old is not None:
|
||||
asyncio.create_task(old.aclose())
|
||||
size = cfg["gateway_vision_cache_size"]
|
||||
if self._vision_cache is None or size != self._vision_cache_size:
|
||||
self._vision_cache = VisionCache(size)
|
||||
self._vision_cache_size = size
|
||||
self._breaker.configure(cfg["gateway_circuit_threshold"], cfg["gateway_circuit_cooldown_seconds"])
|
||||
return self._client, self._sem
|
||||
|
||||
async def aclose(self) -> None:
|
||||
if self._client is not None:
|
||||
await self._client.aclose()
|
||||
self._client = None
|
||||
self._instances = 0
|
||||
|
||||
async def _send(self, client, sem, method, url, headers, cfg, log, json_body=None, content=None):
|
||||
timing = {"queue_wait_ms": 0.0, "upstream_latency_ms": 0.0, "connect_ms": 0.0,
|
||||
"retries_attempted": 0, "retry_succeeded": False, "circuit_open": False}
|
||||
if not self._breaker.allow():
|
||||
timing["circuit_open"] = True
|
||||
log("circuit breaker open, rejecting upstream call")
|
||||
return None, None, timing
|
||||
self.requests += 1
|
||||
self.in_flight += 1
|
||||
if self.in_flight > self.peak_in_flight:
|
||||
self.peak_in_flight = self.in_flight
|
||||
wait_start = time.monotonic()
|
||||
connect_holder = {"ms": 0.0}
|
||||
attempts = 1
|
||||
resp = None
|
||||
exc = None
|
||||
try:
|
||||
async with sem:
|
||||
timing["queue_wait_ms"] = round((time.monotonic() - wait_start) * 1000, 3)
|
||||
|
||||
async def do_call():
|
||||
request = client.build_request(method, url, headers=headers,
|
||||
json=json_body, content=content)
|
||||
request.extensions["trace"] = _connect_tracer(connect_holder)
|
||||
return await client.send(request)
|
||||
|
||||
send_start = time.monotonic()
|
||||
resp, exc, attempts = await retry_send(
|
||||
do_call, cfg["gateway_max_retries"], cfg["gateway_retry_backoff_ms"], log)
|
||||
timing["upstream_latency_ms"] = round((time.monotonic() - send_start) * 1000, 3)
|
||||
finally:
|
||||
self.in_flight -= 1
|
||||
timing["connect_ms"] = round(connect_holder["ms"], 3)
|
||||
timing["retries_attempted"] = max(attempts - 1, 0)
|
||||
self.last_latency_ms = int(timing["upstream_latency_ms"])
|
||||
if exc is not None:
|
||||
self.errors += 1
|
||||
self._breaker.record_failure()
|
||||
log(f"{method} {url} connection failed after {attempts} attempt(s): {exc}")
|
||||
return None, exc, timing
|
||||
self.last_status = resp.status_code
|
||||
if resp.status_code >= 500:
|
||||
self.errors += 1
|
||||
self._breaker.record_failure()
|
||||
else:
|
||||
self._breaker.record_success()
|
||||
timing["retry_succeeded"] = attempts > 1
|
||||
return resp, None, timing
|
||||
|
||||
async def handle_chat(self, body: dict, cfg: dict, owner: tuple, user_agent: str, log=None):
|
||||
log = log or (lambda message: None)
|
||||
client, sem = self._ensure(cfg)
|
||||
pricing = pricing_from_cfg(cfg)
|
||||
context_map = parse_context_map(cfg.get("gateway_model_context_map"))
|
||||
params = extract_params(body)
|
||||
handle_start = time.monotonic()
|
||||
messages = body.get("messages", []) or []
|
||||
|
||||
if cfg["gateway_vision_enabled"]:
|
||||
augmenter = VisionAugmenter(
|
||||
cfg["gateway_vision_url"], cfg["gateway_vision_model"], cfg["gateway_vision_key"],
|
||||
self._vision_cache, ledger=self._ledger, owner=owner, pricing=pricing,
|
||||
context_map=context_map,
|
||||
)
|
||||
messages = await augmenter.augment_messages(client, messages)
|
||||
self.vision_calls += augmenter.calls
|
||||
|
||||
requested = body.get("model")
|
||||
if cfg["gateway_force_model"] or not requested or requested == "molodetz":
|
||||
model = cfg["gateway_model"]
|
||||
else:
|
||||
model = requested
|
||||
|
||||
stream = bool(body.get("stream"))
|
||||
payload = dict(body)
|
||||
payload["model"] = model
|
||||
payload["messages"] = messages
|
||||
payload["stream"] = False
|
||||
|
||||
headers = {"Content-Type": "application/json"}
|
||||
if cfg["gateway_api_key"]:
|
||||
headers["Authorization"] = f"Bearer {cfg['gateway_api_key']}"
|
||||
else:
|
||||
log("No upstream API key configured (gateway_api_key / DEEPSEEK_API_KEY / OPENROUTER_API_KEY); upstream will likely reject the request")
|
||||
|
||||
resp, exc, timing = await self._send(
|
||||
client, sem, "POST", cfg["gateway_upstream_url"], headers, cfg, log, json_body=payload)
|
||||
|
||||
base = {
|
||||
"owner_kind": owner[0], "owner_id": owner[1], "backend": "chat",
|
||||
"endpoint": "chat/completions", "model": model, "user_agent": user_agent,
|
||||
**params, **timing,
|
||||
}
|
||||
|
||||
def finalize(status_code, success, category, usage=None):
|
||||
base["total_latency_ms"] = round((time.monotonic() - handle_start) * 1000, 3)
|
||||
base["gateway_overhead_ms"] = round(max(
|
||||
base["total_latency_ms"] - timing["upstream_latency_ms"] - timing["queue_wait_ms"], 0.0), 3)
|
||||
base["status_code"] = status_code
|
||||
base["success"] = success
|
||||
base["error_category"] = category
|
||||
base["usage"] = usage
|
||||
self._ledger.record(base, pricing, context_map)
|
||||
|
||||
if timing["circuit_open"]:
|
||||
finalize(503, False, "circuit_open")
|
||||
return JSONResponse(status_code=503, content={"error": {"message": "Upstream temporarily unavailable", "type": "circuit_open"}})
|
||||
if exc is not None:
|
||||
finalize(502, False, classify_error(0, exc))
|
||||
return JSONResponse(status_code=502, content={"error": {"message": f"Upstream connection failed: {exc}", "type": "upstream_error"}})
|
||||
if resp.status_code != 200:
|
||||
finalize(resp.status_code, False, classify_error(resp.status_code, None, resp.text))
|
||||
log(f"chat upstream POST -> {resp.status_code}: {resp.text[:300]}")
|
||||
return JSONResponse(status_code=resp.status_code, content={"error": {"message": resp.text, "type": "upstream_error"}})
|
||||
try:
|
||||
data = resp.json()
|
||||
except ValueError:
|
||||
self.errors += 1
|
||||
finalize(502, False, "gateway")
|
||||
log("chat upstream returned 200 but body was not valid JSON")
|
||||
return JSONResponse(status_code=502, content={"error": {"message": "invalid upstream response", "type": "upstream_error"}})
|
||||
finalize(200, True, None, data.get("usage"))
|
||||
log(f"chat POST -> 200 ({timing['upstream_latency_ms']:.0f}ms)")
|
||||
if stream:
|
||||
return StreamingResponse(_fake_stream(data, model), media_type="text/event-stream")
|
||||
return JSONResponse(content=data)
|
||||
|
||||
async def handle_passthrough(self, method: str, subpath: str, content_type: str, body: bytes,
|
||||
cfg: dict, owner: tuple, user_agent: str, log=None):
|
||||
log = log or (lambda message: None)
|
||||
client, sem = self._ensure(cfg)
|
||||
pricing = pricing_from_cfg(cfg)
|
||||
context_map = parse_context_map(cfg.get("gateway_model_context_map"))
|
||||
handle_start = time.monotonic()
|
||||
base_url = cfg["gateway_upstream_url"]
|
||||
if base_url.endswith("/chat/completions"):
|
||||
base_url = base_url[: -len("/chat/completions")]
|
||||
url = f"{base_url.rstrip('/')}/{subpath}"
|
||||
headers = {}
|
||||
if cfg["gateway_api_key"]:
|
||||
headers["Authorization"] = f"Bearer {cfg['gateway_api_key']}"
|
||||
if content_type:
|
||||
headers["Content-Type"] = content_type
|
||||
|
||||
resp, exc, timing = await self._send(client, sem, method, url, headers, cfg, log, content=body)
|
||||
|
||||
base = {
|
||||
"owner_kind": owner[0], "owner_id": owner[1], "backend": "chat",
|
||||
"endpoint": subpath, "model": cfg["gateway_model"], "user_agent": user_agent,
|
||||
**timing,
|
||||
}
|
||||
|
||||
def finalize(status_code, success, category, usage=None):
|
||||
base["total_latency_ms"] = round((time.monotonic() - handle_start) * 1000, 3)
|
||||
base["gateway_overhead_ms"] = round(max(
|
||||
base["total_latency_ms"] - timing["upstream_latency_ms"] - timing["queue_wait_ms"], 0.0), 3)
|
||||
base["status_code"] = status_code
|
||||
base["success"] = success
|
||||
base["error_category"] = category
|
||||
base["usage"] = usage
|
||||
self._ledger.record(base, pricing, context_map)
|
||||
|
||||
if timing["circuit_open"]:
|
||||
finalize(503, False, "circuit_open")
|
||||
return JSONResponse(status_code=503, content={"error": {"message": "Upstream temporarily unavailable", "type": "circuit_open"}})
|
||||
if exc is not None:
|
||||
finalize(502, False, classify_error(0, exc))
|
||||
return JSONResponse(status_code=502, content={"error": {"message": f"Upstream connection failed: {exc}", "type": "upstream_error"}})
|
||||
usage = None
|
||||
if resp.status_code < 400 and "application/json" in (resp.headers.get("content-type") or ""):
|
||||
try:
|
||||
usage = resp.json().get("usage")
|
||||
except ValueError:
|
||||
usage = None
|
||||
finalize(resp.status_code, resp.status_code < 400,
|
||||
None if resp.status_code < 400 else classify_error(resp.status_code, None, resp.text), usage)
|
||||
log(f"passthrough {method} {url} -> {resp.status_code} ({timing['upstream_latency_ms']:.0f}ms)")
|
||||
return Response(content=resp.content, status_code=resp.status_code,
|
||||
media_type=resp.headers.get("content-type"))
|
||||
|
||||
def metrics(self) -> dict:
|
||||
return {
|
||||
"requests": self.requests,
|
||||
"errors": self.errors,
|
||||
"in_flight": self.in_flight,
|
||||
"peak_in_flight": self.peak_in_flight,
|
||||
"vision_calls": self.vision_calls,
|
||||
"last_status": self.last_status,
|
||||
"last_latency_ms": self.last_latency_ms,
|
||||
"pool": self._instances,
|
||||
"circuit_open": self._breaker.is_open,
|
||||
}
|
||||
@@ -0,0 +1,92 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import time
|
||||
from typing import Awaitable, Callable, Optional
|
||||
|
||||
import httpx
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def percentile(sorted_values: list[float], q: float) -> float:
|
||||
if not sorted_values:
|
||||
return 0.0
|
||||
if len(sorted_values) == 1:
|
||||
return float(sorted_values[0])
|
||||
rank = (len(sorted_values) - 1) * q
|
||||
low = int(rank)
|
||||
high = min(low + 1, len(sorted_values) - 1)
|
||||
frac = rank - low
|
||||
return float(sorted_values[low] + (sorted_values[high] - sorted_values[low]) * frac)
|
||||
|
||||
|
||||
class CircuitBreaker:
|
||||
def __init__(self, threshold: int, cooldown_seconds: int):
|
||||
self.threshold = threshold
|
||||
self.cooldown_seconds = cooldown_seconds
|
||||
self.failures = 0
|
||||
self.opened_at: Optional[float] = None
|
||||
|
||||
def configure(self, threshold: int, cooldown_seconds: int) -> None:
|
||||
self.threshold = threshold
|
||||
self.cooldown_seconds = cooldown_seconds
|
||||
|
||||
@property
|
||||
def is_open(self) -> bool:
|
||||
return self.opened_at is not None
|
||||
|
||||
def allow(self) -> bool:
|
||||
if self.opened_at is None:
|
||||
return True
|
||||
if (time.monotonic() - self.opened_at) >= self.cooldown_seconds:
|
||||
self.opened_at = None
|
||||
self.failures = 0
|
||||
return True
|
||||
return False
|
||||
|
||||
def record_success(self) -> None:
|
||||
self.failures = 0
|
||||
self.opened_at = None
|
||||
|
||||
def record_failure(self) -> None:
|
||||
self.failures += 1
|
||||
if self.threshold > 0 and self.failures >= self.threshold:
|
||||
self.opened_at = time.monotonic()
|
||||
|
||||
|
||||
async def _backoff(backoff_ms: int, attempt: int) -> None:
|
||||
delay = max(0, backoff_ms) * attempt / 1000.0
|
||||
if delay > 0:
|
||||
await asyncio.sleep(delay)
|
||||
|
||||
|
||||
async def retry_send(
|
||||
do_call: Callable[[], Awaitable[httpx.Response]],
|
||||
max_retries: int,
|
||||
backoff_ms: int,
|
||||
log: Optional[Callable[[str], None]] = None,
|
||||
) -> tuple[Optional[httpx.Response], Optional[Exception], int]:
|
||||
log = log or (lambda message: None)
|
||||
attempts = 0
|
||||
last_exc: Optional[Exception] = None
|
||||
while attempts <= max_retries:
|
||||
attempts += 1
|
||||
try:
|
||||
resp = await do_call()
|
||||
except httpx.RequestError as exc:
|
||||
last_exc = exc
|
||||
if attempts > max_retries:
|
||||
return None, exc, attempts
|
||||
log(f"upstream connection failed, retrying ({attempts}/{max_retries}): {exc}")
|
||||
await _backoff(backoff_ms, attempts)
|
||||
continue
|
||||
if resp.status_code >= 500 and attempts <= max_retries:
|
||||
log(f"upstream {resp.status_code}, retrying ({attempts}/{max_retries})")
|
||||
await _backoff(backoff_ms, attempts)
|
||||
continue
|
||||
return resp, None, attempts
|
||||
return None, last_exc, attempts
|
||||
@@ -0,0 +1,231 @@
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
|
||||
from fastapi import HTTPException, Request
|
||||
|
||||
from devplacepy.database import get_int_setting
|
||||
from devplacepy.services.base import BaseService, ConfigField
|
||||
from devplacepy.services.openai_gateway import config
|
||||
from devplacepy.services.openai_gateway.analytics import summary_metrics
|
||||
from devplacepy.services.openai_gateway.gateway import GatewayRuntime
|
||||
from devplacepy.utils import get_current_user
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _presented_key(request: Request) -> str:
|
||||
key = request.headers.get("X-API-KEY")
|
||||
if key:
|
||||
return key.strip()
|
||||
scheme, _, credentials = request.headers.get("Authorization", "").partition(" ")
|
||||
if scheme.lower() == "bearer" and credentials.strip():
|
||||
return credentials.strip()
|
||||
return ""
|
||||
|
||||
|
||||
class GatewayService(BaseService):
|
||||
default_enabled = True
|
||||
min_interval = 5
|
||||
title = "OpenAI Gateway"
|
||||
description = (
|
||||
"An OpenAI-compatible LLM endpoint at /openai/v1/* that forwards requests to "
|
||||
"the configured upstream (DeepSeek by default). Image content is described by "
|
||||
"a vision model first so vision-less upstreams still work. Access is gated by "
|
||||
"a static key or DevPlace credentials, and throughput scales with the "
|
||||
"instances setting."
|
||||
)
|
||||
config_fields = [
|
||||
ConfigField("gateway_upstream_url", "Upstream URL", type="url", default=config.UPSTREAM_URL_DEFAULT,
|
||||
help="OpenAI-compatible chat-completions endpoint requests are forwarded to.",
|
||||
group="Upstream"),
|
||||
ConfigField("gateway_model", "Model", type="str", default=config.MODEL_DEFAULT,
|
||||
help="Model sent upstream.", group="Upstream"),
|
||||
ConfigField("gateway_force_model", "Force model", type="bool", default=True,
|
||||
help="Override the client-requested model with the configured model.", group="Upstream"),
|
||||
ConfigField("gateway_api_key", "Upstream API key", type="str", default="",
|
||||
help="The key currently in use; auto-migrated from DEEPSEEK_API_KEY or OPENROUTER_API_KEY on boot. Editable.",
|
||||
group="Upstream"),
|
||||
ConfigField("gateway_timeout", "Upstream timeout (seconds)", type="int", default=config.TIMEOUT_DEFAULT, minimum=1,
|
||||
help="Per-request upstream timeout.", group="Upstream"),
|
||||
ConfigField("gateway_instances", "Instances (concurrency)", type="int",
|
||||
default=config.INSTANCES_DEFAULT, minimum=1, maximum=64,
|
||||
help="Max concurrent upstream forwards per worker (connection pool + semaphore).",
|
||||
group="Upstream"),
|
||||
ConfigField("gateway_vision_enabled", "Vision augmentation", type="bool", default=True,
|
||||
help="Describe image content via the vision model before forwarding.", group="Vision"),
|
||||
ConfigField("gateway_vision_url", "Vision URL", type="url", default=config.VISION_URL_DEFAULT,
|
||||
help="OpenAI-compatible endpoint used to describe images.", group="Vision"),
|
||||
ConfigField("gateway_vision_model", "Vision model", type="str", default=config.VISION_MODEL_DEFAULT,
|
||||
help="Vision-capable model name.", group="Vision"),
|
||||
ConfigField("gateway_vision_key", "Vision API key", type="str", default="",
|
||||
help="The key currently in use; auto-migrated from OPENROUTER_API_KEY on boot. Editable.",
|
||||
group="Vision"),
|
||||
ConfigField("gateway_vision_cache_size", "Vision cache size", type="int",
|
||||
default=config.VISION_CACHE_SIZE_DEFAULT, minimum=0,
|
||||
help="Image-description LRU cache entries (0 disables caching).", group="Vision"),
|
||||
ConfigField("gateway_require_auth", "Require authentication", type="bool", default=True,
|
||||
help="When off, the gateway is open to anyone.", group="Access"),
|
||||
ConfigField("gateway_allow_admins", "Allow admins", type="bool", default=True,
|
||||
help="Admin users (API key / Bearer / Basic / session) may call the gateway.", group="Access"),
|
||||
ConfigField("gateway_allow_users", "Allow users", type="bool", default=False,
|
||||
help="Any authenticated user may call the gateway.", group="Access"),
|
||||
ConfigField("gateway_access_key", "Static access key", type="password", default="", secret=True,
|
||||
help="A standalone key that always grants access (sent as X-API-KEY or Bearer).",
|
||||
group="Access"),
|
||||
ConfigField("gateway_internal_key", "Internal key", type="str", default="",
|
||||
help="Auto-generated on boot. DevPlace's own services authenticate to the gateway "
|
||||
"with this key. Clear it and restart to rotate.",
|
||||
group="Access"),
|
||||
ConfigField("gateway_price_cache_hit_per_m", "Chat price cache-hit / 1M ($)", type="float",
|
||||
default=config.PRICE_CACHE_HIT_PER_M_DEFAULT, minimum=0,
|
||||
help="Estimates chat cost when the upstream returns no native cost field (DeepSeek).",
|
||||
group="Pricing"),
|
||||
ConfigField("gateway_price_cache_miss_per_m", "Chat price cache-miss / 1M ($)", type="float",
|
||||
default=config.PRICE_CACHE_MISS_PER_M_DEFAULT, minimum=0, group="Pricing"),
|
||||
ConfigField("gateway_price_output_per_m", "Chat price output / 1M ($)", type="float",
|
||||
default=config.PRICE_OUTPUT_PER_M_DEFAULT, minimum=0, group="Pricing"),
|
||||
ConfigField("gateway_vision_price_input_per_m", "Vision price input / 1M ($)", type="float",
|
||||
default=config.VISION_PRICE_INPUT_PER_M_DEFAULT, minimum=0,
|
||||
help="Fallback only; used when the vision upstream returns no native cost.",
|
||||
group="Pricing"),
|
||||
ConfigField("gateway_vision_price_output_per_m", "Vision price output / 1M ($)", type="float",
|
||||
default=config.VISION_PRICE_OUTPUT_PER_M_DEFAULT, minimum=0, group="Pricing"),
|
||||
ConfigField("gateway_max_retries", "Max retries", type="int", default=config.MAX_RETRIES_DEFAULT,
|
||||
minimum=0, maximum=10,
|
||||
help="Retry attempts on timeout, connection error, or upstream 5xx.", group="Reliability"),
|
||||
ConfigField("gateway_retry_backoff_ms", "Retry backoff (ms)", type="int",
|
||||
default=config.RETRY_BACKOFF_MS_DEFAULT, minimum=0,
|
||||
help="Linear backoff multiplied by the attempt number.", group="Reliability"),
|
||||
ConfigField("gateway_circuit_threshold", "Circuit breaker threshold", type="int",
|
||||
default=config.CIRCUIT_THRESHOLD_DEFAULT, minimum=0,
|
||||
help="Consecutive upstream failures before the breaker opens (0 disables).", group="Reliability"),
|
||||
ConfigField("gateway_circuit_cooldown_seconds", "Circuit breaker cooldown (s)", type="int",
|
||||
default=config.CIRCUIT_COOLDOWN_SECONDS_DEFAULT, minimum=1, group="Reliability"),
|
||||
ConfigField("gateway_usage_retention_hours", "Usage retention (hours)", type="int",
|
||||
default=config.USAGE_RETENTION_HOURS_DEFAULT, minimum=1,
|
||||
help="How long per-call usage rows are kept before pruning.", group="Tracking"),
|
||||
ConfigField("gateway_model_context_map", "Model context map (JSON)", type="str",
|
||||
default=json.dumps(config.MODEL_CONTEXT_MAP_DEFAULT),
|
||||
help="JSON object mapping model name to max context tokens for utilization tracking.",
|
||||
group="Tracking"),
|
||||
]
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(name="openai", interval_seconds=30)
|
||||
self._runtime = None
|
||||
|
||||
def runtime(self) -> GatewayRuntime:
|
||||
if self._runtime is None:
|
||||
self._runtime = GatewayRuntime()
|
||||
return self._runtime
|
||||
|
||||
def effective_config(self) -> dict:
|
||||
cfg = self.get_config()
|
||||
cfg["gateway_api_key"] = (cfg["gateway_api_key"]
|
||||
or os.environ.get("DEEPSEEK_API_KEY", "")
|
||||
or os.environ.get("OPENROUTER_API_KEY", ""))
|
||||
cfg["gateway_vision_key"] = (cfg["gateway_vision_key"]
|
||||
or os.environ.get("OPENROUTER_API_KEY", ""))
|
||||
return cfg
|
||||
|
||||
def authorize(self, request: Request) -> bool:
|
||||
cfg = self.get_config()
|
||||
if not cfg["gateway_require_auth"]:
|
||||
return True
|
||||
access_key = cfg["gateway_access_key"]
|
||||
internal_key = cfg["gateway_internal_key"]
|
||||
presented = _presented_key(request)
|
||||
if presented and access_key and presented == access_key:
|
||||
return True
|
||||
if presented and internal_key and presented == internal_key:
|
||||
return True
|
||||
user = get_current_user(request)
|
||||
if user:
|
||||
if user.get("role") == "Admin" and cfg["gateway_allow_admins"]:
|
||||
return True
|
||||
if cfg["gateway_allow_users"]:
|
||||
return True
|
||||
return False
|
||||
|
||||
def resolve_owner(self, request: Request) -> tuple:
|
||||
cfg = self.get_config()
|
||||
presented = _presented_key(request)
|
||||
if presented and cfg["gateway_internal_key"] and presented == cfg["gateway_internal_key"]:
|
||||
return ("internal", "devii")
|
||||
if presented and cfg["gateway_access_key"] and presented == cfg["gateway_access_key"]:
|
||||
return ("key", "access")
|
||||
user = get_current_user(request)
|
||||
if user:
|
||||
kind = "admin" if user.get("role") == "Admin" else "user"
|
||||
return (kind, user.get("uid") or "unknown")
|
||||
return ("anonymous", "anonymous")
|
||||
|
||||
async def handle(self, request: Request, subpath: str):
|
||||
if not self.is_enabled():
|
||||
raise HTTPException(status_code=503, detail="Gateway is disabled")
|
||||
if not self.authorize(request):
|
||||
self.log(f"Rejected {request.method} /{subpath}: unauthorized")
|
||||
raise HTTPException(status_code=401, detail="Unauthorized")
|
||||
cfg = self.effective_config()
|
||||
runtime = self.runtime()
|
||||
owner = self.resolve_owner(request)
|
||||
user_agent = request.headers.get("user-agent", "")
|
||||
if subpath == "chat/completions" and request.method == "POST":
|
||||
try:
|
||||
body = await request.json()
|
||||
except Exception:
|
||||
self.log("Rejected chat request: invalid JSON body")
|
||||
raise HTTPException(status_code=400, detail="Invalid JSON body")
|
||||
if not isinstance(body, dict):
|
||||
self.log("Rejected chat request: JSON body was not an object")
|
||||
raise HTTPException(status_code=400, detail="Invalid JSON body")
|
||||
return await runtime.handle_chat(body, cfg, owner, user_agent, self.log)
|
||||
body = await request.body()
|
||||
content_type = request.headers.get("content-type", "")
|
||||
return await runtime.handle_passthrough(request.method, subpath, content_type, body, cfg, owner, user_agent, self.log)
|
||||
|
||||
async def run_once(self) -> None:
|
||||
if not self.is_enabled():
|
||||
return
|
||||
runtime = self.runtime()
|
||||
runtime._ensure(self.effective_config())
|
||||
runtime._ledger.sample_concurrency(runtime.in_flight)
|
||||
retention = get_int_setting("gateway_usage_retention_hours", config.USAGE_RETENTION_HOURS_DEFAULT)
|
||||
ledger_removed, samples_removed = runtime._ledger.prune(retention)
|
||||
if ledger_removed or samples_removed:
|
||||
self.log(f"Pruned {ledger_removed} usage rows and {samples_removed} concurrency samples")
|
||||
|
||||
async def on_disable(self) -> None:
|
||||
if self._runtime is not None:
|
||||
await self._runtime.aclose()
|
||||
|
||||
def collect_metrics(self) -> dict:
|
||||
cfg = self.get_config()
|
||||
m = self._runtime.metrics() if self._runtime is not None else {
|
||||
"requests": 0, "errors": 0, "in_flight": 0, "peak_in_flight": 0, "vision_calls": 0,
|
||||
"last_status": 0, "last_latency_ms": 0, "pool": 0, "circuit_open": False,
|
||||
}
|
||||
s = summary_metrics()
|
||||
stats = [
|
||||
{"label": "Requests (lifetime)", "value": m["requests"]},
|
||||
{"label": "In flight", "value": m["in_flight"]},
|
||||
{"label": "Vision calls", "value": m["vision_calls"]},
|
||||
{"label": "Last status", "value": m["last_status"] or "-"},
|
||||
{"label": "Last latency", "value": f"{m['last_latency_ms']} ms"},
|
||||
{"label": "Pool size", "value": m["pool"]},
|
||||
{"label": "Circuit", "value": "open" if m["circuit_open"] else "closed"},
|
||||
{"label": "Model", "value": cfg["gateway_model"]},
|
||||
{"label": "Requests 24h", "value": s["requests"]},
|
||||
{"label": "Success 24h", "value": f"{s['success_pct']}%"},
|
||||
{"label": "Error rate 24h", "value": f"{s['error_pct']}%"},
|
||||
{"label": "Cost this hour", "value": f"${s['cost_hour']:.4f}"},
|
||||
{"label": "Cost 24h", "value": f"${s['cost_24h']:.2f}"},
|
||||
{"label": "Tokens 24h", "value": s["tokens_24h"]},
|
||||
{"label": "Avg latency 24h", "value": f"{s['avg_latency_ms']:.0f} ms"},
|
||||
{"label": "Avg tokens/s 24h", "value": s["avg_tps"]},
|
||||
{"label": "Peak concurrency 24h", "value": s["peak_concurrency"]},
|
||||
{"label": "Top model 24h", "value": s["top_model"]},
|
||||
{"label": "Top caller 24h", "value": s["top_caller"]},
|
||||
]
|
||||
return {"stats": stats}
|
||||
@@ -0,0 +1,227 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Any, Optional
|
||||
|
||||
import httpx
|
||||
|
||||
from devplacepy.database import db, get_table
|
||||
from devplacepy.services.openai_gateway import config
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
GATEWAY_LEDGER = "gateway_usage_ledger"
|
||||
GATEWAY_CONCURRENCY = "gateway_concurrency_samples"
|
||||
PER_MILLION = 1_000_000
|
||||
|
||||
|
||||
def _now() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def _iso(moment: datetime) -> str:
|
||||
return moment.isoformat()
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Pricing:
|
||||
chat_cache_hit_per_m: float
|
||||
chat_cache_miss_per_m: float
|
||||
chat_output_per_m: float
|
||||
vision_input_per_m: float
|
||||
vision_output_per_m: float
|
||||
|
||||
|
||||
def pricing_from_cfg(cfg: dict) -> Pricing:
|
||||
return Pricing(
|
||||
chat_cache_hit_per_m=float(cfg.get("gateway_price_cache_hit_per_m", config.PRICE_CACHE_HIT_PER_M_DEFAULT)),
|
||||
chat_cache_miss_per_m=float(cfg.get("gateway_price_cache_miss_per_m", config.PRICE_CACHE_MISS_PER_M_DEFAULT)),
|
||||
chat_output_per_m=float(cfg.get("gateway_price_output_per_m", config.PRICE_OUTPUT_PER_M_DEFAULT)),
|
||||
vision_input_per_m=float(cfg.get("gateway_vision_price_input_per_m", config.VISION_PRICE_INPUT_PER_M_DEFAULT)),
|
||||
vision_output_per_m=float(cfg.get("gateway_vision_price_output_per_m", config.VISION_PRICE_OUTPUT_PER_M_DEFAULT)),
|
||||
)
|
||||
|
||||
|
||||
def parse_context_map(raw: Any) -> dict[str, int]:
|
||||
if isinstance(raw, dict):
|
||||
return {str(k): int(v) for k, v in raw.items()}
|
||||
if not raw:
|
||||
return dict(config.MODEL_CONTEXT_MAP_DEFAULT)
|
||||
try:
|
||||
loaded = json.loads(raw)
|
||||
if isinstance(loaded, dict):
|
||||
return {str(k): int(v) for k, v in loaded.items()}
|
||||
except (ValueError, TypeError):
|
||||
logger.warning("Invalid gateway_model_context_map, using defaults")
|
||||
return dict(config.MODEL_CONTEXT_MAP_DEFAULT)
|
||||
|
||||
|
||||
def normalize_usage(usage: Optional[dict]) -> dict:
|
||||
usage = usage or {}
|
||||
prompt = int(usage.get("prompt_tokens", 0) or 0)
|
||||
completion = int(usage.get("completion_tokens", 0) or 0)
|
||||
total = int(usage.get("total_tokens", prompt + completion) or 0)
|
||||
|
||||
hit = usage.get("prompt_cache_hit_tokens")
|
||||
if hit is None:
|
||||
details = usage.get("prompt_tokens_details") or {}
|
||||
hit = details.get("cached_tokens", 0)
|
||||
hit = int(hit or 0)
|
||||
|
||||
miss = usage.get("prompt_cache_miss_tokens")
|
||||
if miss is None:
|
||||
miss = max(prompt - hit, 0)
|
||||
miss = int(miss or 0)
|
||||
|
||||
completion_details = usage.get("completion_tokens_details") or {}
|
||||
reasoning = int(completion_details.get("reasoning_tokens", 0) or 0)
|
||||
return {
|
||||
"prompt": prompt,
|
||||
"completion": completion,
|
||||
"total": total,
|
||||
"cache_hit": hit,
|
||||
"cache_miss": miss,
|
||||
"reasoning": reasoning,
|
||||
}
|
||||
|
||||
|
||||
def compute_cost(usage: dict, norm: dict, pricing: Pricing, backend: str) -> tuple[float, float, float, bool]:
|
||||
native = usage.get("cost") if isinstance(usage, dict) else None
|
||||
if isinstance(native, (int, float)) and not isinstance(native, bool):
|
||||
total = float(native)
|
||||
denom = norm["prompt"] + norm["completion"]
|
||||
input_cost = total * norm["prompt"] / denom if denom > 0 else 0.0
|
||||
return total, input_cost, total - input_cost, True
|
||||
if backend == "vision":
|
||||
input_cost = norm["prompt"] / PER_MILLION * pricing.vision_input_per_m
|
||||
output_cost = norm["completion"] / PER_MILLION * pricing.vision_output_per_m
|
||||
return input_cost + output_cost, input_cost, output_cost, False
|
||||
input_cost = (norm["cache_hit"] / PER_MILLION * pricing.chat_cache_hit_per_m
|
||||
+ norm["cache_miss"] / PER_MILLION * pricing.chat_cache_miss_per_m)
|
||||
output_cost = norm["completion"] / PER_MILLION * pricing.chat_output_per_m
|
||||
return input_cost + output_cost, input_cost, output_cost, False
|
||||
|
||||
|
||||
def context_utilization(total_tokens: int, model: str, context_map: dict) -> tuple[Optional[int], Optional[float]]:
|
||||
window = context_map.get(model)
|
||||
if not window or window <= 0:
|
||||
return None, None
|
||||
return int(window), round(total_tokens / window, 4)
|
||||
|
||||
|
||||
def extract_params(body: Any) -> dict:
|
||||
if not isinstance(body, dict):
|
||||
return {"requested_model": "", "stream_requested": False, "temperature": None,
|
||||
"top_p": None, "max_tokens": None, "has_tools": False}
|
||||
temperature = body.get("temperature")
|
||||
top_p = body.get("top_p")
|
||||
max_tokens = body.get("max_tokens")
|
||||
if max_tokens is None:
|
||||
max_tokens = body.get("max_completion_tokens")
|
||||
return {
|
||||
"requested_model": body.get("model") or "",
|
||||
"stream_requested": bool(body.get("stream")),
|
||||
"temperature": float(temperature) if isinstance(temperature, (int, float)) and not isinstance(temperature, bool) else None,
|
||||
"top_p": float(top_p) if isinstance(top_p, (int, float)) and not isinstance(top_p, bool) else None,
|
||||
"max_tokens": int(max_tokens) if isinstance(max_tokens, (int, float)) and not isinstance(max_tokens, bool) else None,
|
||||
"has_tools": bool(body.get("tools") or body.get("functions")),
|
||||
}
|
||||
|
||||
|
||||
def classify_error(status_code: int, exc: Optional[Exception] = None, message: str = "") -> str:
|
||||
if exc is not None:
|
||||
if isinstance(exc, httpx.TimeoutException):
|
||||
return "timeout"
|
||||
return "gateway"
|
||||
if status_code == 429:
|
||||
return "rate_limit"
|
||||
if status_code in (401, 403):
|
||||
return "auth"
|
||||
if status_code == 404:
|
||||
return "model_not_found"
|
||||
if status_code in (400, 422):
|
||||
lowered = (message or "").lower()
|
||||
if "context" in lowered or "maximum" in lowered or "too long" in lowered:
|
||||
return "context_length"
|
||||
return "bad_request"
|
||||
if status_code and status_code >= 500:
|
||||
return "upstream_error"
|
||||
return "gateway"
|
||||
|
||||
|
||||
class GatewayUsageLedger:
|
||||
def record(self, raw: dict, pricing: Pricing, context_map: dict) -> None:
|
||||
try:
|
||||
usage = raw.get("usage") or {}
|
||||
norm = normalize_usage(usage)
|
||||
cost_usd, input_cost, output_cost, native = compute_cost(usage, norm, pricing, raw["backend"])
|
||||
window, util = context_utilization(norm["total"], raw.get("model") or "", context_map)
|
||||
upstream_ms = float(raw.get("upstream_latency_ms") or 0)
|
||||
completion = norm["completion"]
|
||||
tps = completion / (upstream_ms / 1000.0) if upstream_ms > 0 and completion else 0.0
|
||||
row = {
|
||||
"created_at": _iso(_now()),
|
||||
"owner_kind": raw.get("owner_kind") or "unknown",
|
||||
"owner_id": raw.get("owner_id") or "unknown",
|
||||
"backend": raw["backend"],
|
||||
"endpoint": raw.get("endpoint") or "",
|
||||
"requested_model": raw.get("requested_model") or "",
|
||||
"model": raw.get("model") or "",
|
||||
"status_code": int(raw.get("status_code") or 0),
|
||||
"success": 1 if raw.get("success") else 0,
|
||||
"error_category": raw.get("error_category"),
|
||||
"upstream_latency_ms": upstream_ms,
|
||||
"gateway_overhead_ms": float(raw.get("gateway_overhead_ms") or 0),
|
||||
"queue_wait_ms": float(raw.get("queue_wait_ms") or 0),
|
||||
"connect_ms": float(raw.get("connect_ms") or 0),
|
||||
"total_latency_ms": float(raw.get("total_latency_ms") or 0),
|
||||
"prompt_tokens": norm["prompt"],
|
||||
"completion_tokens": norm["completion"],
|
||||
"cache_hit_tokens": norm["cache_hit"],
|
||||
"cache_miss_tokens": norm["cache_miss"],
|
||||
"reasoning_tokens": norm["reasoning"],
|
||||
"total_tokens": norm["total"],
|
||||
"tokens_per_second": round(tps, 3),
|
||||
"context_window": window,
|
||||
"context_utilization": util,
|
||||
"cost_usd": round(cost_usd, 8),
|
||||
"input_cost_usd": round(input_cost, 8),
|
||||
"output_cost_usd": round(output_cost, 8),
|
||||
"native_cost": 1 if native else 0,
|
||||
"stream_requested": 1 if raw.get("stream_requested") else 0,
|
||||
"temperature": raw.get("temperature"),
|
||||
"top_p": raw.get("top_p"),
|
||||
"max_tokens": raw.get("max_tokens"),
|
||||
"has_tools": 1 if raw.get("has_tools") else 0,
|
||||
"retries_attempted": int(raw.get("retries_attempted") or 0),
|
||||
"retry_succeeded": 1 if raw.get("retry_succeeded") else 0,
|
||||
"circuit_open": 1 if raw.get("circuit_open") else 0,
|
||||
"user_agent": (raw.get("user_agent") or "")[:300],
|
||||
}
|
||||
get_table(GATEWAY_LEDGER).insert(row)
|
||||
except Exception as exc:
|
||||
logger.warning("gateway usage record failed: %s", exc)
|
||||
|
||||
def sample_concurrency(self, in_flight: int) -> None:
|
||||
try:
|
||||
get_table(GATEWAY_CONCURRENCY).insert({
|
||||
"created_at": _iso(_now()),
|
||||
"in_flight": int(in_flight),
|
||||
})
|
||||
except Exception as exc:
|
||||
logger.warning("gateway concurrency sample failed: %s", exc)
|
||||
|
||||
def prune(self, older_than_hours: int) -> tuple[int, int]:
|
||||
cutoff = _iso(_now() - timedelta(hours=max(1, older_than_hours)))
|
||||
ledger_removed = 0
|
||||
samples_removed = 0
|
||||
if GATEWAY_LEDGER in db.tables:
|
||||
ledger_removed = int(get_table(GATEWAY_LEDGER).delete(created_at={"<": cutoff}))
|
||||
if GATEWAY_CONCURRENCY in db.tables:
|
||||
samples_removed = int(get_table(GATEWAY_CONCURRENCY).delete(created_at={"<": cutoff}))
|
||||
return ledger_removed, samples_removed
|
||||
@@ -0,0 +1,180 @@
|
||||
import asyncio
|
||||
import hashlib
|
||||
import logging
|
||||
import time
|
||||
from collections import OrderedDict
|
||||
from typing import Any, Optional
|
||||
|
||||
import httpx
|
||||
|
||||
from devplacepy.services.openai_gateway.config import VISION_INSTRUCTION
|
||||
from devplacepy.services.openai_gateway.usage import classify_error
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class VisionCache:
|
||||
def __init__(self, size: int = 256):
|
||||
self.size = size
|
||||
self._store: "OrderedDict[str, str]" = OrderedDict()
|
||||
|
||||
def get(self, key: str) -> Optional[str]:
|
||||
if self.size <= 0:
|
||||
return None
|
||||
value = self._store.get(key)
|
||||
if value is not None:
|
||||
self._store.move_to_end(key)
|
||||
return value
|
||||
|
||||
def put(self, key: str, value: str) -> None:
|
||||
if self.size <= 0:
|
||||
return
|
||||
self._store[key] = value
|
||||
self._store.move_to_end(key)
|
||||
while len(self._store) > self.size:
|
||||
self._store.popitem(last=False)
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self._store)
|
||||
|
||||
|
||||
def has_vision_blocks(content: Any) -> bool:
|
||||
if not isinstance(content, list):
|
||||
return False
|
||||
return any(
|
||||
isinstance(b, dict) and b.get("type") in ("image_url", "image")
|
||||
for b in content
|
||||
)
|
||||
|
||||
|
||||
def split_text_and_images(content: list) -> tuple[str, list]:
|
||||
texts: list[str] = []
|
||||
images: list[dict] = []
|
||||
for block in content:
|
||||
if not isinstance(block, dict):
|
||||
continue
|
||||
btype = block.get("type")
|
||||
if btype == "text":
|
||||
t = block.get("text", "")
|
||||
if t:
|
||||
texts.append(t)
|
||||
elif btype in ("image_url", "image"):
|
||||
images.append(block)
|
||||
return "\n".join(texts).strip(), images
|
||||
|
||||
|
||||
def _image_cache_key(image_block: dict) -> str:
|
||||
iu = image_block.get("image_url") or image_block.get("image") or ""
|
||||
url = iu.get("url", "") if isinstance(iu, dict) else str(iu)
|
||||
return hashlib.sha256(url.encode("utf-8", errors="replace")).hexdigest()[:32]
|
||||
|
||||
|
||||
def _format_vision_block(descriptions: list) -> str:
|
||||
if len(descriptions) == 1:
|
||||
return f"[Image seen by vision model:\n{descriptions[0]}\n]"
|
||||
parts = [f"Image {i}:\n{d}" for i, d in enumerate(descriptions, 1)]
|
||||
return f"[{len(descriptions)} images seen by vision model:\n" + "\n\n".join(parts) + "\n]"
|
||||
|
||||
|
||||
class VisionAugmenter:
|
||||
def __init__(self, vision_url: str, vision_model: str, vision_key: str,
|
||||
cache: VisionCache, referer: str = "", title: str = "",
|
||||
ledger=None, owner: tuple = ("unknown", "unknown"), pricing=None, context_map=None):
|
||||
self.vision_url = vision_url
|
||||
self.vision_model = vision_model
|
||||
self.vision_key = vision_key
|
||||
self.cache = cache
|
||||
self.referer = referer
|
||||
self.title = title
|
||||
self.ledger = ledger
|
||||
self.owner = owner
|
||||
self.pricing = pricing
|
||||
self.context_map = context_map or {}
|
||||
self.calls = 0
|
||||
|
||||
def _record(self, latency_ms, status_code, success, category, usage):
|
||||
if self.ledger is None or self.pricing is None:
|
||||
return
|
||||
self.ledger.record({
|
||||
"owner_kind": self.owner[0], "owner_id": self.owner[1], "backend": "vision",
|
||||
"endpoint": "chat/completions", "model": self.vision_model,
|
||||
"requested_model": self.vision_model, "temperature": 0.2,
|
||||
"upstream_latency_ms": latency_ms, "total_latency_ms": latency_ms,
|
||||
"status_code": status_code, "success": success, "error_category": category,
|
||||
"usage": usage,
|
||||
}, self.pricing, self.context_map)
|
||||
|
||||
async def _describe_one(self, client: httpx.AsyncClient, image_block: dict) -> str:
|
||||
if not self.vision_key:
|
||||
return "[vision unavailable: vision API key not configured]"
|
||||
payload = {
|
||||
"model": self.vision_model,
|
||||
"messages": [{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "text", "text": VISION_INSTRUCTION},
|
||||
image_block,
|
||||
],
|
||||
}],
|
||||
"temperature": 0.2,
|
||||
"stream": False,
|
||||
}
|
||||
headers = {
|
||||
"Authorization": f"Bearer {self.vision_key}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
if self.referer:
|
||||
headers["HTTP-Referer"] = self.referer
|
||||
if self.title:
|
||||
headers["X-Title"] = self.title
|
||||
start = time.monotonic()
|
||||
try:
|
||||
resp = await client.post(self.vision_url, json=payload, headers=headers, timeout=120.0)
|
||||
except httpx.RequestError as e:
|
||||
logger.warning("vision connection failed: %s", e)
|
||||
self._record((time.monotonic() - start) * 1000, 502, False, classify_error(0, e), None)
|
||||
return f"[vision call failed: {e}]"
|
||||
latency_ms = (time.monotonic() - start) * 1000
|
||||
if resp.status_code != 200:
|
||||
logger.warning("vision %s: %s", resp.status_code, resp.text[:200])
|
||||
self._record(latency_ms, resp.status_code, False, classify_error(resp.status_code, None, resp.text), None)
|
||||
return f"[vision failed: HTTP {resp.status_code}]"
|
||||
try:
|
||||
data = resp.json()
|
||||
text = data["choices"][0]["message"].get("content") or ""
|
||||
self._record(latency_ms, 200, True, None, data.get("usage"))
|
||||
return text.strip() or "[vision returned empty response]"
|
||||
except (KeyError, IndexError, ValueError) as e:
|
||||
self._record(latency_ms, 200, False, "gateway", None)
|
||||
return f"[vision parse error: {e}]"
|
||||
|
||||
async def _describe_images(self, client: httpx.AsyncClient, images: list) -> list:
|
||||
keys = [_image_cache_key(img) for img in images]
|
||||
descriptions: list = [self.cache.get(k) for k in keys]
|
||||
miss = [i for i, d in enumerate(descriptions) if d is None]
|
||||
if miss:
|
||||
self.calls += len(miss)
|
||||
results = await asyncio.gather(*[self._describe_one(client, images[i]) for i in miss])
|
||||
for idx, result in zip(miss, results):
|
||||
descriptions[idx] = result
|
||||
if not result.startswith("[vision"):
|
||||
self.cache.put(keys[idx], result)
|
||||
return [d or "" for d in descriptions]
|
||||
|
||||
async def augment_messages(self, client: httpx.AsyncClient, messages: list) -> list:
|
||||
if not any(has_vision_blocks(m.get("content")) for m in messages if isinstance(m, dict)):
|
||||
return messages
|
||||
out: list = []
|
||||
for m in messages:
|
||||
content = m.get("content") if isinstance(m, dict) else None
|
||||
if not has_vision_blocks(content):
|
||||
out.append(m)
|
||||
continue
|
||||
user_text, images = split_text_and_images(content)
|
||||
descriptions = await self._describe_images(client, images)
|
||||
vision_block = _format_vision_block(descriptions)
|
||||
merged = f"{user_text}\n\n{vision_block}" if user_text else vision_block
|
||||
new_msg = dict(m)
|
||||
new_msg["content"] = merged
|
||||
out.append(new_msg)
|
||||
return out
|
||||
Reference in New Issue
Block a user