Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import os | |
| import time | |
| from typing import List, Dict, TypedDict, Any, Generator | |
| import gradio as gr | |
| from tavily import TavilyClient | |
| from openai import AzureOpenAI | |
| from dotenv import load_dotenv | |
| # ---------- Load environment ---------- | |
| load_dotenv() | |
| print("✅ Environment variables loaded.") | |
| # ---------- State ---------- | |
| class ResearchState(TypedDict): | |
| query: str | |
| expanded_queries: List[str] | |
| results: List[Dict[str, Any]] | |
| summary_md: str | |
| # ---------- Config ---------- | |
| AZURE_ENDPOINT = os.getenv("AZURE_OPENAI_ENDPOINT") | |
| AZURE_API_KEY = os.getenv("AZURE_OPENAI_API_KEY") | |
| AZURE_API_VERSION = os.getenv("AZURE_OPENAI_API_VERSION") | |
| AZURE_DEPLOYMENT = os.getenv("AZURE_OPENAI_DEPLOYMENT") | |
| TAVILY_API_KEY = os.getenv("TAVILY_API_KEY") | |
| # ---------- Helpers ---------- | |
| def _get_azure_client() -> AzureOpenAI: | |
| """Initialize Azure OpenAI client""" | |
| print("🔑 Connecting to Azure OpenAI...") | |
| return AzureOpenAI( | |
| api_version=AZURE_API_VERSION, | |
| azure_endpoint=AZURE_ENDPOINT, | |
| api_key=AZURE_API_KEY, | |
| ) | |
| def _expand_queries(base_query: str) -> List[str]: | |
| """Expand query into detailed football-related search topics""" | |
| topics = [ | |
| "match result and performance analysis", | |
| "team and player statistics", | |
| "injuries, lineups, and latest team news", | |
| "odds and betting market overview", | |
| "weather conditions, referee, and travel factors", | |
| "tactical analysis or game plan strategy", | |
| "coach win rate and managerial history", | |
| "head-to-head historical stats and rivalry trends", | |
| "first 5 and last 5 minutes match dynamics", | |
| "discipline and yellow card referee patterns", | |
| "corner and throw-in frequency stats", | |
| "midweek fixtures, fatigue, and FIFA day effects", | |
| "domestic cup or side competitions relevance", | |
| "psychological edge or 'bogey team' analysis", | |
| ] | |
| return [f"{base_query} - {topic}" for topic in topics] | |
| # ---------- Main Stream Function ---------- | |
| def stream_research(user_query: str) -> Generator: | |
| """Stream step-by-step process to UI.""" | |
| if not user_query.strip(): | |
| yield "⚠️ Please enter a valid football query.", "", "", "", "", "❌ Waiting for input..." | |
| return | |
| print(f"🧾 USER QUERY: {user_query}") | |
| phase = "🔄 Expanding queries..." | |
| yield "## ⏳ Running Analysis...", "", "", "", "", phase | |
| # === 1️⃣ Expand Queries === | |
| expanded = _expand_queries(user_query) | |
| expanded_md = "### 🧩 Expanded Queries\n" + "\n".join([f"- {q}" for q in expanded]) | |
| yield "", expanded_md, "", "", "", "🧠 Step 1/3: Query Expansion" | |
| time.sleep(0.5) | |
| # === 2️⃣ Tavily Search === | |
| client = TavilyClient(TAVILY_API_KEY) | |
| all_results = [] | |
| results_md = "### 🌐 Web Search Results\n" | |
| yield "", expanded_md, results_md, "", "", "🌍 Step 2/3: Searching for relevant data..." | |
| for idx, q in enumerate(expanded, start=1): | |
| try: | |
| print(f"🔎 Searching: {q}") | |
| resp = client.search(query=q, max_results=2, include_raw_content=True) | |
| for r in resp.get("results", []): | |
| title = r.get("title", "(no title)") | |
| content = (r.get("content") or r.get("raw_content") or "")[:300].replace("\n", " ") | |
| url = r.get("url", "") | |
| all_results.append({ | |
| "query": q, | |
| "title": title, | |
| "content": content, | |
| "url": url, | |
| }) | |
| results_md += f"**{title}**\n\n{content}...\n\n[🔗 {url}]({url})\n\n" | |
| yield "", expanded_md, results_md, "", "", f"🌍 Searching... ({idx}/{len(expanded)})" | |
| except Exception as e: | |
| print(f"⚠️ Error: {e}") | |
| print(f"📦 Total results collected: {len(all_results)}") | |
| # === 3️⃣ Summarize via Azure GPT === | |
| client_azure = _get_azure_client() | |
| sources_text = "\n\n".join( | |
| f"[{i+1}] {r['title']} — {r['url']}\n({r['query']})\n{r['content']}" | |
| for i, r in enumerate(all_results) | |
| ) | |
| summary_md = "### ⚽ Final Analysis Report\n" | |
| yield summary_md, expanded_md, results_md, "", "", "🧠 Step 3/3: Summarizing insights..." | |
| messages = [ | |
| { | |
| "role": "system", | |
| "content": ( | |
| "You are a professional football data analyst.\n" | |
| "Summarize all findings into a clean markdown report.\n" | |
| "Include sections:\n" | |
| "1️⃣ Overview & Summary\n" | |
| "2️⃣ Tactics / Strategy\n" | |
| "3️⃣ Key Player & Team Stats\n" | |
| "4️⃣ Injuries / Lineups\n" | |
| "5️⃣ Odds / Market Trends\n" | |
| "6️⃣ Referee & Weather Factors\n" | |
| "7️⃣ Historical / Psychological Context\n" | |
| "8️⃣ Final Verdict / Prediction\n" | |
| "9️⃣ Source List" | |
| ), | |
| }, | |
| {"role": "user", "content": f"Query: {user_query}\n\nSources:\n{sources_text}"}, | |
| ] | |
| try: | |
| response = client_azure.chat.completions.create( | |
| model=AZURE_DEPLOYMENT, | |
| messages=messages, | |
| stream=True, | |
| ) | |
| for chunk in response: | |
| if hasattr(chunk, "choices") and chunk.choices: | |
| delta = chunk.choices[0].delta | |
| if delta and getattr(delta, "content", None): | |
| text = delta.content | |
| summary_md += text | |
| yield summary_md, "", "", "", "", "✅ Generating Final Report..." | |
| print("✅ Summary complete.") | |
| except Exception as e: | |
| yield f"Error: {e}", "", "", "", "", "❌ Summarization failed." | |
| # === Final Tabs Content === | |
| refs_md = "### 🌍 Web References\n\n" + "\n".join( | |
| [f"- [{r['title']}]({r['url']})" for r in all_results] | |
| ) | |
| expand_tab_md = expanded_md | |
| search_tab_md = results_md | |
| yield summary_md, "", "", expand_tab_md + "\n\n---\n\n" + search_tab_md, refs_md, "✅ All steps completed!" | |
| # ---------- Gradio Interface ---------- | |
| if __name__ == "__main__": | |
| with gr.Blocks(theme=gr.themes.Soft(primary_hue="blue", secondary_hue="gray")) as demo: | |
| gr.HTML(""" | |
| <div style='text-align:center; font-size:28px; font-weight:bold;'> | |
| ⚽ Football Intelligence Agent — Pro Dashboard | |
| </div> | |
| <div style='text-align:center; color:gray; font-size:16px; margin-bottom:20px;'> | |
| AI-powered multi-source football analysis in real-time | |
| </div> | |
| """) | |
| status = gr.Markdown("⏸ Waiting for input...", elem_id="status-box") | |
| # === Final result always on top === | |
| summary_box = gr.Markdown("### ⚽ Final Analysis Report will appear here...", elem_id="summary-box") | |
| with gr.Row(): | |
| # -------- Right panel: tabs -------- | |
| with gr.Column(scale=1): | |
| gr.Markdown("## 📚 Research History") | |
| with gr.Tab("🧩 Expanded Queries + Search"): | |
| expand_tab_output = gr.Markdown("No data yet.") | |
| with gr.Tab("🌐 Web References"): | |
| refs_output = gr.Markdown("No references yet.") | |
| # -------- Left panel: input + streaming results -------- | |
| with gr.Column(scale=3): | |
| user_input = gr.Textbox( | |
| label="Enter your football query", | |
| placeholder="e.g. Chelsea vs Sunderland analysis", | |
| ) | |
| run_button = gr.Button("🚀 Run Football Research", variant="primary") | |
| expanded_box = gr.Markdown(label="Expanded Queries", elem_id="expand-box") | |
| results_box = gr.Markdown(label="Search Results", elem_id="results-box") | |
| # Bind button | |
| run_button.click( | |
| fn=stream_research, | |
| inputs=user_input, | |
| outputs=[summary_box, expanded_box, results_box, expand_tab_output, refs_output, status], | |
| queue=True | |
| ) | |
| demo.queue() | |
| demo.launch() | |