389 lines
19 KiB
Python
389 lines
19 KiB
Python
import asyncio
|
|
import json
|
|
import logging
|
|
from io import BytesIO
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
import aiohttp
|
|
import openai
|
|
|
|
from .ai_responder import AIResponder, async_cache_to_file, exponential_backoff, pp
|
|
from .igdblib import IGDBQuery
|
|
from .leonardo_draw import LeonardoAIDrawMixIn
|
|
|
|
|
|
@async_cache_to_file("openai_chat.dat")
|
|
async def openai_chat(client, *args, **kwargs):
|
|
return await client.chat.completions.create(*args, **kwargs)
|
|
|
|
|
|
@async_cache_to_file("openai_chat.dat")
|
|
async def openai_image(client, *args, **kwargs):
|
|
response = await client.images.generate(*args, **kwargs)
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(response.data[0].url) as image:
|
|
return BytesIO(await image.read())
|
|
|
|
|
|
class OpenAIResponder(AIResponder, LeonardoAIDrawMixIn):
|
|
def __init__(self, config: Dict[str, Any], channel: Optional[str] = None) -> None:
|
|
super().__init__(config, channel)
|
|
self.client = openai.AsyncOpenAI(api_key=self.config.get("openai-token", self.config.get("openai-key", "")))
|
|
|
|
# Initialize IGDB if enabled
|
|
self.igdb = None
|
|
logging.info("IGDB Configuration Check:")
|
|
logging.info(f" enable-game-info: {self.config.get('enable-game-info', 'NOT SET')}")
|
|
logging.info(f" igdb-client-id: {'SET' if self.config.get('igdb-client-id') else 'NOT SET'}")
|
|
logging.info(f" igdb-access-token: {'SET' if self.config.get('igdb-access-token') else 'NOT SET'}")
|
|
|
|
if self.config.get("enable-game-info", False) and self.config.get("igdb-client-id") and self.config.get("igdb-access-token"):
|
|
try:
|
|
self.igdb = IGDBQuery(self.config["igdb-client-id"], self.config["igdb-access-token"])
|
|
logging.info("✅ IGDB integration SUCCESSFULLY enabled for game information")
|
|
logging.info(f" Client ID: {self.config['igdb-client-id'][:8]}...")
|
|
logging.info(f" Available functions: {len(self.igdb.get_openai_functions())}")
|
|
except Exception as e:
|
|
logging.error(f"❌ Failed to initialize IGDB: {e}")
|
|
self.igdb = None
|
|
else:
|
|
logging.warning("❌ IGDB integration DISABLED - missing configuration or disabled in config")
|
|
|
|
async def draw_openai(self, description: str) -> BytesIO:
|
|
for _ in range(3):
|
|
try:
|
|
response = await openai_image(self.client, prompt=description, n=1, size="1024x1024", model="dall-e-3")
|
|
logging.info(f"Drawed a picture with DALL-E on this description: {repr(description)}")
|
|
return response
|
|
except Exception as err:
|
|
logging.warning(f"Failed to generate image {repr(description)}: {repr(err)}")
|
|
raise RuntimeError(f"Failed to generate image {repr(description)} after multiple retries")
|
|
|
|
async def chat(self, messages: List[Dict[str, Any]], limit: int) -> Tuple[Optional[Dict[str, Any]], int]:
|
|
# Safety check for mock objects in tests
|
|
if not isinstance(messages, list) or len(messages) == 0:
|
|
logging.warning("Invalid messages format in chat method")
|
|
return None, limit
|
|
|
|
try:
|
|
# Clean up any orphaned tool messages from previous conversations
|
|
clean_messages = []
|
|
for i, msg in enumerate(messages):
|
|
if msg.get("role") == "tool":
|
|
# Skip tool messages that don't have a corresponding assistant message with tool_calls
|
|
if i == 0 or messages[i - 1].get("role") != "assistant" or not messages[i - 1].get("tool_calls"):
|
|
logging.debug(f"Removing orphaned tool message at position {i}")
|
|
continue
|
|
clean_messages.append(msg)
|
|
messages = clean_messages
|
|
|
|
last_message_content = messages[-1]["content"]
|
|
if isinstance(last_message_content, str):
|
|
model = self.config["model"]
|
|
elif "model-vision" in self.config:
|
|
model = self.config["model-vision"]
|
|
else:
|
|
messages[-1]["content"] = messages[-1]["content"][0]["text"]
|
|
except (KeyError, IndexError, TypeError) as e:
|
|
logging.warning(f"Error accessing message content: {e}")
|
|
return None, limit
|
|
try:
|
|
# Prepare function calls if IGDB is enabled
|
|
chat_kwargs = {
|
|
"model": model,
|
|
"messages": messages,
|
|
}
|
|
|
|
if self.igdb and self.config.get("enable-game-info", False):
|
|
try:
|
|
igdb_functions = self.igdb.get_openai_functions()
|
|
if igdb_functions and isinstance(igdb_functions, list):
|
|
chat_kwargs["tools"] = [{"type": "function", "function": func} for func in igdb_functions]
|
|
chat_kwargs["tool_choice"] = "auto"
|
|
logging.info(f"🎮 IGDB functions available to AI: {[f['name'] for f in igdb_functions]}")
|
|
logging.debug(f" Full chat_kwargs with tools: {list(chat_kwargs.keys())}")
|
|
except (TypeError, AttributeError) as e:
|
|
logging.warning(f"Error setting up IGDB functions: {e}")
|
|
else:
|
|
logging.debug(
|
|
"🎮 IGDB not available for this request (igdb={}, enabled={})".format(
|
|
self.igdb is not None, self.config.get("enable-game-info", False)
|
|
)
|
|
)
|
|
|
|
result = await openai_chat(self.client, **chat_kwargs)
|
|
|
|
# Handle function calls if present
|
|
message = result.choices[0].message
|
|
|
|
# Log what we received from OpenAI
|
|
logging.debug(f"📨 OpenAI Response: content={bool(message.content)}, has_tool_calls={hasattr(message, 'tool_calls')}")
|
|
if hasattr(message, "tool_calls") and message.tool_calls:
|
|
tool_names = [tc.function.name for tc in message.tool_calls]
|
|
logging.info(f"🔧 OpenAI requested function calls: {tool_names}")
|
|
|
|
# Check if we have function/tool calls and IGDB is enabled
|
|
has_tool_calls = (
|
|
hasattr(message, "tool_calls") and message.tool_calls and self.igdb and self.config.get("enable-game-info", False)
|
|
)
|
|
|
|
# Clean up any existing tool messages in the history to avoid conflicts
|
|
if has_tool_calls:
|
|
messages = [msg for msg in messages if msg.get("role") != "tool"]
|
|
|
|
if has_tool_calls:
|
|
logging.info(f"🎮 Processing {len(message.tool_calls)} IGDB function call(s)...")
|
|
try:
|
|
# Process function calls - serialize tool_calls properly
|
|
tool_calls_data = []
|
|
for tc in message.tool_calls:
|
|
tool_calls_data.append(
|
|
{"id": tc.id, "type": "function", "function": {"name": tc.function.name, "arguments": tc.function.arguments}}
|
|
)
|
|
|
|
messages.append({"role": "assistant", "content": message.content or "", "tool_calls": tool_calls_data})
|
|
|
|
# Execute function calls
|
|
for tool_call in message.tool_calls:
|
|
function_name = tool_call.function.name
|
|
function_args = json.loads(tool_call.function.arguments)
|
|
|
|
logging.info(f"🎮 Executing IGDB function: {function_name} with args: {function_args}")
|
|
|
|
# Execute IGDB function
|
|
function_result = await self._execute_igdb_function(function_name, function_args)
|
|
|
|
logging.info(f"🎮 IGDB function result: {type(function_result)} - {str(function_result)[:200]}...")
|
|
|
|
messages.append(
|
|
{
|
|
"role": "tool",
|
|
"tool_call_id": tool_call.id,
|
|
"content": json.dumps(function_result) if function_result else "No results found",
|
|
}
|
|
)
|
|
|
|
# Get final response after function execution - remove tools for final call
|
|
final_chat_kwargs = {
|
|
"model": model,
|
|
"messages": messages,
|
|
}
|
|
logging.debug(f"🔧 Sending final request to OpenAI with {len(messages)} messages (no tools)")
|
|
logging.debug(f"🔧 Last few messages: {messages[-3:] if len(messages) > 3 else messages}")
|
|
|
|
final_result = await openai_chat(self.client, **final_chat_kwargs)
|
|
answer_obj = final_result.choices[0].message
|
|
|
|
logging.debug(
|
|
f"🔧 Final OpenAI response: content_length={len(answer_obj.content) if answer_obj.content else 0}, has_tool_calls={hasattr(answer_obj, 'tool_calls') and answer_obj.tool_calls}"
|
|
)
|
|
if answer_obj.content:
|
|
logging.debug(f"🔧 Response preview: {answer_obj.content[:200]}")
|
|
else:
|
|
logging.warning(f"🔧 OpenAI returned NULL content despite {final_result.usage.completion_tokens} completion tokens")
|
|
|
|
# If OpenAI returns null content after function calling, use empty string
|
|
if not answer_obj.content and function_result:
|
|
logging.warning("OpenAI returned null after function calling, using empty string")
|
|
answer_obj.content = ""
|
|
except Exception as e:
|
|
# If function calling fails, fall back to regular response
|
|
logging.warning(f"Function calling failed, using regular response: {e}")
|
|
answer_obj = message
|
|
else:
|
|
answer_obj = message
|
|
|
|
# Handle null content from OpenAI
|
|
content = answer_obj.content
|
|
if content is None:
|
|
logging.warning("OpenAI returned null content, using empty string")
|
|
content = ""
|
|
|
|
answer = {"content": content, "role": answer_obj.role}
|
|
self.rate_limit_backoff = exponential_backoff()
|
|
logging.info(f"generated response {result.usage}: {repr(answer)}")
|
|
return answer, limit
|
|
except openai.BadRequestError as err:
|
|
if "maximum context length is" in str(err) and limit > 4:
|
|
logging.warning(f"context length exceeded, reduce the limit {limit}: {str(err)}")
|
|
limit -= 1
|
|
return None, limit
|
|
raise err
|
|
except openai.RateLimitError as err:
|
|
rate_limit_sleep = next(self.rate_limit_backoff)
|
|
if "retry-model" in self.config:
|
|
model = self.config["retry-model"]
|
|
logging.warning(f"got an rate limit error, sleep for {rate_limit_sleep} seconds: {str(err)}")
|
|
await asyncio.sleep(rate_limit_sleep)
|
|
except Exception as err:
|
|
import traceback
|
|
|
|
logging.warning(f"failed to generate response: {repr(err)}")
|
|
logging.debug(f"Full traceback: {traceback.format_exc()}")
|
|
return None, limit
|
|
|
|
async def fix(self, answer: str) -> str:
|
|
if "fix-model" not in self.config:
|
|
return answer
|
|
|
|
# Handle null/empty answer
|
|
if not answer:
|
|
logging.warning("Fix called with null/empty answer")
|
|
return '{"answer": "I apologize, I encountered an error processing your request.", "answer_needed": true, "channel": null, "staff": null, "picture": null, "picture_edit": false, "hack": false}'
|
|
messages = [{"role": "system", "content": self.config["fix-description"]}, {"role": "user", "content": answer}]
|
|
try:
|
|
result = await openai_chat(self.client, model=self.config["fix-model"], messages=messages)
|
|
logging.info(f"got this message as fix:\n{pp(result.choices[0].message.content)}")
|
|
response = result.choices[0].message.content
|
|
start, end = response.find("{"), response.rfind("}")
|
|
if start == -1 or end == -1 or (start + 3) >= end:
|
|
return answer
|
|
response = response[start : end + 1]
|
|
logging.info(f"fixed answer:\n{pp(response)}")
|
|
return response
|
|
except Exception as err:
|
|
logging.warning(f"failed to execute a fix for the answer: {repr(err)}")
|
|
return answer
|
|
|
|
async def translate(self, text: str, language: str = "english") -> str:
|
|
if "fix-model" not in self.config:
|
|
return text
|
|
message = [
|
|
{
|
|
"role": "system",
|
|
"content": f"You are an professional translator to {language} language,"
|
|
f" you translate everything you get directly to {language}"
|
|
f" if it is not already in {language}, otherwise you just copy it.",
|
|
},
|
|
{"role": "user", "content": text},
|
|
]
|
|
try:
|
|
result = await openai_chat(self.client, model=self.config["fix-model"], messages=message)
|
|
response = result.choices[0].message.content
|
|
logging.info(f"got this translated message:\n{pp(response)}")
|
|
return response
|
|
except Exception as err:
|
|
logging.warning(f"failed to translate the text: {repr(err)}")
|
|
return text
|
|
|
|
async def memory_rewrite(self, memory: str, message_user: str, answer_user: str, question: str, answer: str) -> str:
|
|
if "memory-model" not in self.config:
|
|
return memory
|
|
messages = [
|
|
{"role": "system", "content": self.config.get("memory-system", "You are an memory assistant.")},
|
|
{
|
|
"role": "user",
|
|
"content": f"Here is my previous memory:\n```\n{memory}\n```\n\n"
|
|
f"Here is my conversanion:\n```\n{message_user}: {question}\n\n{answer_user}: {answer}\n```\n\n"
|
|
f"Please rewrite the memory in a way, that it contain the content mentioned in conversation. "
|
|
f"Summarize the memory if required, try to keep important information. "
|
|
f"Write just new memory data without any comments.",
|
|
},
|
|
]
|
|
logging.info(f"Rewrite memory:\n{pp(messages)}")
|
|
try:
|
|
# logging.info(f'send this memory request:\n{pp(messages)}')
|
|
result = await openai_chat(self.client, model=self.config["memory-model"], messages=messages)
|
|
new_memory = result.choices[0].message.content
|
|
logging.info(f"new memory:\n{new_memory}")
|
|
return new_memory
|
|
except Exception as err:
|
|
logging.warning(f"failed to create new memory: {repr(err)}")
|
|
return memory
|
|
|
|
async def _execute_igdb_function(self, function_name: str, function_args: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Execute IGDB function calls from OpenAI.
|
|
"""
|
|
logging.info(f"🎮 _execute_igdb_function called: {function_name}")
|
|
|
|
if not self.igdb:
|
|
logging.error("🎮 IGDB function called but self.igdb is None!")
|
|
return {"error": "IGDB not available"}
|
|
|
|
try:
|
|
if function_name == "search_games":
|
|
query = function_args.get("query", "")
|
|
limit = function_args.get("limit", 5)
|
|
|
|
logging.info(f"🎮 Searching IGDB for: '{query}' (limit: {limit})")
|
|
|
|
if not query:
|
|
logging.warning("🎮 No search query provided to search_games")
|
|
return {"error": "No search query provided"}
|
|
|
|
results = self.igdb.search_games(query, limit)
|
|
logging.info(f"🎮 IGDB search returned: {len(results) if results and isinstance(results, list) else 0} results")
|
|
|
|
if results and isinstance(results, list) and len(results) > 0:
|
|
return {"games": results}
|
|
else:
|
|
return {"games": [], "message": f"No games found matching '{query}'"}
|
|
|
|
elif function_name == "get_games_by_release_date":
|
|
year = function_args.get("year")
|
|
month = function_args.get("month")
|
|
platform = function_args.get("platform")
|
|
limit = function_args.get("limit", 10)
|
|
|
|
logging.info(
|
|
f"🎮 Searching IGDB for games releasing in {year}/{month or 'all'} on {platform or 'all platforms'} (limit: {limit})"
|
|
)
|
|
|
|
if not year:
|
|
logging.warning("🎮 No year provided to get_games_by_release_date")
|
|
return {"error": "No year provided"}
|
|
|
|
results = self.igdb.get_games_by_release_date(year, month, platform, limit)
|
|
logging.info(f"🎮 IGDB release date search returned: {len(results) if results and isinstance(results, list) else 0} results")
|
|
|
|
if results and isinstance(results, list) and len(results) > 0:
|
|
return {"games": results}
|
|
else:
|
|
period = f"{year}/{month}" if month else str(year)
|
|
platform_text = f" on {platform}" if platform else ""
|
|
return {"games": [], "message": f"No games found releasing in {period}{platform_text}"}
|
|
|
|
elif function_name == "get_games_by_platform":
|
|
platform = function_args.get("platform", "")
|
|
genre = function_args.get("genre")
|
|
limit = function_args.get("limit", 10)
|
|
|
|
logging.info(f"🎮 Searching IGDB for games on {platform} {f'in {genre} genre' if genre else ''} (limit: {limit})")
|
|
|
|
if not platform:
|
|
logging.warning("🎮 No platform provided to get_games_by_platform")
|
|
return {"error": "No platform provided"}
|
|
|
|
results = self.igdb.get_games_by_platform(platform, genre, limit)
|
|
logging.info(f"🎮 IGDB platform search returned: {len(results) if results and isinstance(results, list) else 0} results")
|
|
|
|
if results and isinstance(results, list) and len(results) > 0:
|
|
return {"games": results}
|
|
else:
|
|
genre_text = f" in {genre} genre" if genre else ""
|
|
return {"games": [], "message": f"No games found for {platform}{genre_text}"}
|
|
|
|
elif function_name == "get_game_details":
|
|
game_id = function_args.get("game_id")
|
|
|
|
logging.info(f"🎮 Getting IGDB details for game ID: {game_id}")
|
|
|
|
if not game_id:
|
|
logging.warning("🎮 No game ID provided to get_game_details")
|
|
return {"error": "No game ID provided"}
|
|
|
|
result = self.igdb.get_game_details(game_id)
|
|
logging.info(f"🎮 IGDB game details returned: {bool(result)}")
|
|
|
|
if result:
|
|
return {"game": result}
|
|
else:
|
|
return {"error": f"Game with ID {game_id} not found"}
|
|
else:
|
|
return {"error": f"Unknown function: {function_name}"}
|
|
|
|
except Exception as e:
|
|
logging.error(f"Error executing IGDB function {function_name}: {e}")
|
|
return {"error": f"Failed to execute {function_name}: {str(e)}"}
|