Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-llm/src/lexigram/ai/llm/clients/openrouter.py: 22%
171 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""OpenRouter API client.
3OpenRouter provides an OpenAI-compatible API surface. This client implements a
4lightweight wrapper compatible with the existing OpenAI/OpenAICompatible clients
5so it can be used interchangeably in higher-level code.
6"""
8from __future__ import annotations
10import asyncio
11import types
12from typing import TYPE_CHECKING, Any, cast
14import aiohttp
16from lexigram.ai.llm.clients._tools_utils import (
17 parse_openai_tool_calls,
18 serialize_message_for_openai,
19 tool_to_openai_format,
20)
21from lexigram.ai.llm.exceptions import (
22 LLMAuthenticationError,
23 LLMError,
24 LLMModelNotFoundError,
25 LLMQuotaExceededError,
26 LLMRateLimitError,
27 LLMTimeoutError,
28)
29from lexigram.ai.llm.http.client import ResilientHTTPClient
30from lexigram.ai.llm.types import (
31 AIError,
32 ChatMessage,
33 Completion,
34 Role,
35 StreamChunk,
36 ThinkingResult,
37 TokenUsage,
38)
39from lexigram.contracts.core.health import HealthCheckResult, HealthStatus
40from lexigram.contracts.web.http_models import HttpStatusError
41from lexigram.logging import (
42 get_logger,
43)
44from lexigram.result import Err, Ok, Result
45from lexigram.serialization import loads
46from lexigram.validation import SecretStr
48logger = get_logger(__name__)
51from lexigram.ai.llm.clients.base import AbstractLLMClient
53if TYPE_CHECKING:
54 from collections.abc import AsyncIterator
56 from lexigram.ai.llm.config import ClientConfig
59class OpenRouterClient(AbstractLLMClient):
60 """Client for OpenRouter (OpenAI-compatible) API.
62 Conforms to: :class:`~lexigram.contracts.ai.LLMClientProtocol` protocol via structural typing.
63 """
65 def __init__(self, config: ClientConfig):
66 """Initialize OpenRouter client.
68 Args:
69 config: LLM configuration
70 """
71 super().__init__(config=config)
72 self._client: ResilientHTTPClient | None = None
74 @property
75 def api_key(self) -> SecretStr:
76 """Get API key from config."""
77 return self.config.api_key or SecretStr("")
79 @property
80 def base_url(self) -> str:
81 """Get base URL from config."""
82 return self.config.api_base or "https://api.openrouter.ai/v1"
84 @property
85 def model(self) -> str:
86 """Get default model from config."""
87 return self.config.model
89 async def _get_client(self) -> ResilientHTTPClient:
90 """Get or create a resilient HTTP client for OpenRouter."""
91 if self._client is None:
92 headers = {
93 "Authorization": f"Bearer {self.api_key.get_secret_value()}",
94 "Content-Type": "application/json",
95 }
96 self._client = ResilientHTTPClient(
97 base_url=self.base_url,
98 headers=headers,
99 timeout=self.config.timeout,
100 name="openrouter-client",
101 )
102 return self._client
104 async def _do_complete(
105 self,
106 messages: list[Any],
107 **kwargs: Any,
108 ) -> Result[Completion, LLMError]:
109 """Generate chat completion.
111 Accepts same message shape as other LLM clients used in the project.
113 Returns:
114 ``Ok(Completion)`` for non-streaming success.
115 ``Ok(AsyncIterator[StreamChunk])`` for streaming success.
116 ``Err(LLMError)`` for recoverable failures.
117 """
118 client = await self._get_client()
119 temperature = kwargs.pop("temperature", self.config.temperature)
120 max_tokens = kwargs.pop("max_tokens", self.config.max_tokens)
121 tools = kwargs.pop("tools", None)
123 message_dicts: list[dict[str, Any]] = []
124 for msg in messages:
125 if isinstance(msg, ChatMessage):
126 message_dicts.append(serialize_message_for_openai(msg))
127 else:
128 message_dicts.append(msg)
130 payload: dict[str, Any] = {
131 "model": kwargs.pop("model", self.model),
132 "messages": message_dicts,
133 "temperature": temperature,
134 "stream": False,
135 **kwargs,
136 }
138 if max_tokens:
139 payload["max_tokens"] = max_tokens
140 if tools:
141 payload["tools"] = [
142 converted
143 for tool in tools
144 if (converted := tool_to_openai_format(tool)) is not None
145 ]
146 self._apply_thinking(payload)
148 try:
149 return Ok(await self._complete(client, payload))
150 except (
151 OSError,
152 ConnectionError,
153 TimeoutError,
154 RuntimeError,
155 ValueError,
156 HttpStatusError,
157 ) as e:
158 return self._handle_error_as_result(e)
160 async def _do_stream_chat(
161 self,
162 messages: list[Any],
163 **kwargs: Any,
164 ) -> Result[AsyncIterator[StreamChunk], LLMError]:
165 """Start a streaming completion (protocol-aligned method).
167 Args:
168 messages: List of chat messages (ChatMessage or raw dicts).
169 **kwargs: Additional OpenRouter API parameters.
171 Returns:
172 ``Ok(AsyncIterator[StreamChunk])`` on success.
173 ``Err(LLMError)`` for recoverable failures.
174 """
175 try:
176 client = await self._get_client()
177 temperature = kwargs.pop("temperature", self.config.temperature)
178 max_tokens = kwargs.pop("max_tokens", self.config.max_tokens)
180 message_dicts: list[dict[str, Any]] = []
181 for msg in messages:
182 if isinstance(msg, ChatMessage):
183 message_dicts.append(
184 {"role": msg.role.value, "content": msg.content}
185 )
186 else:
187 message_dicts.append(msg)
189 payload: dict[str, Any] = {
190 "model": kwargs.pop("model", self.model),
191 "messages": message_dicts,
192 "temperature": temperature,
193 "stream": True,
194 **kwargs,
195 }
197 if max_tokens:
198 payload["max_tokens"] = max_tokens
199 self._apply_thinking(payload)
201 return Ok(self._stream_completion(client, payload))
202 except (OSError, ConnectionError, TimeoutError, RuntimeError, ValueError) as e:
203 return self._handle_error_as_result(e)
205 async def _do_chat(
206 self,
207 messages: list[Any],
208 tools: list[Any] | None = None,
209 **kwargs: Any,
210 ) -> Result[Completion, LLMError]:
211 return await self._do_complete(messages, tools=tools, **kwargs)
213 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult:
214 """Perform a lightweight health check against the OpenRouter API.
216 Calls the models listing endpoint to verify the API key is valid and
217 the service is reachable.
219 Args:
220 timeout: Maximum seconds to wait for the response.
222 Returns:
223 :class:`~lexigram.contracts.core.health.HealthCheckResult`.
224 """
225 try:
226 client = await self._get_client()
227 response = await client.get("/models")
228 response.raise_for_status()
229 return HealthCheckResult(
230 component="llm.openrouter",
231 status=HealthStatus.HEALTHY,
232 details={"provider": "openrouter", "model": self.config.model},
233 )
234 except (
235 OSError,
236 ConnectionError,
237 TimeoutError,
238 RuntimeError,
239 HttpStatusError,
240 ) as exc:
241 return HealthCheckResult(
242 component="llm.openrouter",
243 status=HealthStatus.UNHEALTHY,
244 error=str(exc),
245 )
247 async def _complete(
248 self,
249 client: ResilientHTTPClient,
250 payload: dict[str, Any],
251 ) -> Completion:
252 try:
253 response = await client.post("/chat/completions", json=payload)
254 response.raise_for_status()
255 data: Any = response.json
257 choice = data["choices"][0]
258 message = choice["message"]
260 tool_calls = parse_openai_tool_calls(message.get("tool_calls"))
262 token_usage = None
263 if "usage" in data:
264 token_usage = TokenUsage(
265 prompt_tokens=data["usage"].get("prompt_tokens", 0),
266 completion_tokens=data["usage"].get("completion_tokens", 0),
267 total_tokens=data["usage"].get("total_tokens", 0),
268 )
270 # OpenRouter: reasoning field present when include_reasoning=True
271 reasoning: str | None = message.get("reasoning") or None
272 thinking: ThinkingResult | None = (
273 ThinkingResult(content=reasoning) if reasoning else None
274 )
276 return Completion(
277 content=message.get("content", ""),
278 role=Role(message.get("role", "assistant")),
279 tool_calls=tool_calls,
280 finish_reason=choice.get("finish_reason"),
281 usage=token_usage,
282 thinking=thinking,
283 model=data.get("model", self.model),
284 )
285 except (HttpStatusError, TimeoutError):
286 # Propagate unchanged so _do_complete maps these to typed
287 # LLMErrors via _handle_error_as_result (429 → rate limit,
288 # timeout → LLMTimeoutError, etc.).
289 raise
290 except (
291 aiohttp.ClientError,
292 ValueError,
293 TypeError,
294 ) as e:
295 raise AIError(f"OpenRouter error: {e}") from e
297 async def _stream_completion(
298 self,
299 client: ResilientHTTPClient,
300 payload: dict[str, Any],
301 ) -> AsyncIterator[StreamChunk]:
302 try:
303 stream_ctx = client.stream("POST", "/chat/completions", json=payload)
304 if asyncio.iscoroutine(stream_ctx):
305 stream_ctx = await stream_ctx
307 async with stream_ctx as response:
308 # Prefer content iteration when available, otherwise fall back
309 # to aiohttp's line iterator for compatibility with various
310 # underlying client implementations.
311 if hasattr(response, "content"):
312 iterator = cast("AsyncIterator[bytes | str]", response.content)
313 else:
314 iterator = cast(
315 "AsyncIterator[bytes | str]", response.aiter_lines()
316 )
318 async for raw_line in iterator:
319 # aiter_lines() yields str; content iterators may yield bytes.
320 if isinstance(raw_line, bytes):
321 line = raw_line.decode("utf-8").strip()
322 else:
323 line = str(raw_line).strip()
324 if not line or not line.startswith("data: "):
325 continue
326 data_str = line[6:]
327 if data_str.strip() == "[DONE]":
328 break
329 try:
330 data = loads(data_str)
331 choice = data["choices"][0]
332 delta = choice.get("delta", {})
333 content = delta.get("content", "")
334 role = delta.get("role")
335 # OpenRouter: reasoning delta when include_reasoning=True
336 reasoning = delta.get("reasoning") or ""
338 if reasoning:
339 yield StreamChunk(
340 thinking_delta=reasoning,
341 is_thinking=True,
342 role=Role(role) if role else None,
343 finish_reason=choice.get("finish_reason"),
344 model=data.get("model", self.model),
345 )
346 elif content:
347 yield StreamChunk(
348 delta=content,
349 role=Role(role) if role else None,
350 finish_reason=choice.get("finish_reason"),
351 model=data.get("model", self.model),
352 )
353 except (ValueError, TypeError) as e:
354 logger.debug("Failed to parse SSE data: %s", e)
355 continue
356 except (TimeoutError, aiohttp.ClientError, ValueError, TypeError, OSError) as e:
357 raise AIError(f"OpenRouter streaming error: {e}") from e
359 async def embeddings(self, texts: list[str], **kwargs: Any) -> list[list[float]]:
360 client = await self._get_client()
361 try:
362 response = await client.post(
363 "/embeddings",
364 json={
365 "model": kwargs.pop("model", self.model),
366 "input": texts,
367 **kwargs,
368 },
369 )
370 response.raise_for_status()
371 data: Any = response.json
372 embeddings = sorted(data.get("data", []), key=lambda x: x.get("index", 0))
373 return [item["embedding"] for item in embeddings]
374 except (aiohttp.ClientError, ValueError, TypeError, KeyError) as e:
375 raise AIError(f"OpenRouter error: {e}") from e
377 async def close(self) -> None:
378 if self._client:
379 await self._client.close()
380 self._client = None
381 await super().close()
383 def _apply_thinking(self, payload: dict[str, Any]) -> None:
384 """Inject OpenRouter reasoning parameters into the API payload.
386 Sets ``include_reasoning = True`` when ``config.thinking`` is configured,
387 which instructs OpenRouter to return the model's reasoning text in the
388 ``reasoning`` field of the response message.
390 Args:
391 payload: Mutable API payload dict.
392 """
393 if self.config.thinking is not None:
394 payload["include_reasoning"] = True
396 def _handle_error_as_result(self, error: Exception) -> Result[Any, LLMError]:
397 """Map a caught exception to ``Err`` (recoverable) or re-raise (infra).
399 Handles both raw HTTP errors and already-converted AIError subclasses
400 (since ``_complete()`` converts errors via ``_handle_error()`` before raising).
401 """
402 if isinstance(error, TimeoutError):
403 return Err(
404 LLMTimeoutError(
405 f"OpenRouter request timed out after {self.config.timeout}s"
406 )
407 )
408 # Handle raw HTTP errors (e.g. from streaming setup)
409 status = None
410 if isinstance(error, (aiohttp.ClientResponseError, HttpStatusError)):
411 status = error.status
412 if status == 401:
413 raise LLMAuthenticationError(
414 f"OpenRouter authentication failed: {error}"
415 ) from error
416 if status == 429:
417 return Err(LLMRateLimitError(f"OpenRouter rate limit exceeded: {error}"))
418 if status == 402:
419 return Err(LLMQuotaExceededError(f"OpenRouter quota exceeded: {error}"))
420 if status == 404:
421 return Err(LLMModelNotFoundError(f"OpenRouter model not found: {error}"))
422 raise AIError(f"OpenRouter infrastructure error: {error}") from error
424 async def __aenter__(self) -> Any:
425 return self
427 async def __aexit__(
428 self,
429 exc_type: type[BaseException] | None,
430 exc_val: BaseException | None,
431 exc_tb: types.TracebackType | None,
432 ) -> Any:
433 await self.close()