Spaces:
Runtime error
Runtime error
File size: 9,223 Bytes
06cb2a3 |
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 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 |
"""
Game Recap - LangChain tool for retrieving and generating game recaps
This module provides functions to:
1. Search for games in Neo4j based on natural language queries
2. Generate game recaps from the structured data
3. Return both text summaries and data for UI components
"""
# Import Gradio-specific modules directly
import sys
import os
# Add parent directory to path to access gradio modules
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from gradio_llm import llm
from gradio_graph import graph
from langchain_neo4j import GraphCypherQAChain
from langchain_core.prompts import PromptTemplate, ChatPromptTemplate
# Create a global variable to store the last retrieved game data
# This is a workaround for LangChain dropping structured data
LAST_GAME_DATA = None
# Function to get the cached game data
def get_last_game_data():
global LAST_GAME_DATA
return LAST_GAME_DATA
# Function to set the cached game data
def set_last_game_data(game_data):
global LAST_GAME_DATA
LAST_GAME_DATA = game_data
print(f"STORED GAME DATA IN CACHE: {game_data}")
# Create the Cypher generation prompt for game search
GAME_SEARCH_TEMPLATE = """
You are an expert Neo4j Developer translating user questions about NFL games into Cypher queries.
Your goal is to find a specific game in the database based on the user's description.
Convert the user's question based on the schema.
IMPORTANT NOTES:
1. Always return the FULL game node with ALL its properties.
2. Always use case-insensitive comparisons in your Cypher queries by applying toLower() to both the property and the search string.
3. If the question mentions a specific date, look for games on that date.
4. If the question mentions teams, look for games where those teams played.
5. If the question uses phrases like "last game", "most recent game", etc., you should add an ORDER BY clause.
6. NEVER use the embedding property in your queries.
7. ALWAYS include "g.game_id, g.date, g.location, g.home_team, g.away_team, g.result, g.summary, g.home_team_logo_url, g.away_team_logo_url, g.highlight_video_url" in your RETURN statement.
Example Questions and Queries:
1. "Tell me about the 49ers game against the Jets"
```
MATCH (g:Game)
WHERE (toLower(g.home_team) CONTAINS toLower("49ers") AND toLower(g.away_team) CONTAINS toLower("Jets"))
OR (toLower(g.away_team) CONTAINS toLower("49ers") AND toLower(g.home_team) CONTAINS toLower("Jets"))
RETURN g.game_id, g.date, g.location, g.home_team, g.away_team, g.result, g.summary,
g.home_team_logo_url, g.away_team_logo_url, g.highlight_video_url
```
2. "What happened in the 49ers game on October 9th?"
```
MATCH (g:Game)
WHERE (toLower(g.home_team) CONTAINS toLower("49ers") OR toLower(g.away_team) CONTAINS toLower("49ers"))
AND toLower(g.date) CONTAINS toLower("10/09")
RETURN g.game_id, g.date, g.location, g.home_team, g.away_team, g.result, g.summary,
g.home_team_logo_url, g.away_team_logo_url, g.highlight_video_url
```
3. "Show me the most recent 49ers game"
```
MATCH (g:Game)
WHERE (toLower(g.home_team) CONTAINS toLower("49ers") OR toLower(g.away_team) CONTAINS toLower("49ers"))
RETURN g.game_id, g.date, g.location, g.home_team, g.away_team, g.result, g.summary,
g.home_team_logo_url, g.away_team_logo_url, g.highlight_video_url
ORDER BY g.date DESC
LIMIT 1
```
Schema:
{schema}
Question:
{question}
"""
game_search_prompt = PromptTemplate.from_template(GAME_SEARCH_TEMPLATE)
# Create the game recap generation prompt
GAME_RECAP_TEMPLATE = """
You are a professional sports commentator for the NFL. Write an engaging and informative recap of the game described below.
Game Details:
- Date: {date}
- Location: {location}
- Home Team: {home_team}
- Away Team: {away_team}
- Final Score: {result}
- Summary: {summary}
Instructions:
1. Begin with an attention-grabbing opening that mentions both teams and the outcome.
2. Include key moments from the summary if available.
3. Mention the venue/location.
4. Conclude with what this means for the teams going forward.
5. Keep the tone professional and engaging - like an ESPN or NFL Network broadcast.
6. Write 2-3 paragraphs maximum.
7. If the 49ers are one of the teams, focus slightly more on their perspective.
8. IMPORTANT: Do NOT include any Markdown images, team logos, or links in your text response. Provide text only.
Write your recap:
"""
recap_prompt = PromptTemplate.from_template(GAME_RECAP_TEMPLATE)
# Create the Cypher QA chain for game search
game_search = GraphCypherQAChain.from_llm(
llm,
graph=graph,
verbose=True,
cypher_prompt=game_search_prompt,
return_direct=True, # Return the raw results instead of passing through LLM
allow_dangerous_requests=True # Required to enable Cypher queries
)
# Function to parse game data from Cypher result
def parse_game_data(result):
"""Parse the game data from the Cypher result into a structured format."""
if not result or not isinstance(result, list) or len(result) == 0:
return None
game = result[0]
# Extract home and away teams to determine winner
home_team = game.get('g.home_team', '')
away_team = game.get('g.away_team', '')
result_str = game.get('g.result', 'N/A')
# Parse the score if available
home_score = away_score = 'N/A'
winner = None
if result_str and result_str != 'N/A':
try:
scores = result_str.split('-')
if len(scores) == 2:
home_score = scores[0].strip()
away_score = scores[1].strip()
# Determine winner
home_score_int = int(home_score)
away_score_int = int(away_score)
winner = 'home' if home_score_int > away_score_int else 'away'
except (ValueError, IndexError):
pass
# Build the structured game data
game_data = {
'game_id': game.get('g.game_id', ''),
'date': game.get('g.date', ''),
'location': game.get('g.location', ''),
'home_team': home_team,
'away_team': away_team,
'home_score': home_score,
'away_score': away_score,
'result': result_str,
'winner': winner,
'summary': game.get('g.summary', ''),
'home_team_logo_url': game.get('g.home_team_logo_url', ''),
'away_team_logo_url': game.get('g.away_team_logo_url', ''),
'highlight_video_url': game.get('g.highlight_video_url', '')
}
return game_data
# Function to generate a game recap using LLM
def generate_game_recap(game_data):
"""Generate a natural language recap of the game using the LLM."""
if not game_data:
return "I couldn't find information about that game."
# Format the prompt with game data
formatted_prompt = recap_prompt.format(
date=game_data.get('date', 'N/A'),
location=game_data.get('location', 'N/A'),
home_team=game_data.get('home_team', 'N/A'),
away_team=game_data.get('away_team', 'N/A'),
result=game_data.get('result', 'N/A'),
summary=game_data.get('summary', 'N/A')
)
# Generate the recap using the LLM
recap = llm.invoke(formatted_prompt)
return recap.content if hasattr(recap, 'content') else str(recap)
# Main function to search for a game and generate a recap
def game_recap_qa(input_text):
"""
Search for a game based on the input text and generate a recap.
Args:
input_text (str): Natural language query about a game
Returns:
dict: Response containing text recap and structured game data
"""
try:
# Log the incoming query
print(f"Processing game recap query: {input_text}")
# Search for the game
search_result = game_search.invoke({"query": input_text})
# Check if we have a result
if not search_result or not search_result.get('result'):
return {
"output": "I couldn't find information about that game. Could you provide more details?",
"game_data": None
}
# Parse the game data
game_data = parse_game_data(search_result.get('result'))
if not game_data:
return {
"output": "I found information about the game, but couldn't process it correctly.",
"game_data": None
}
# Generate the recap
recap_text = generate_game_recap(game_data)
# CRITICAL: Store the game data in our cache so it can be retrieved later
# This is a workaround for LangChain dropping structured data
set_last_game_data(game_data)
# Return both the text and structured data
return {
"output": recap_text,
"game_data": game_data
}
except Exception as e:
print(f"Error in game_recap_qa: {str(e)}")
import traceback
traceback.print_exc()
return {
"output": "I encountered an error while searching for the game. Please try again with a different query.",
"game_data": None
} |