Spaces:
Sleeping
Sleeping
File size: 5,558 Bytes
98777b4 | 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 | """
LLM Connector - Multi-Provider Intelligence
=============================================
Connects to available LLM APIs for content generation and reasoning.
Falls back gracefully between providers.
"""
import json
import logging
import urllib.request
import urllib.error
from typing import Optional
logger = logging.getLogger("openclaw.llm")
class LLMConnector:
"""Multi-provider LLM connector."""
PROVIDERS = {
"groq": {
"url": "https://api.groq.com/openai/v1/chat/completions",
"model": "llama-3.3-70b-versatile",
"header_key": "Authorization",
"header_prefix": "Bearer ",
},
"gemini": {
"url": "https://generativelanguage.googleapis.com/v1beta/models/gemini-2.0-flash:generateContent",
"model": "gemini-2.0-flash",
"header_key": "x-goog-api-key",
"header_prefix": "",
},
"nvidia": {
"url": "https://integrate.api.nvidia.com/v1/chat/completions",
"model": "meta/llama-3.1-70b-instruct",
"header_key": "Authorization",
"header_prefix": "Bearer ",
},
}
def __init__(self, provider: str, api_key: str):
self.provider = provider
self.api_key = api_key
self.config = self.PROVIDERS.get(provider, {})
def generate(self, prompt: str, system: str = "", max_tokens: int = 1024, temperature: float = 0.7) -> Optional[str]:
"""Generate text using the configured LLM."""
if not self.api_key or not self.config:
logger.warning(f"LLM provider '{self.provider}' not configured")
return None
try:
if self.provider == "gemini":
return self._generate_gemini(prompt, system, max_tokens, temperature)
else:
return self._generate_openai_compat(prompt, system, max_tokens, temperature)
except Exception as e:
logger.error(f"LLM generation failed ({self.provider}): {e}")
return None
def _generate_openai_compat(self, prompt: str, system: str, max_tokens: int, temperature: float) -> Optional[str]:
"""Generate using OpenAI-compatible API (Groq, NVIDIA)."""
messages = []
if system:
messages.append({"role": "system", "content": system})
messages.append({"role": "user", "content": prompt})
data = json.dumps({
"model": self.config["model"],
"messages": messages,
"max_tokens": max_tokens,
"temperature": temperature,
}).encode()
headers = {
"Content-Type": "application/json",
self.config["header_key"]: f"{self.config['header_prefix']}{self.api_key}",
}
req = urllib.request.Request(self.config["url"], data=data, headers=headers, method="POST")
with urllib.request.urlopen(req, timeout=60) as resp:
result = json.loads(resp.read().decode())
return result["choices"][0]["message"]["content"]
def _generate_gemini(self, prompt: str, system: str, max_tokens: int, temperature: float) -> Optional[str]:
"""Generate using Google Gemini API."""
url = f"{self.config['url']}?key={self.api_key}"
parts = []
if system:
parts.append({"text": f"System: {system}\n\nUser: {prompt}"})
else:
parts.append({"text": prompt})
data = json.dumps({
"contents": [{"parts": parts}],
"generationConfig": {
"maxOutputTokens": max_tokens,
"temperature": temperature,
}
}).encode()
headers = {"Content-Type": "application/json"}
req = urllib.request.Request(url, data=data, headers=headers, method="POST")
with urllib.request.urlopen(req, timeout=60) as resp:
result = json.loads(resp.read().decode())
return result["candidates"][0]["content"]["parts"][0]["text"]
class MultiLLM:
"""Try multiple LLM providers in order, with key rotation."""
def __init__(self, providers: dict[str, str]):
"""providers: dict of {provider_name: api_key} or {provider_name: 'key1,key2,key3'}"""
self.connectors = []
# Priority order: nvidia (working), groq (fast), gemini (free)
for name in ["nvidia", "groq", "gemini"]:
if name in providers and providers[name]:
# Support comma-separated multiple keys
keys = [k.strip() for k in providers[name].split(",") if k.strip()]
for key in keys:
self.connectors.append(LLMConnector(name, key))
def generate(self, prompt: str, system: str = "", max_tokens: int = 1024, temperature: float = 0.7) -> str:
"""Try each provider until one works."""
for connector in self.connectors:
try:
result = connector.generate(prompt, system, max_tokens, temperature)
if result:
logger.info(f"LLM response from {connector.provider}")
return result
except Exception as e:
logger.warning(f"Provider {connector.provider} failed: {e}")
continue
logger.warning("All LLM providers failed, using template")
return ""
@property
def available(self) -> bool:
return len(self.connectors) > 0
|