Compare commits

..
Author SHA1 Message Date
Shroominic ea655b748b optimize startup time 2025-12-26 12:33:42 +01:00
9qeklajc 2d247ddc8b fmt 2025-12-26 10:29:51 +01:00
9qeklajc c064452aea Merge branch 'v0.2.1' into force-sats-pricing 2025-12-26 10:27:11 +01:00
9qeklajcandGitHub 1c6a603042 Merge pull request #271 from Routstr/embedding-with-aliases
Embedding with aliases
2025-12-26 10:26:38 +01:00
9qeklajc 84b0007b05 force sats pricig 2025-12-26 09:55:15 +01:00
Shroominic 525476ccfa fix alias not displaying correctly 2025-12-25 22:34:52 +01:00
Shroominic b54812cb04 fix: alias_id not displayed correctly by dashboard 2025-12-25 22:28:26 +01:00
Shroominic ee508cbb3a rm model cleanup 2025-12-25 21:50:09 +01:00
Shroominic 0c61fdee07 ruff format 2025-12-25 21:48:20 +01:00
Shroominic b9418db31f rm model cleanup task 2025-12-25 21:47:17 +01:00
Shroominic 71e7c2171b add missing migration 2025-12-25 19:36:26 +01:00
Shroominic a3e8d5fd38 ignore cloudflare headers from logs 2025-12-24 15:09:03 +01:00
9qeklajc 5b8e56f590 Merge branch 'v0.2.1' into embedding-with-aliases
# Conflicts:
#	routstr/upstream/base.py
2025-12-23 23:43:46 +01:00
9qeklajc a6d0bd1a19 update aliases 2025-12-23 23:43:16 +01:00
9qeklajc b01c7b2e56 Merge branch 'v0.2.1' into embeddings 2025-12-22 21:04:28 +01:00
9qeklajc 8df0c17bc3 embedding integration 2025-12-03 22:20:31 +01:00
9qeklajc 7bc9ee0653 Merge branch 'gemini-upstream' into embeddings 2025-12-03 22:01:39 +01:00
9qeklajc 355f8601c1 embedding integration 2025-12-03 21:58:31 +01:00
27 changed files with 242 additions and 492 deletions
+37
View File
@@ -0,0 +1,37 @@
import os
import openai
client = openai.OpenAI(
api_key=os.environ["CASHU_TOKEN"],
base_url=os.environ.get("ROUTSTR_API_URL", "https://api.routstr.com/v1"),
# base_url="http://roustrjfsdgfiueghsklchg.onion/v1",
# client=httpx.AsyncClient(
# proxies={"http": "socks5://localhost:9050"},
# ), # to use onion proxy (tor)
)
history: list = []
def chat() -> None:
while True:
user_msg = {"role": "user", "content": input("\nYou: ")}
history.append(user_msg)
ai_msg = {"role": "assistant", "content": ""}
for chunk in client.chat.completions.create(
model=os.environ.get("MODEL", "openai/gpt-4o-mini"),
messages=history,
stream=True,
):
if len(chunk.choices) > 0:
content = chunk.choices[0].delta.content
if content is not None:
ai_msg["content"] += content
print(content, end="", flush=True)
print()
history.append(ai_msg)
if __name__ == "__main__":
chat()
-11
View File
@@ -1,11 +0,0 @@
import os
import httpx
# Use your Cashu token or API key as the Bearer token,
# cashu token is hashed on the server and acts as an Temporary API key
headers = {"Authorization": f"Bearer {os.environ.get('TOKEN')}"}
base_url = os.environ.get("API_URL", "https://api.routstr.com/v1")
resp = httpx.get(f"{base_url}/balance/info", headers=headers)
print(resp.json())
-15
View File
@@ -1,15 +0,0 @@
import os
import httpx
# Send a Cashu token to the /create endpoint to get a persistent API key
token = os.environ.get("TOKEN")
if not token:
print("Please set TOKEN environment variable with a Cashu token")
exit(1)
base_url = os.environ.get("API_URL", "https://api.routstr.com/v1")
resp = httpx.get(f"{base_url}/balance/create", params={"initial_balance_token": token})
print(resp.json())
-12
View File
@@ -1,12 +0,0 @@
import os
import httpx
# Use your Cashu token or API key as the Bearer token
headers = {"Authorization": f"Bearer {os.environ.get('TOKEN')}"}
base_url = os.environ.get("API_URL", "https://api.routstr.com/v1")
resp = httpx.post(f"{base_url}/balance/refund", headers=headers)
print("Refund successful!")
print(resp.json())
-16
View File
@@ -1,16 +0,0 @@
import os
import httpx
# Use your Cashu token or API key as the Bearer token
headers = {"Authorization": f"Bearer {os.environ.get('TOKEN')}"}
base_url = os.environ.get("API_URL", "https://api.routstr.com/v1")
# The Cashu token to top up with
cashu_token = input("Enter Cashu token to top up: ")
resp = httpx.post(
f"{base_url}/balance/topup", headers=headers, json={"cashu_token": cashu_token}
)
print(resp.json())
-15
View File
@@ -1,15 +0,0 @@
import os
from openai import OpenAI
client = OpenAI(
api_key=os.environ.get("TOKEN"),
base_url=os.environ.get("API_URL", "https://api.routstr.com/v1"),
)
response = client.chat.completions.create(
model=os.environ.get("MODEL", "gpt-5-nano"),
messages=[{"role": "user", "content": "Hello!"}],
)
print(response.choices[0].message.content)
-19
View File
@@ -1,19 +0,0 @@
import os
import httpx
from openai import OpenAI
client = OpenAI(
api_key=os.environ.get("TOKEN", ""),
base_url=os.environ.get("API_URL", "https://api.routstr.com/v1"),
)
for model in client.models.list():
print(model.id)
# OR
models = httpx.get(
f"{client.base_url}/v1/models",
headers={"Authorization": f"Bearer {client.api_key}"},
).json()
-31
View File
@@ -1,31 +0,0 @@
import os
from openai import OpenAI
client = OpenAI(
api_key=os.environ.get("TOKEN"),
base_url=os.environ.get("API_URL", "https://api.routstr.com/v1"),
)
conversation = [] # type: ignore
# First turn
response1 = client.responses.create( # type: ignore
model="o4-mini",
input="Hi, my name is Alice.",
conversation=conversation,
)
print("Response 1:", response1.output)
# Note: The 'conversation' parameter might need to be constructed differently
# depending on exact SDK/API spec. Typically, you pass back the previous turn's data.
# Assuming the SDK manages or returns a conversation object/ID:
# conversation.append(response1)
# Second turn - demonstrating intent, actual implementation depends on strict API spec
# response2 = client.responses.create(
# model="openai/gpt-4o-mini",
# input="What is my name?",
# conversation=conversation,
# )
# print("Response 2:", response2.output)
-17
View File
@@ -1,17 +0,0 @@
import os
from openai import OpenAI
# The OpenAI SDK handles the 'responses' endpoint if it's updated to the latest version
# and the base_url points to a compatible proxy like Routstr.
client = OpenAI(
api_key=os.environ.get("TOKEN"),
base_url=os.environ.get("API_URL", "https://api.routstr.com/v1"),
)
response = client.responses.create(
model="gpt-5-mini",
input="Tell me a three sentence bedtime story about a unicorn.",
)
print(response.output)
-20
View File
@@ -1,20 +0,0 @@
import os
from openai import OpenAI
client = OpenAI(
api_key=os.environ.get("TOKEN"),
base_url=os.environ.get("API_URL", "https://api.routstr.com/v1"),
)
stream = client.responses.create(
model="claude-4.5-sonnet",
input="Write a short poem about rust.",
stream=True,
)
for event in stream:
# Note: Depending on the SDK version and response structure,
# you might access event.output_delta or similar fields
print(event, end="", flush=True)
print()
-16
View File
@@ -1,16 +0,0 @@
import os
from openai import OpenAI
client = OpenAI(
api_key=os.environ.get("TOKEN"),
base_url=os.environ.get("API_URL", "https://api.routstr.com/v1"),
)
response = client.responses.create(
model="gpt-5-mini",
input="What is the latest news about AI?",
tools=[{"type": "web_search"}], # type: ignore
)
print(response.output)
-28
View File
@@ -1,28 +0,0 @@
import os
from openai import OpenAI
client = OpenAI(
api_key=os.environ.get("TOKEN"),
base_url=os.environ.get("API_URL", "https://api.routstr.com/v1"),
)
messages = []
while True:
messages.append({"role": "user", "content": input("\nYou: ")})
stream = client.chat.completions.create(
model=os.environ.get("MODEL", "gpt-5.1-mini"),
messages=messages, # type: ignore
stream=True,
)
print("AI: ", end="")
response_content = ""
for chunk in stream:
if content := chunk.choices[0].delta.content: # type: ignore
print(content, end="", flush=True)
response_content += content
print()
messages.append({"role": "assistant", "content": response_content})
-20
View File
@@ -1,20 +0,0 @@
import os
import httpx
from openai import OpenAI
# Requires `pip install "httpx[socks]"` and a running Tor proxy on port 9050
client = OpenAI(
api_key=os.environ.get("TOKEN"),
base_url=os.environ.get("ONION_URL", "http://roustrjfsdgfiueghsklchg.onion/v1"),
http_client=httpx.Client(proxies="socks5://localhost:9050"),
)
print(
client.chat.completions.create(
model="openai/gpt-4o-mini",
messages=[{"role": "user", "content": "Hello from Tor!"}],
)
.choices[0]
.message.content
)
@@ -0,0 +1,37 @@
"""alias-ids
Revision ID: b9667ffc5701
Revises: lightning_invoices
Create Date: 2025-12-25 19:30:44.673350
"""
import sqlalchemy as sa
import sqlmodel
from alembic import op
# revision identifiers, used by Alembic.
revision = "b9667ffc5701"
down_revision = "lightning_invoices"
branch_labels = None
depends_on = None
def upgrade() -> None:
# ### commands auto generated by Alembic ###
op.add_column(
"models",
sa.Column("canonical_slug", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
)
op.add_column(
"models",
sa.Column("alias_ids", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_column("models", "alias_ids")
op.drop_column("models", "canonical_slug")
# ### end Alembic commands ###
-1
View File
@@ -73,7 +73,6 @@ packages = ["routstr"]
[tool.ruff.lint]
select = ["E", "F", "I"]
ignore = ["E501"]
exclude = ["examples"]
[tool.mypy]
python_version = "3.11"
+7 -6
View File
@@ -206,19 +206,20 @@ def create_model_mappings(
alias: str, model: "Model", provider: "BaseUpstreamProvider"
) -> None:
"""Set alias to model/provider if not set or if new model is preferred."""
existing_model = model_instances.get(alias)
alias_lower = alias.lower()
existing_model = model_instances.get(alias_lower)
if not existing_model:
# No existing mapping, set it
model_instances[alias] = model
provider_map[alias] = provider
model_instances[alias_lower] = model
provider_map[alias_lower] = provider
else:
# Check if candidate should replace existing
existing_provider = provider_map[alias]
existing_provider = provider_map[alias_lower]
if should_prefer_model(
model, provider, existing_model, existing_provider, alias
):
model_instances[alias] = model
provider_map[alias] = provider
model_instances[alias_lower] = model
provider_map[alias_lower] = provider
def process_provider_models(
upstream: "BaseUpstreamProvider", is_openrouter: bool = False
+78 -110
View File
@@ -1493,20 +1493,8 @@ class ModelCreate(BaseModel):
per_request_limits: dict[str, object] | None = None
top_provider: dict[str, object] | None = None
upstream_provider_id: int | None = None
enabled: bool = True
class ModelUpdate(BaseModel):
id: str
name: str
description: str
created: int
context_length: int
architecture: dict[str, object]
pricing: dict[str, object]
per_request_limits: dict[str, object] | None = None
top_provider: dict[str, object] | None = None
upstream_provider_id: int | None = None
canonical_slug: str | None = None
alias_ids: list[str] | None = None
enabled: bool = True
@@ -2416,44 +2404,80 @@ async def admin_upstream_providers(request: Request) -> str:
"/api/upstream-providers/{provider_id}/models",
dependencies=[Depends(require_admin_api)],
)
async def create_provider_model(
async def upsert_provider_model(
provider_id: int, payload: ModelCreate
) -> dict[str, object]:
print(payload)
logger.info(
f"UPSERT_PROVIDER_MODEL called: provider_id={provider_id}, model_id={payload.id}"
)
async with create_session() as session:
provider = await session.get(UpstreamProviderRow, provider_id)
if not provider:
raise HTTPException(status_code=404, detail="Provider not found")
exists = await session.get(ModelRow, (payload.id, provider_id))
if exists:
raise HTTPException(
status_code=409,
detail="Model with this ID already exists for this provider",
)
# Try to get existing model
existing_row = await session.get(ModelRow, (payload.id, provider_id))
row = ModelRow(
id=payload.id,
name=payload.name,
description=payload.description,
created=int(payload.created),
context_length=int(payload.context_length),
architecture=json.dumps(payload.architecture),
pricing=json.dumps(payload.pricing),
sats_pricing=None,
per_request_limits=(
if existing_row:
# Update existing model
logger.info(f"Updating existing model: {payload.id}")
existing_row.name = payload.name
existing_row.description = payload.description
existing_row.created = int(payload.created)
existing_row.context_length = int(payload.context_length)
existing_row.architecture = json.dumps(payload.architecture)
existing_row.pricing = json.dumps(payload.pricing)
existing_row.sats_pricing = None
existing_row.per_request_limits = (
json.dumps(payload.per_request_limits)
if payload.per_request_limits is not None
else None
),
top_provider=(
)
existing_row.top_provider = (
json.dumps(payload.top_provider) if payload.top_provider else None
),
upstream_provider_id=provider_id,
enabled=payload.enabled,
)
session.add(row)
await session.commit()
await session.refresh(row)
)
existing_row.canonical_slug = payload.canonical_slug
existing_row.alias_ids = (
json.dumps(payload.alias_ids) if payload.alias_ids else None
)
existing_row.enabled = payload.enabled
session.add(existing_row)
await session.commit()
await session.refresh(existing_row)
row = existing_row
else:
# Create new model
logger.info(f"Creating new model: {payload.id}")
row = ModelRow(
id=payload.id,
name=payload.name,
description=payload.description,
created=int(payload.created),
context_length=int(payload.context_length),
architecture=json.dumps(payload.architecture),
pricing=json.dumps(payload.pricing),
sats_pricing=None,
per_request_limits=(
json.dumps(payload.per_request_limits)
if payload.per_request_limits is not None
else None
),
top_provider=(
json.dumps(payload.top_provider) if payload.top_provider else None
),
canonical_slug=payload.canonical_slug,
alias_ids=(
json.dumps(payload.alias_ids) if payload.alias_ids else None
),
upstream_provider_id=provider_id,
enabled=payload.enabled,
)
session.add(row)
await session.commit()
await session.refresh(row)
await refresh_model_maps()
return _row_to_model(
@@ -2461,6 +2485,20 @@ async def create_provider_model(
).dict() # type: ignore
@admin_router.patch(
"/api/upstream-providers/{provider_id}/models/{model_id:path}",
dependencies=[Depends(require_admin_api)],
)
async def update_provider_model_legacy(
provider_id: int, model_id: str, payload: ModelCreate
) -> dict[str, object]:
"""Legacy PATCH endpoint - redirects to upsert POST endpoint for backward compatibility."""
logger.info(
f"LEGACY_PATCH_UPDATE called: provider_id={provider_id}, model_id={model_id}"
)
return await upsert_provider_model(provider_id, payload)
@admin_router.get(
"/api/upstream-providers/{provider_id}/models/{model_id:path}",
dependencies=[Depends(require_admin_api)],
@@ -2481,76 +2519,6 @@ async def get_provider_model(provider_id: int, model_id: str) -> dict[str, objec
).dict() # type: ignore
@admin_router.patch(
"/api/upstream-providers/{provider_id}/models/{model_id:path}",
dependencies=[Depends(require_admin_api)],
)
async def update_provider_model(
provider_id: int, model_id: str, payload: ModelUpdate
) -> dict[str, object]:
if payload.id != model_id:
raise HTTPException(status_code=400, detail="Path id does not match payload id")
async with create_session() as session:
provider = await session.get(UpstreamProviderRow, provider_id)
if not provider:
raise HTTPException(status_code=404, detail="Provider not found")
row = await session.get(ModelRow, (model_id, provider_id))
if not row:
raise HTTPException(
status_code=404, detail="Model not found for this provider"
)
row.name = payload.name
row.description = payload.description
row.created = int(payload.created)
row.context_length = int(payload.context_length)
row.architecture = json.dumps(payload.architecture)
row.pricing = json.dumps(payload.pricing)
row.sats_pricing = None
row.per_request_limits = (
json.dumps(payload.per_request_limits)
if payload.per_request_limits is not None
else None
)
row.top_provider = (
json.dumps(payload.top_provider) if payload.top_provider else None
)
was_disabled = not row.enabled
row.enabled = payload.enabled
session.add(row)
await session.commit()
await session.refresh(row)
if was_disabled and payload.enabled:
from ..payment.models import _cleanup_enabled_models_once
try:
await _cleanup_enabled_models_once()
except Exception as e:
logger.warning(
f"Failed to run model cleanup after enabling: {e}",
extra={"model_id": model_id, "error": str(e)},
)
await refresh_model_maps()
return _row_to_model(
row, apply_provider_fee=True, provider_fee=provider.provider_fee
).dict() # type: ignore
@admin_router.put(
"/api/upstream-providers/{provider_id}/models/{model_id:path}",
dependencies=[Depends(require_admin_api)],
)
async def update_provider_model_put(
provider_id: int, model_id: str, payload: ModelUpdate
) -> dict[str, object]:
return await update_provider_model(provider_id, model_id, payload)
@admin_router.delete(
"/api/upstream-providers/{provider_id}/models/{model_id:path}",
dependencies=[Depends(require_admin_api)],
+4
View File
@@ -68,6 +68,10 @@ class ModelRow(SQLModel, table=True): # type: ignore
sats_pricing: str | None = Field(default=None)
per_request_limits: str | None = Field(default=None)
top_provider: str | None = Field(default=None)
canonical_slug: str | None = Field(default=None, description="Canonical model slug")
alias_ids: str | None = Field(
default=None, description="JSON array of model alias IDs"
)
enabled: bool = Field(default=True, description="Whether this model is enabled")
upstream_provider: "UpstreamProviderRow" = Relationship(back_populates="models")
+5 -12
View File
@@ -14,7 +14,6 @@ from ..balance import balance_router, deprecated_wallet_router
from ..discovery import providers_cache_refresher, providers_router
from ..nip91 import announce_provider
from ..payment.models import (
cleanup_enabled_models_periodically,
models_router,
update_sats_pricing,
)
@@ -49,7 +48,6 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
nip91_task = None
providers_task = None
models_refresh_task = None
models_cleanup_task = None
model_maps_refresh_task = None
try:
@@ -80,13 +78,17 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
_update_prices_task = asyncio.create_task(_update_prices())
_initialize_upstreams_task = asyncio.create_task(initialize_upstreams())
# ensure both setup tasks complete
await asyncio.gather(
_update_prices_task, _initialize_upstreams_task, return_exceptions=True
)
btc_price_task = asyncio.create_task(update_prices_periodically())
pricing_task = asyncio.create_task(update_sats_pricing())
if global_settings.models_refresh_interval_seconds > 0:
models_refresh_task = asyncio.create_task(
refresh_upstreams_models_periodically(get_upstreams())
)
models_cleanup_task = asyncio.create_task(cleanup_enabled_models_periodically())
model_maps_refresh_task = asyncio.create_task(refresh_model_maps_periodically())
payout_task = asyncio.create_task(periodic_payout())
if global_settings.nsec:
@@ -94,11 +96,6 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
if global_settings.providers_refresh_interval_seconds > 0:
providers_task = asyncio.create_task(providers_cache_refresher())
# ensure both setup tasks complete
await asyncio.gather(
_update_prices_task, _initialize_upstreams_task, return_exceptions=True
)
yield
except asyncio.CancelledError:
@@ -125,8 +122,6 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
providers_task.cancel()
if models_refresh_task is not None:
models_refresh_task.cancel()
if models_cleanup_task is not None:
models_cleanup_task.cancel()
if model_maps_refresh_task is not None:
model_maps_refresh_task.cancel()
@@ -144,8 +139,6 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
tasks_to_wait.append(providers_task)
if models_refresh_task is not None:
tasks_to_wait.append(models_refresh_task)
if models_cleanup_task is not None:
tasks_to_wait.append(models_cleanup_task)
if model_maps_refresh_task is not None:
tasks_to_wait.append(model_maps_refresh_task)
+10 -1
View File
@@ -55,7 +55,16 @@ class LoggingMiddleware(BaseHTTPMiddleware):
"headers": {
k: v
for k, v in request.headers.items()
if k.lower() not in ["authorization", "x-cashu", "cookie"]
if k.lower()
not in [
"authorization",
"x-cashu",
"cookie",
"cf-connecting-ip",
"cf-ipcountry",
"x-forwarded-for",
"x-real-ip",
]
},
"body_size": len(request_body) if request_body else 0,
},
+11 -112
View File
@@ -5,10 +5,9 @@ import random
import httpx
from fastapi import APIRouter, Depends
from pydantic.v1 import BaseModel
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
from ..core.db import ModelRow, create_session, get_session
from ..core.db import ModelRow, get_session
from ..core.logging import get_logger
from ..core.settings import settings
from .price import sats_usd_price
@@ -185,6 +184,7 @@ def _row_to_model(
enabled=row.enabled,
upstream_provider_id=row.upstream_provider_id,
canonical_slug=getattr(row, "canonical_slug", None),
alias_ids=json.loads(row.alias_ids) if row.alias_ids else None,
)
if apply_provider_fee:
@@ -382,7 +382,7 @@ def _update_model_sats_pricing(model: Model, sats_to_usd: float) -> Model:
async def _update_sats_pricing_once() -> None:
"""Update sats pricing once for all provider models (in-memory only)."""
from ..proxy import get_upstreams
from ..proxy import get_upstreams, refresh_model_maps
upstreams = get_upstreams()
sats_to_usd = sats_usd_price()
@@ -399,6 +399,7 @@ async def _update_sats_pricing_once() -> None:
if updated_count > 0:
logger.info("Updated sats pricing", extra={"models_updated": updated_count})
await refresh_model_maps()
async def update_sats_pricing() -> None:
@@ -409,7 +410,13 @@ async def update_sats_pricing() -> None:
except Exception:
pass
await _update_sats_pricing_once()
try:
await _update_sats_pricing_once()
except Exception as e:
logger.warning(
"Initial sats pricing update failed (will retry in loop)",
extra={"error": str(e)},
)
while True:
try:
@@ -433,114 +440,6 @@ async def update_sats_pricing() -> None:
logger.error(f"Error updating sats pricing: {e}")
async def cleanup_enabled_models_periodically() -> None:
"""Background task to clean up enabled models that match upstream pricing.
When model is enabled (enabled=True), remove it from DB if it matches upstream pricing.
Keep it in DB only if pricing differs from upstream or if it's disabled.
"""
interval = getattr(
settings, "models_cleanup_interval_seconds", 300
) # 5 minutes default
if not interval or interval <= 0:
return
while True:
try:
await _cleanup_enabled_models_once()
except asyncio.CancelledError:
break
except Exception as e:
logger.error(
"Error during enabled models cleanup",
extra={"error": str(e), "error_type": type(e).__name__},
)
try:
jitter = max(0.0, float(interval) * 0.1)
await asyncio.sleep(interval + random.uniform(0, jitter))
except asyncio.CancelledError:
break
async def _cleanup_enabled_models_once() -> None:
"""Clean up enabled models that match upstream pricing."""
from ..proxy import get_upstreams
async with create_session() as session:
# Get all enabled models from DB
result = await session.exec(
select(ModelRow).where(
ModelRow.enabled, # Only enabled models
)
)
db_models = result.all()
if not db_models:
return
upstreams = get_upstreams()
models_to_remove = []
for db_model in db_models:
# Find corresponding upstream model
upstream_model = None
for upstream in upstreams:
upstream_model = upstream.get_cached_model_by_id(db_model.id)
if upstream_model:
break
if not upstream_model:
continue
# Compare pricing to see if they match
db_pricing = json.loads(db_model.pricing)
upstream_pricing = upstream_model.pricing.dict()
# Check if pricing matches (with small tolerance for float comparison)
pricing_matches = _pricing_matches(db_pricing, upstream_pricing)
if pricing_matches:
models_to_remove.append(db_model)
logger.info(
f"Removing enabled model {db_model.id} - matches upstream pricing",
extra={"model_id": db_model.id},
)
# Remove models that match upstream pricing
for model in models_to_remove:
await session.delete(model)
if models_to_remove:
await session.commit()
logger.info(
f"Cleaned up {len(models_to_remove)} enabled models that match upstream pricing"
)
def _pricing_matches(
db_pricing: dict, upstream_pricing: dict, tolerance: float = 0.0
) -> bool:
"""Check if pricing dictionaries match within tolerance."""
keys_to_compare = [
"prompt",
"completion",
"request",
"image",
"web_search",
"internal_reasoning",
]
for key in keys_to_compare:
db_val = int(float(db_pricing.get(key, 0.0)) * 1000000)
upstream_val = int(float(upstream_pricing.get(key, 0.0)) * 1000000)
if abs(db_val - upstream_val) > tolerance:
return False
return True
@models_router.get("/v1/models")
@models_router.get("/models", include_in_schema=False)
async def models(session: AsyncSession = Depends(get_session)) -> dict:
+18 -22
View File
@@ -70,7 +70,7 @@ def get_model_instance(model_id: str) -> Model | None:
def get_provider_for_model(model_id: str) -> BaseUpstreamProvider | None:
"""Get UpstreamProvider for model ID from global cache."""
return _provider_map.get(model_id)
return _provider_map.get(model_id.lower())
def get_unique_models() -> list[Model]:
@@ -80,31 +80,27 @@ def get_unique_models() -> list[Model]:
async def refresh_model_maps() -> None:
"""Refresh global model and provider maps using the cost-based algorithm."""
from sqlalchemy.orm import selectinload
global _model_instances, _provider_map, _unique_models
# Gather database overrides and disabled models
async with create_session() as session:
result = await session.exec(select(ModelRow).where(ModelRow.enabled))
override_rows = result.all()
provider_result = await session.exec(select(UpstreamProviderRow))
providers_by_id = {p.id: p for p in provider_result.all()}
overrides_by_id: dict[str, tuple[ModelRow, float]] = {
row.id: (
row,
providers_by_id[row.upstream_provider_id].provider_fee
if row.upstream_provider_id in providers_by_id
else 1.01,
)
for row in override_rows
if row.upstream_provider_id is not None
}
disabled_result = await session.exec(
select(ModelRow.id).where(ModelRow.enabled == False) # noqa: E712
# Fetch all providers with their models in a single logical operation
query = select(UpstreamProviderRow).options(
selectinload(UpstreamProviderRow.models) # type: ignore
)
disabled_model_ids = {row for row in disabled_result.all()}
result = await session.exec(query)
provider_rows = result.all()
overrides_by_id: dict[str, tuple[ModelRow, float]] = {}
disabled_model_ids: set[str] = set()
for provider in provider_rows:
for model in provider.models:
if model.enabled:
overrides_by_id[model.id] = (model, provider.provider_fee)
else:
disabled_model_ids.add(model.id)
_model_instances, _provider_map, _unique_models = create_model_mappings(
upstreams=_upstreams,
+23 -6
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import asyncio
import os
import re
from typing import TYPE_CHECKING
@@ -7,6 +8,7 @@ from typing import TYPE_CHECKING
if TYPE_CHECKING:
from ..core.settings import Settings
from sqlmodel import select
from ..core import get_logger
from ..core.db import AsyncSession, ModelRow, UpstreamProviderRow, create_session
@@ -145,6 +147,17 @@ async def refresh_upstreams_models_periodically(
f"Error refreshing models for {upstream.base_url}",
extra={"error": str(e), "error_type": type(e).__name__},
)
try:
from ..payment.models import _update_sats_pricing_once
await _update_sats_pricing_once()
except Exception as e:
logger.warning(f"Failed to update pricing after model refresh: {e}")
from ..proxy import refresh_model_maps
await refresh_model_maps()
except asyncio.CancelledError:
break
except Exception as e:
@@ -166,8 +179,6 @@ async def init_upstreams() -> list[BaseUpstreamProvider]:
Seeds database with providers from settings if empty, then loads and instantiates
provider instances from database records, and refreshes their models cache.
"""
from sqlmodel import select
from ..core.settings import settings
async with create_session() as session:
@@ -183,16 +194,16 @@ async def init_upstreams() -> list[BaseUpstreamProvider]:
result = await session.exec(select(UpstreamProviderRow))
existing_providers = result.all()
upstreams: list[BaseUpstreamProvider] = []
for provider_row in existing_providers:
async def _init_single_provider(
provider_row: UpstreamProviderRow,
) -> BaseUpstreamProvider | None:
if not provider_row.enabled:
logger.debug(f"Skipping disabled provider: {provider_row.base_url}")
continue
return None
provider = _instantiate_provider(provider_row)
if provider:
await provider.refresh_models_cache()
upstreams.append(provider)
logger.debug(
f"Initialized {provider_row.provider_type} provider",
extra={
@@ -200,6 +211,12 @@ async def init_upstreams() -> list[BaseUpstreamProvider]:
"models_cached": len(provider.get_cached_models()),
},
)
return provider
return None
tasks = [_init_single_provider(row) for row in existing_providers]
results = await asyncio.gather(*tasks)
upstreams = [p for p in results if p is not None]
return upstreams
+4
View File
@@ -41,6 +41,10 @@ export function ModelSearchFilter({
model.name.toLowerCase().includes(query) ||
model.full_name.toLowerCase().includes(query) ||
model.provider.toLowerCase().includes(query) ||
(model.alias_ids &&
model.alias_ids.some((alias) =>
alias.toLowerCase().includes(query)
)) ||
(model.description &&
model.description.toLowerCase().includes(query)) ||
model.modelType.toLowerCase().includes(query)
+2
View File
@@ -542,6 +542,7 @@ export function ModelSelector({
top_provider: null,
upstream_provider_id: providerId,
enabled: model.isEnabled,
alias_ids: model.alias_ids || null,
};
setModelDialogState({
@@ -585,6 +586,7 @@ export function ModelSelector({
top_provider: null,
upstream_provider_id: providerId,
enabled: model.isEnabled,
alias_ids: model.alias_ids || null,
};
setModelDialogState({
+1
View File
@@ -31,6 +31,7 @@ export const ModelSchema = z.object({
// API key type indicators
has_own_api_key: z.boolean(),
api_key_type: z.string(), // "individual" or "group"
alias_ids: z.array(z.string()).nullable().optional(),
});
// Schema for a model with additional provider-specific settings
+5 -2
View File
@@ -121,6 +121,7 @@ export interface AdminModelAsModel {
soft_deleted?: boolean;
has_own_api_key: boolean;
api_key_type: string;
alias_ids?: string[] | null;
}
export interface AdminModelGroup {
@@ -207,6 +208,7 @@ export class AdminService {
soft_deleted: !adminModel.enabled,
has_own_api_key: false,
api_key_type: 'group',
alias_ids: adminModel.alias_ids,
};
}
@@ -354,8 +356,9 @@ export class AdminService {
original: data.pricing,
converted: payload.pricing,
});
const model = await apiClient.patch<AdminModel>(
`/admin/api/upstream-providers/${providerId}/models/${encodeURIComponent(modelId)}`,
// Use the same POST endpoint for both create and update (upsert)
const model = await apiClient.post<AdminModel>(
`/admin/api/upstream-providers/${providerId}/models`,
payload
);
return {