Spaces:
Sleeping
Sleeping
File size: 6,335 Bytes
aa04092 6c4292b aa04092 6c4292b aa04092 6c4292b aa04092 6c4292b af01c91 aa04092 337ec9f aa04092 6c4292b 337ec9f aa04092 337ec9f aa04092 337ec9f aa04092 337ec9f aa04092 337ec9f aa04092 337ec9f aa04092 6c4292b |
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 |
import os
from typing import List, Dict
import openai
from config import Config
from openrouter_client import OpenRouterClient
from deepseek_client import DeepseekClient
class LLMClient:
def __init__(self):
self.openai_client = None
self.openrouter_client = None
self.deepseek_client = None
self.test_mode = False # For testing error scenarios
self.initialize_clients()
def initialize_clients(self):
"""Initialize API clients with proper error handling"""
try:
openai_key = Config.get_openai_key()
openrouter_key = Config.get_openrouter_key()
deepseek_key = Config.get_deepseek_key()
if openai_key:
self.openai_client = openai.OpenAI(api_key=openai_key)
if openrouter_key:
self.openrouter_client = OpenRouterClient()
if deepseek_key:
try:
self.deepseek_client = DeepseekClient(api_key=deepseek_key)
except Exception as e:
print(f"Failed to initialize DeepseekClient: {str(e)}")
print("Check your Deepseek API key and endpoint configuration")
raise
except Exception as e:
print(f"Error initializing API clients: {str(e)}")
def set_test_mode(self, enabled: bool = True):
"""Enable or disable test mode for simulating errors"""
self.test_mode = enabled
def chat_auto(self, messages: List[Dict[str, str]]) -> str:
"""Send chat completion request to Auto via OpenRouter"""
if not self.openrouter_client:
raise ValueError("OpenRouter client not initialized. Please check your API key.")
try:
return self.openrouter_client.create(
messages=messages,
model=Config.AUTO_MODEL
)
except Exception as e:
raise Exception(f"OpenRouter AUTO API error: {str(e)}")
def chat_openai(self, messages: List[Dict[str, str]]) -> str:
"""Send chat completion request to OpenAI API"""
if not self.openai_client:
raise ValueError("OpenAI client not initialized. Please check your API key.")
if self.test_mode:
if messages and "test_error" in messages[-1].get("content", "").lower():
error_type = messages[-1]["content"].lower()
if "api_key" in error_type:
raise ValueError("Invalid API key")
elif "rate_limit" in error_type:
raise Exception("Rate limit exceeded")
elif "network" in error_type:
raise Exception("Network connection error")
try:
response = self.openai_client.chat.completions.create(
model=Config.OPENAI_MODEL,
messages=messages
)
return response.choices[0].message.content
except openai.APIError as e:
raise Exception(f"OpenAI API error: {str(e)}")
def chat_deepseek(self, messages: List[Dict[str, str]]) -> str:
"""Send chat completion request to Deepseek API"""
if not self.deepseek_client:
raise ValueError("Deepseek client not initialized. Please check your API key.")
if self.test_mode:
if messages and "test_error" in messages[-1].get("content", "").lower():
error_type = messages[-1]["content"].lower()
if "api_key" in error_type:
raise ValueError("Invalid API key")
elif "rate_limit" in error_type:
raise Exception("Rate limit exceeded")
elif "network" in error_type:
raise Exception("Network connection error")
try:
return self.deepseek_client.create(
messages=messages,
model=Config.DEEPSEEK_MODEL
)
except Exception as e:
raise Exception(f"Deepseek API error: {str(e)}")
def chat_claude(self, messages: List[Dict[str, str]]) -> str:
"""Send chat completion request to Claude via OpenRouter"""
if not self.openrouter_client:
raise ValueError("OpenRouter client not initialized. Please check your API key.")
try:
return self.openrouter_client.create(
messages=messages,
model=Config.CLAUDE_MODEL
)
except Exception as e:
raise Exception(f"Claude API error: {str(e)}")
def chat_gemini(self, messages: List[Dict[str, str]]) -> str:
"""Send chat completion request to Gemini via OpenRouter"""
if not self.openrouter_client:
raise ValueError("OpenRouter client not initialized. Please check your API key.")
try:
return self.openrouter_client.create(
messages=messages,
model=Config.GEMINI_MODEL
)
except Exception as e:
raise Exception(f"Gemini API error: {str(e)}")
def generate_context_summary(self, messages: List[Dict[str, str]]) -> str:
"""Generate a summary of the conversation context."""
if not self.openai_client:
print("OpenAI client not initialized. Skipping context summary.")
return ""
try:
summary_prompt = {
"role": "system",
"content": (
"Summarize the key points of this conversation in a concise way. "
"Focus on the main topics and important details. "
f"Keep the summary within {Config.MEMORY_SUMMARY_TOKENS} tokens."
)
}
# Create a list of messages for summarization
summary_messages = [summary_prompt] + messages[-Config.CONTEXT_WINDOW_MESSAGES:]
response = self.openai_client.chat.completions.create(
model=Config.OPENAI_MODEL,
messages=summary_messages,
max_tokens=Config.MEMORY_SUMMARY_TOKENS
)
return response.choices[0].message.content
except Exception as e:
print(f"Failed to generate context summary: {str(e)}")
return "" # Return empty string if summarization fails
|