support openrouter as an llm provider - #190
Conversation
| # Transient stream failures and how many times to re-issue the request. The SDK's | ||
| # own `max_retries` covers *establishing* a request; these surface while the SSE | ||
| # body is being consumed, by which point the request has already succeeded, so | ||
| # nothing below us retries them and they take the whole run down. `TimeoutError` |
There was a problem hiding this comment.
not true anymore, fwiw, we have graph level retry mechanisms....
| """OpenRouter's model roster, by id: one blocking unauthenticated GET, no retry, | ||
| from :meth:`OpenRouterModelProvider.create` at startup. Any failure yields an | ||
| empty catalog and a run on conservative defaults.""" | ||
| try: |
| _MODELS_URL, exc, _FALLBACK_CONTEXT_WINDOW, | ||
| ) | ||
| return {} | ||
| records = payload.get("data") if isinstance(payload, dict) else None |
There was a problem hiding this comment.
assuming this json is returned according to a documented schema, just use pydantic for heaven's sake rather than this "hand" rolled parsing...
| def _positive_int(value: Any) -> int | None: | ||
| return value if isinstance(value, int) and not isinstance(value, bool) and value > 0 else None |
There was a problem hiding this comment.
what the blue hell? again, all of this can be declaratively described using pydantic I think ...
| # --- pricing --------------------------------------------------------------- | ||
|
|
||
| # OpenRouter quotes USD per token; PriceTier is USD per million. | ||
| _TOKENS_PER_MTOK = 1_000_000 |
There was a problem hiding this comment.
"tokens per million tokens == 1 million"
word?
| class OpenRouterRenderer(OpenAIRenderer): | ||
| """Content blocks in the Chat-Completions shape, which is what langchain converts | ||
| from: ``_convert_chat_completions_blocks_to_responses`` turns ``file`` into | ||
| ``input_file`` and ``image_url`` into ``input_image`` on the way out. Only the | ||
| file block differs from OpenAI's; the text block is inherited.""" | ||
|
|
||
| @override | ||
| def file_block( | ||
| self, file_id: str, *, filename: str, cache_level: CacheLevel = CacheLevel.NONE | ||
| ) -> dict: | ||
| # `file_id` is a `data:` URL rather than a remote id: OpenRouter has no | ||
| # Files API, so InlineFileUploader carries the bytes here instead. | ||
| if file_id.startswith("data:image/"): | ||
| return {"type": "image_url", "image_url": {"url": file_id}} | ||
| return {"type": "file", "file": {"filename": filename, "file_data": file_id}} | ||
|
|
||
|
|
||
| @dataclass | ||
| class InlineFileUploader(UploaderBase): | ||
| """``FileUploader`` impl for a provider with no Files API: the "upload" is a | ||
| ``data:`` URL built in memory, which the renderer inlines into the request. So a | ||
| large binary is re-sent with every request carrying it, and a PDF no route reads | ||
| natively goes through OpenRouter's ``file-parser`` plugin, which falls back to | ||
| ``mistral-ocr`` at $2/1K pages (pin an engine via ``plugins`` to avoid that).""" | ||
|
|
||
| renderer: ContentRenderer = field(default_factory=OpenRouterRenderer) | ||
|
|
||
| async def _upload_bytes( | ||
| self, crc_basename: str, file_data: bytes, mime: str | ||
| ) -> str: | ||
| # No dedup cache: the "id" *is* the content, so there is nothing to reuse. | ||
| # Off-thread because the encode is ~1.5ms/MB of blocked loop, matching how | ||
| # `composer.input.files` already offloads the read. | ||
| encoded = await asyncio.to_thread(base64.b64encode, file_data) | ||
| return f"data:{mime};base64,{encoded.decode('ascii')}" |
There was a problem hiding this comment.
oh. I hate it. yes it works, but its hijacking the fact that the type we're using to represent a file id (str) happens to also allow us to smuggle through the Base64 encoding that is necessary for an "inline" block. Far better to make the file upload optional rather than lying about it.
Luckily, I think the fix is pretty easy; make an analog of InMemoryTextFile called InMemoryBytesFile which invokes a new API on FileRenderer: inline_file_block or something. then the "regular" file_block becomes a NotImplemented error thrown by the openrouter content renderer.
| def __init__(self): | ||
| from graphcore.tools.memory import openai_async_memory_tool | ||
| super().__init__( | ||
| # The OpenAI-flavored memory tool is a plain client-side function tool |
There was a problem hiding this comment.
FFS. Its so weird seeing claude get this wrong. The anthropic memory tool is also client side. It is convinced that the anthropic memory tool we use is server side, it isn't!!!!! Stop implying it is!
| class RetryingChatOpenAI(ChatOpenAI): | ||
| @override | ||
| async def _astream( | ||
| self, *args: Any, **kwargs: Any | ||
| ) -> AsyncIterator["ChatGenerationChunk"]: | ||
| for attempt in range(1, _STREAM_ATTEMPTS + 1): | ||
| # Buffered, not forwarded as they arrive: a retry must not emit a | ||
| # partial response twice. Callers aggregate through `ainvoke` | ||
| # anyway, so this costs nothing today — an incremental consumer | ||
| # would lose its incrementality. | ||
| chunks: list["ChatGenerationChunk"] = [] | ||
| try: | ||
| async for chunk in super()._astream(*args, **kwargs): | ||
| chunks.append(chunk) | ||
| except _RETRYABLE_STREAM_ERRORS as exc: | ||
| if attempt == _STREAM_ATTEMPTS: | ||
| raise | ||
| delay = _STREAM_RETRY_BACKOFF_SECONDS * attempt | ||
| logger.warning( | ||
| "OpenRouter stream failed after %d chunk(s) (%s: %s); " | ||
| "re-issuing in %.0fs (attempt %d/%d).", | ||
| len(chunks), type(exc).__name__, exc, delay, | ||
| attempt + 1, _STREAM_ATTEMPTS, | ||
| ) | ||
| await asyncio.sleep(delay) | ||
| continue | ||
| for chunk in chunks: | ||
| yield chunk | ||
| return | ||
|
|
There was a problem hiding this comment.
Is this mooted by the retry mechanisms I recently landed? Or is it a middle level of retrying (in which case we are probably retrying at three different levels...)
| provider: OpenRouterService = field(default_factory=_openrouter_service) | ||
|
|
||
| @staticmethod | ||
| def create(model_name: str, options: ModelConfiguration) -> "OpenRouterModelProvider": |
There was a problem hiding this comment.
oh, because this isn't async. hum.
| model=self.model_name, | ||
| base_url=BASE_URL, | ||
| api_key=SecretStr(self.api_key), | ||
| # Load-bearing, not a default: the Responses API is the only surface on |
| raise NotImplementedError( | ||
| "OpenAI content is uploaded to the Files API, not inlined." | ||
| ) |
There was a problem hiding this comment.
technically we could allow inline blocks too, they are supported just fine on anthropic/openai. But I don't think we do; its usually a disaster for our checkpointing...
| (``utc_start``/``utc_end``), which a per-call price curve cannot express, so | ||
| those arrive with ``min_prompt_tokens`` unset and are skipped.""" | ||
|
|
||
| min_prompt_tokens: PositiveInt | None = None |
There was a problem hiding this comment.
oh, and positiveint is even built in to pydantic. nice!
| class _ModelRecord(BaseModel): | ||
| id: str | ||
| context_length: PositiveInt | None = None | ||
| top_provider: _TopProvider = Field(default_factory=_TopProvider) |
There was a problem hiding this comment.
It doesn't matter, but I'm trying to figure out what "Top" means here. Like, the top provider of a specific model?
There was a problem hiding this comment.
openrouteer uses several providers for each model (e.g. kimi-k3 has 17 different possible providers) and it uses some sort of weighted load balancing to choose between them for each call. the top provider is the one openrouter thinks is best. but looking at this again, this actually exposed an issue - some providers may have a smaller context lengths, which would have caused issues if it would try to route to them. So I'm changing this to instead take only providers that provide a long enough context.
2 participants
It matches any model with the
author/modelnaming scheme.