| import random |
| import httpx |
| import asyncio |
| import json |
|
|
| class OFFDeepInfraAPI: |
| headers = { |
| 'Accept-Language': 'en-US,en;q=0.9,ja;q=0.8', |
| 'Connection': 'keep-alive', |
| 'Content-Type': 'application/json', |
| 'Origin': 'https://deepinfra.com', |
| 'Referer': 'https://deepinfra.com/', |
| 'Sec-Fetch-Dest': 'empty', |
| 'Sec-Fetch-Mode': 'cors', |
| 'Sec-Fetch-Site': 'same-site', |
| 'User-Agent': 'Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.0.0 Mobile Safari/537.36', |
| 'X-Deepinfra-Source': 'web-embed', |
| 'accept': 'text/event-stream', |
| 'sec-ch-ua': '"Chromium";v="136", "Google Chrome";v="136", "Not.A/Brand";v="99"', |
| 'sec-ch-ua-mobile': '?1', |
| 'sec-ch-ua-platform': '"Android"', |
| } |
|
|
| def __init__(self): |
| self.base_url = "https://api.deepinfra.com/v1/openai/chat/completions" |
|
|
| def get_model_list(self): |
| return [ |
| 'meta-llama/Llama-3.3-70B-Instruct-Turbo', |
| 'deepseek-ai/DeepSeek-R1-Turbo', |
| 'deepseek-ai/DeepSeek-R1-Distill-Llama-70B', |
| 'deepseek-ai/DeepSeek-R1-Distill-Qwen-32B' |
| ] |
|
|
| async def generate(self, json_data: dict): |
| json_data['stream'] = True |
| json_data['stream_options'] = { |
| 'include_usage': True, |
| 'continuous_usage_stats': True, |
| } |
|
|
| chunk_id = "chipling-deepinfraoff-" + "".join(random.choices("0123456789abcdef", k=32)) |
| created = int(asyncio.get_event_loop().time()) |
| total_completion_tokens = 0 |
| model_name = json_data.get("model", "unknown") |
|
|
| try: |
| async with httpx.AsyncClient(timeout=None) as client: |
| async with client.stream( |
| "POST", |
| self.base_url, |
| headers=OFFDeepInfraAPI.headers, |
| json=json_data |
| ) as response: |
| if response.status_code != 200: |
| yield f"data: [Unexpected status code: {response.status_code}]\n\n" |
| return |
|
|
| async for line in response.aiter_lines(): |
| if not line or not line.startswith("data:"): |
| continue |
|
|
| data_str = line.removeprefix("data:").strip() |
| if data_str == "[DONE]": |
| yield "data: [DONE]\n\n" |
| return |
|
|
| try: |
| data = json.loads(data_str) |
| delta = data["choices"][0].get("delta", {}) |
| content = delta.get("content", "") |
| finish_reason = data["choices"][0].get("finish_reason", None) |
|
|
| if content or finish_reason: |
| transformed = { |
| "id": chunk_id, |
| "object": "chat.completion.chunk", |
| "created": created, |
| "choices": [{ |
| "index": 0, |
| "text": content, |
| "logprobs": None, |
| "finish_reason": finish_reason, |
| "delta": { |
| "token_id": None, |
| "role": delta.get("role", "assistant"), |
| "content": content, |
| "tool_calls": delta.get("tool_calls"), |
| } |
| }], |
| "model": model_name, |
| "usage": None |
| } |
| yield f"data: {json.dumps(transformed)}\n\n" |
|
|
| |
| usage = data.get("usage") |
| if usage: |
| total_completion_tokens = usage.get("completion_tokens", total_completion_tokens) |
|
|
| except json.JSONDecodeError: |
| continue |
|
|
| |
| final = { |
| "id": chunk_id, |
| "object": "chat.completion.chunk", |
| "created": created, |
| "choices": [], |
| "model": model_name, |
| "usage": { |
| "prompt_tokens": 0, |
| "completion_tokens": total_completion_tokens, |
| "total_tokens": total_completion_tokens |
| } |
| } |
| yield f"data: {json.dumps(final)}\n\n" |
| yield "data: [DONE]\n\n" |
|
|
| except Exception as e: |
| yield f"data: [Connection error: {str(e)}]\n\n" |
