File size: 5,640 Bytes
9f9844d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 |
import requests
import json
from typing import List, Dict, Optional
class ResponseWrapper:
def __init__(self, response_data):
"""
Wrap the response data to support both dict-like and attribute-like access
:param response_data: The raw response dictionary from OpenRouter
"""
self._data = response_data
def __getattr__(self, name):
"""
Allow attribute-style access to the response data
:param name: Attribute name to access
:return: Corresponding value from the response data
"""
if name in self._data:
value = self._data[name]
return self._wrap(value)
raise AttributeError(f"'{type(self).__name__}' object has no attribute '{name}'")
def __getitem__(self, key):
"""
Allow dictionary-style access to the response data
:param key: Key to access
:return: Corresponding value from the response data
"""
value = self._data[key]
return self._wrap(value)
def _wrap(self, value):
"""
Recursively wrap dictionaries and lists to support attribute access
:param value: Value to wrap
:return: Wrapped value
"""
if isinstance(value, dict):
return ResponseWrapper(value)
elif isinstance(value, list):
return [self._wrap(item) for item in value]
return value
def __iter__(self):
"""
Allow iteration over the wrapped dictionary
"""
return iter(self._data)
def get(self, key, default=None):
"""
Provide a get method similar to dictionary
"""
return self._wrap(self._data.get(key, default))
def keys(self):
"""
Return dictionary keys
"""
return self._data.keys()
def items(self):
"""
Return dictionary items
"""
return [(k, self._wrap(v)) for k, v in self._data.items()]
def __str__(self):
"""
Return a JSON string representation of the response data
:return: JSON-formatted string of the response
"""
return json.dumps(self._data, indent=2)
def __repr__(self):
"""
Return a string representation for debugging
:return: Representation of the ResponseWrapper
"""
return f"ResponseWrapper({json.dumps(self._data, indent=2)})"
class OpenRouter:
def __init__(self, api_key: str, base_url: str = "https://openrouter.ai/api/v1"):
"""
Initialize OpenRouter client
:param api_key: API key for OpenRouter
:param base_url: Base URL for OpenRouter API (default is standard endpoint)
"""
self.api_key = api_key
self.base_url = base_url
self.chat = self.ChatNamespace(self)
class ChatNamespace:
def __init__(self, client):
self._client = client
self.completions = self.CompletionsNamespace(client)
class CompletionsNamespace:
def __init__(self, client):
self._client = client
def create(
self,
model: str,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: Optional[int] = None,
**kwargs
):
"""
Create a chat completion request
:param model: Model to use
:param messages: List of message dictionaries
:param temperature: Sampling temperature
:param max_tokens: Maximum number of tokens to generate
:return: Wrapped response object
"""
headers = {
"Authorization": f"Bearer {self._client.api_key}",
"Content-Type": "application/json",
"HTTP-Referer": kwargs.get("http_referer", "https://your-app-domain.com"),
"X-Title": kwargs.get("x_title", "AI Ad Generator")
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
}
if model.startswith("deepseek"):
payload["provider"] = {
"order": [
"DeepSeek",
"DeepInfra",
"Fireworks",
],
"allow_fallbacks": False
}
if max_tokens is not None:
payload["max_tokens"] = max_tokens
# Add any additional parameters
payload.update({k: v for k, v in kwargs.items()
if k not in ["http_referer", "x_title"]})
try:
response = requests.post(
f"{self._client.base_url}/chat/completions",
headers=headers,
data=json.dumps(payload)
)
response.raise_for_status()
# Wrap the response data
return ResponseWrapper(response.json())
except requests.RequestException as e:
raise Exception(f"OpenRouter API request failed: {e}") |