Implement comprehensive IGDB integration for real-time game information
## Major Features Added - **Enhanced igdblib.py**: * Added search_games() method with fuzzy game search * Added get_game_details() for comprehensive game information * Added AI-friendly data formatting with _format_game_for_ai() * Added OpenAI function definitions via get_openai_functions() - **OpenAI Function Calling Integration**: * Modified OpenAIResponder to support function calling * Added IGDB function execution with _execute_igdb_function() * Backward compatible - gracefully falls back if IGDB unavailable * Auto-detects gaming queries and fetches real-time data - **Configuration & Setup**: * Added IGDB configuration options to config.toml * Updated system prompt to inform AI of gaming capabilities * Added comprehensive IGDB_SETUP.md documentation * Graceful initialization with proper error handling ## Technical Implementation - **Function Calling**: Uses OpenAI's tools/function calling API - **Smart Game Search**: Includes ratings, platforms, developers, genres - **Error Handling**: Robust fallbacks and logging - **Data Formatting**: Optimized for AI comprehension and user presentation - **Rate Limiting**: Respects IGDB API limits ## Usage Users can now ask natural gaming questions: - "Tell me about Elden Ring" - "What are good RPG games from 2023?" - "Is Cyberpunk 2077 on PlayStation?" The AI automatically detects gaming queries, calls IGDB API, and presents accurate, real-time game information seamlessly. 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -1,4 +1,6 @@
|
||||
import logging
|
||||
from functools import cache
|
||||
from typing import Any, Dict, List, Optional, Union
|
||||
|
||||
import requests
|
||||
|
||||
@@ -89,3 +91,173 @@ class IGDBQuery(object):
|
||||
limit=100,
|
||||
)
|
||||
return game_info
|
||||
|
||||
def search_games(self, query: str, limit: int = 5) -> Optional[List[Dict[str, Any]]]:
|
||||
"""
|
||||
Search for games with a flexible query string.
|
||||
Returns formatted game information suitable for AI responses.
|
||||
"""
|
||||
if not query or not query.strip():
|
||||
return None
|
||||
|
||||
try:
|
||||
# Search for games with fuzzy matching
|
||||
games = self.generalized_igdb_query(
|
||||
{"name": query.strip()},
|
||||
"games",
|
||||
[
|
||||
"id", "name", "summary", "storyline", "rating", "aggregated_rating",
|
||||
"first_release_date", "genres.name", "platforms.name", "developers.name",
|
||||
"publishers.name", "game_modes.name", "themes.name", "cover.url"
|
||||
],
|
||||
additional_filters={"category": "= 0"}, # Main games only
|
||||
limit=limit
|
||||
)
|
||||
|
||||
if not games:
|
||||
return None
|
||||
|
||||
# Format games for AI consumption
|
||||
formatted_games = []
|
||||
for game in games:
|
||||
formatted_game = self._format_game_for_ai(game)
|
||||
if formatted_game:
|
||||
formatted_games.append(formatted_game)
|
||||
|
||||
return formatted_games if formatted_games else None
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"Error searching games for query '{query}': {e}")
|
||||
return None
|
||||
|
||||
def get_game_details(self, game_id: int) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Get detailed information about a specific game by ID.
|
||||
"""
|
||||
try:
|
||||
games = self.generalized_igdb_query(
|
||||
{},
|
||||
"games",
|
||||
[
|
||||
"id", "name", "summary", "storyline", "rating", "aggregated_rating",
|
||||
"first_release_date", "genres.name", "platforms.name", "developers.name",
|
||||
"publishers.name", "game_modes.name", "themes.name", "keywords.name",
|
||||
"similar_games.name", "cover.url", "screenshots.url", "videos.video_id",
|
||||
"release_dates.date", "release_dates.platform.name", "age_ratings.rating"
|
||||
],
|
||||
additional_filters={"id": f"= {game_id}"},
|
||||
limit=1
|
||||
)
|
||||
|
||||
if games and len(games) > 0:
|
||||
return self._format_game_for_ai(games[0], detailed=True)
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"Error getting game details for ID {game_id}: {e}")
|
||||
|
||||
return None
|
||||
|
||||
def _format_game_for_ai(self, game_data: Dict[str, Any], detailed: bool = False) -> Dict[str, Any]:
|
||||
"""
|
||||
Format game data in a way that's easy for AI to understand and present to users.
|
||||
"""
|
||||
try:
|
||||
formatted = {
|
||||
"name": game_data.get("name", "Unknown"),
|
||||
"summary": game_data.get("summary", "No summary available")
|
||||
}
|
||||
|
||||
# Add basic info
|
||||
if "rating" in game_data:
|
||||
formatted["rating"] = f"{game_data['rating']:.1f}/100"
|
||||
if "aggregated_rating" in game_data:
|
||||
formatted["user_rating"] = f"{game_data['aggregated_rating']:.1f}/100"
|
||||
|
||||
# Release information
|
||||
if "first_release_date" in game_data:
|
||||
import datetime
|
||||
release_date = datetime.datetime.fromtimestamp(game_data["first_release_date"])
|
||||
formatted["release_year"] = release_date.year
|
||||
if detailed:
|
||||
formatted["release_date"] = release_date.strftime("%Y-%m-%d")
|
||||
|
||||
# Platforms
|
||||
if "platforms" in game_data and game_data["platforms"]:
|
||||
platforms = [p.get("name", "") for p in game_data["platforms"] if p.get("name")]
|
||||
formatted["platforms"] = platforms[:5] # Limit to prevent overflow
|
||||
|
||||
# Genres
|
||||
if "genres" in game_data and game_data["genres"]:
|
||||
genres = [g.get("name", "") for g in game_data["genres"] if g.get("name")]
|
||||
formatted["genres"] = genres
|
||||
|
||||
# Developers
|
||||
if "developers" in game_data and game_data["developers"]:
|
||||
developers = [d.get("name", "") for d in game_data["developers"] if d.get("name")]
|
||||
formatted["developers"] = developers[:3] # Limit for readability
|
||||
|
||||
# Publishers
|
||||
if "publishers" in game_data and game_data["publishers"]:
|
||||
publishers = [p.get("name", "") for p in game_data["publishers"] if p.get("name")]
|
||||
formatted["publishers"] = publishers[:2]
|
||||
|
||||
if detailed:
|
||||
# Add more detailed info for specific requests
|
||||
if "storyline" in game_data and game_data["storyline"]:
|
||||
formatted["storyline"] = game_data["storyline"]
|
||||
|
||||
if "game_modes" in game_data and game_data["game_modes"]:
|
||||
modes = [m.get("name", "") for m in game_data["game_modes"] if m.get("name")]
|
||||
formatted["game_modes"] = modes
|
||||
|
||||
if "themes" in game_data and game_data["themes"]:
|
||||
themes = [t.get("name", "") for t in game_data["themes"] if t.get("name")]
|
||||
formatted["themes"] = themes
|
||||
|
||||
return formatted
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"Error formatting game data: {e}")
|
||||
return {"name": game_data.get("name", "Unknown"), "summary": "Error retrieving game information"}
|
||||
|
||||
def get_openai_functions(self) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Generate OpenAI function definitions for game-related queries.
|
||||
Returns function definitions that OpenAI can use to call IGDB API.
|
||||
"""
|
||||
return [
|
||||
{
|
||||
"name": "search_games",
|
||||
"description": "Search for video games by name or title. Use when users ask about specific games, game recommendations, or want to know about games.",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"query": {
|
||||
"type": "string",
|
||||
"description": "The game name or search query (e.g., 'Elden Ring', 'Mario', 'Zelda Breath of the Wild')"
|
||||
},
|
||||
"limit": {
|
||||
"type": "integer",
|
||||
"description": "Maximum number of games to return (default: 5, max: 10)",
|
||||
"minimum": 1,
|
||||
"maximum": 10
|
||||
}
|
||||
},
|
||||
"required": ["query"]
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "get_game_details",
|
||||
"description": "Get detailed information about a specific game when you have its ID from a previous search.",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"game_id": {
|
||||
"type": "integer",
|
||||
"description": "The IGDB game ID from a previous search result"
|
||||
}
|
||||
},
|
||||
"required": ["game_id"]
|
||||
}
|
||||
}
|
||||
]
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
from io import BytesIO
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
@@ -7,6 +8,7 @@ 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
|
||||
|
||||
|
||||
@@ -27,6 +29,21 @@ 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
|
||||
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 enabled for game information")
|
||||
except Exception as e:
|
||||
logging.warning(f"Failed to initialize IGDB: {e}")
|
||||
self.igdb = None
|
||||
|
||||
async def draw_openai(self, description: str) -> BytesIO:
|
||||
for _ in range(3):
|
||||
@@ -46,12 +63,61 @@ class OpenAIResponder(AIResponder, LeonardoAIDrawMixIn):
|
||||
else:
|
||||
messages[-1]["content"] = messages[-1]["content"][0]["text"]
|
||||
try:
|
||||
result = await openai_chat(
|
||||
self.client,
|
||||
model=model,
|
||||
messages=messages,
|
||||
)
|
||||
answer_obj = result.choices[0].message
|
||||
# Prepare function calls if IGDB is enabled
|
||||
chat_kwargs = {
|
||||
"model": model,
|
||||
"messages": messages,
|
||||
}
|
||||
|
||||
if self.igdb and self.config.get("enable-game-info", False):
|
||||
chat_kwargs["tools"] = [
|
||||
{"type": "function", "function": func}
|
||||
for func in self.igdb.get_openai_functions()
|
||||
]
|
||||
chat_kwargs["tool_choice"] = "auto"
|
||||
|
||||
result = await openai_chat(self.client, **chat_kwargs)
|
||||
|
||||
# Handle function calls if present
|
||||
message = result.choices[0].message
|
||||
|
||||
# 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))
|
||||
|
||||
if has_tool_calls:
|
||||
try:
|
||||
# Process function calls
|
||||
messages.append({
|
||||
"role": "assistant",
|
||||
"content": message.content or "",
|
||||
"tool_calls": [tc.dict() if hasattr(tc, 'dict') else tc for tc in message.tool_calls]
|
||||
})
|
||||
|
||||
# Execute function calls
|
||||
for tool_call in message.tool_calls:
|
||||
function_name = tool_call.function.name
|
||||
function_args = json.loads(tool_call.function.arguments)
|
||||
|
||||
# Execute IGDB function
|
||||
function_result = await self._execute_igdb_function(function_name, function_args)
|
||||
|
||||
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
|
||||
final_result = await openai_chat(self.client, **chat_kwargs)
|
||||
answer_obj = final_result.choices[0].message
|
||||
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
|
||||
|
||||
answer = {"content": answer_obj.content, "role": answer_obj.role}
|
||||
self.rate_limit_backoff = exponential_backoff()
|
||||
logging.info(f"generated response {result.usage}: {repr(answer)}")
|
||||
@@ -135,3 +201,42 @@ class OpenAIResponder(AIResponder, LeonardoAIDrawMixIn):
|
||||
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.
|
||||
"""
|
||||
if not self.igdb:
|
||||
return None
|
||||
|
||||
try:
|
||||
if function_name == "search_games":
|
||||
query = function_args.get("query", "")
|
||||
limit = function_args.get("limit", 5)
|
||||
|
||||
if not query:
|
||||
return {"error": "No search query provided"}
|
||||
|
||||
results = self.igdb.search_games(query, limit)
|
||||
if results:
|
||||
return {"games": results}
|
||||
else:
|
||||
return {"games": [], "message": f"No games found matching '{query}'"}
|
||||
|
||||
elif function_name == "get_game_details":
|
||||
game_id = function_args.get("game_id")
|
||||
|
||||
if not game_id:
|
||||
return {"error": "No game ID provided"}
|
||||
|
||||
result = self.igdb.get_game_details(game_id)
|
||||
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)}"}
|
||||
|
||||
Reference in New Issue
Block a user