Spaces:
Sleeping
Sleeping
| import os | |
| import gradio as gr | |
| import requests | |
| import inspect | |
| import time | |
| import pandas as pd | |
| from smolagents import DuckDuckGoSearchTool | |
| import threading | |
| from typing import Dict, List, Optional, Tuple | |
| import json | |
| from huggingface_hub import InferenceClient | |
| # --- Constants --- | |
| DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" | |
| # --- Global Cache for Answers --- | |
| cached_answers = {} | |
| cached_questions = [] | |
| processing_status = {"is_processing": False, "progress": 0, "total": 0} | |
| # --- Intelligent Agent with Conditional Search --- | |
| class IntelligentAgent: | |
| def __init__(self, debug: bool = False, model_name: str = "meta-llama/Llama-3.1-8B-Instruct"): | |
| self.search = DuckDuckGoSearchTool() | |
| self.client = InferenceClient(model=model_name) | |
| self.debug = debug | |
| if self.debug: | |
| print(f"IntelligentAgent initialized with model: {model_name}") | |
| def _should_search(self, question: str) -> bool: | |
| """ | |
| Use LLM to determine if search is needed for the question. | |
| Returns True if search is recommended, False otherwise. | |
| """ | |
| decision_prompt = f"""You are an AI assistant that decides whether a web search is needed to answer questions accurately. | |
| Analyze this question and decide if it requires real-time information, recent data, or specific facts that might not be in your training data. | |
| SEARCH IS NEEDED for: | |
| - Current events, news, recent developments | |
| - Real-time data (weather, stock prices, sports scores) | |
| - Specific factual information that changes frequently | |
| - Recent product releases, company information | |
| - Current status of people, organizations, or projects | |
| - Location-specific current information | |
| SEARCH IS NOT NEEDED for: | |
| - General knowledge questions | |
| - Mathematical calculations | |
| - Programming concepts and syntax | |
| - Historical facts (older than 1 year) | |
| - Definitions of well-established concepts | |
| - How-to instructions for common tasks | |
| - Creative writing or opinion-based responses | |
| Question: "{question}" | |
| Respond with only "SEARCH" or "NO_SEARCH" followed by a brief reason (max 20 words). | |
| Example responses: | |
| - "SEARCH - Current weather data needed" | |
| - "NO_SEARCH - Mathematical concept, general knowledge sufficient" | |
| """ | |
| try: | |
| response = self.client.text_generation( | |
| decision_prompt, | |
| max_new_tokens=50, | |
| temperature=0.1, | |
| do_sample=False | |
| ) | |
| decision = response.strip().upper() | |
| should_search = decision.startswith("SEARCH") | |
| if self.debug: | |
| print(f"Decision for '{question}': {decision}") | |
| return should_search | |
| except Exception as e: | |
| if self.debug: | |
| print(f"Error in search decision: {e}, defaulting to search") | |
| # Default to search if decision fails | |
| return True | |
| def _answer_with_llm(self, question: str) -> str: | |
| """ | |
| Generate answer using LLM without search. | |
| """ | |
| answer_prompt = f"""You are a helpful AI assistant. Answer the following question based on your knowledge. Be accurate, concise, and helpful. If you're not certain about something, acknowledge the uncertainty. | |
| Question: {question} | |
| Answer:""" | |
| try: | |
| response = self.client.text_generation( | |
| answer_prompt, | |
| max_new_tokens=500, | |
| temperature=0.3, | |
| do_sample=True | |
| ) | |
| return response.strip() | |
| except Exception as e: | |
| return f"Sorry, I encountered an error generating the response: {e}" | |
| def _answer_with_search(self, question: str) -> str: | |
| """ | |
| Generate answer using search results and LLM. | |
| """ | |
| try: | |
| # Perform search | |
| search_results = self.search(question) | |
| if not search_results: | |
| return "No search results found. Let me try to answer based on my knowledge:\n\n" + self._answer_with_llm(question) | |
| # Format search results | |
| formatted_results = [] | |
| for i, result in enumerate(search_results[:3]): # Use top 3 results | |
| title = result.get("title", "No title") | |
| snippet = result.get("snippet", "").strip() | |
| link = result.get("link", "") | |
| formatted_results.append(f"Result {i+1}:\nTitle: {title}\nContent: {snippet}\nSource: {link}") | |
| search_context = "\n\n".join(formatted_results) | |
| # Generate answer using search context | |
| answer_prompt = f"""You are a helpful AI assistant. Use the provided search results to answer the question accurately. Synthesize information from multiple sources when relevant, and cite sources when appropriate. | |
| Question: {question} | |
| Search Results: | |
| {search_context} | |
| Based on the search results above, provide a comprehensive answer to the question. If the search results don't fully answer the question, you can supplement with your general knowledge but clearly indicate what comes from the search results vs. your knowledge. | |
| Answer:""" | |
| try: | |
| response = self.client.text_generation( | |
| answer_prompt, | |
| max_new_tokens=600, | |
| temperature=0.3, | |
| do_sample=True | |
| ) | |
| return response.strip() | |
| except Exception as e: | |
| # Fallback to simple search result formatting | |
| top_result = search_results[0] | |
| title = top_result.get("title", "No title") | |
| snippet = top_result.get("snippet", "").strip() | |
| link = top_result.get("link", "") | |
| return f"**{title}**\n\n{snippet}\n\nSource: {link}" | |
| except Exception as e: | |
| return f"Search failed: {e}. Let me try to answer based on my knowledge:\n\n" + self._answer_with_llm(question) | |
| def __call__(self, question: str) -> str: | |
| """ | |
| Main entry point - decide whether to search and generate appropriate response. | |
| """ | |
| if self.debug: | |
| print(f"Agent received question: {question}") | |
| # Early validation | |
| if not question or not question.strip(): | |
| return "Please provide a valid question." | |
| try: | |
| # Decide whether to search | |
| if self._should_search(question): | |
| if self.debug: | |
| print("Using search-based approach") | |
| answer = self._answer_with_search(question) | |
| else: | |
| if self.debug: | |
| print("Using LLM-only approach") | |
| answer = self._answer_with_llm(question) | |
| except Exception as e: | |
| answer = f"Sorry, I encountered an error: {e}" | |
| if self.debug: | |
| print(f"Agent returning answer: {answer[:100]}...") | |
| return answer | |
| def fetch_questions() -> Tuple[str, Optional[pd.DataFrame]]: | |
| """ | |
| Fetch questions from the API and cache them. | |
| """ | |
| global cached_questions | |
| api_url = DEFAULT_API_URL | |
| questions_url = f"{api_url}/questions" | |
| print(f"Fetching questions from: {questions_url}") | |
| try: | |
| response = requests.get(questions_url, timeout=15) | |
| response.raise_for_status() | |
| questions_data = response.json() | |
| if not questions_data: | |
| return "Fetched questions list is empty.", None | |
| cached_questions = questions_data | |
| # Create DataFrame for display | |
| display_data = [] | |
| for item in questions_data: | |
| display_data.append({ | |
| "Task ID": item.get("task_id", "Unknown"), | |
| "Question": item.get("question", "") | |
| }) | |
| df = pd.DataFrame(display_data) | |
| status_msg = f"Successfully fetched {len(questions_data)} questions. Ready to generate answers." | |
| return status_msg, df | |
| except requests.exceptions.RequestException as e: | |
| return f"Error fetching questions: {e}", None | |
| except Exception as e: | |
| return f"An unexpected error occurred: {e}", None | |
| def generate_answers_async(model_name: str = "meta-llama/Llama-3.1-8B-Instruct", progress_callback=None): | |
| """ | |
| Generate answers for all cached questions asynchronously using the intelligent agent. | |
| """ | |
| global cached_answers, processing_status | |
| if not cached_questions: | |
| return "No questions available. Please fetch questions first." | |
| processing_status["is_processing"] = True | |
| processing_status["progress"] = 0 | |
| processing_status["total"] = len(cached_questions) | |
| try: | |
| agent = IntelligentAgent(debug=True, model_name=model_name) | |
| cached_answers = {} | |
| for i, item in enumerate(cached_questions): | |
| if not processing_status["is_processing"]: # Check if cancelled | |
| break | |
| task_id = item.get("task_id") | |
| question_text = item.get("question") | |
| if not task_id or question_text is None: | |
| continue | |
| try: | |
| answer = agent(question_text) | |
| cached_answers[task_id] = { | |
| "question": question_text, | |
| "answer": answer | |
| } | |
| except Exception as e: | |
| cached_answers[task_id] = { | |
| "question": question_text, | |
| "answer": f"AGENT ERROR: {e}" | |
| } | |
| processing_status["progress"] = i + 1 | |
| if progress_callback: | |
| progress_callback(i + 1, len(cached_questions)) | |
| except Exception as e: | |
| print(f"Error in generate_answers_async: {e}") | |
| finally: | |
| processing_status["is_processing"] = False | |
| def start_answer_generation(model_choice: str): | |
| """ | |
| Start the answer generation process in a separate thread. | |
| """ | |
| if processing_status["is_processing"]: | |
| return "Answer generation is already in progress.", None | |
| if not cached_questions: | |
| return "No questions available. Please fetch questions first.", None | |
| # Map model choice to actual model name | |
| model_map = { | |
| "Llama 3.1 8B": "meta-llama/Llama-3.1-8B-Instruct", | |
| "Llama 3.1 70B": "meta-llama/Llama-3.1-70B-Instruct", | |
| "Mistral 7B": "mistralai/Mistral-7B-Instruct-v0.3", | |
| "CodeLlama 7B": "codellama/CodeLlama-7b-Instruct-hf" | |
| } | |
| selected_model = model_map.get(model_choice, "meta-llama/Llama-3.1-8B-Instruct") | |
| # Start generation in background thread | |
| thread = threading.Thread(target=generate_answers_async, args=(selected_model,)) | |
| thread.daemon = True | |
| thread.start() | |
| return f"Answer generation started using {model_choice}. Check progress below.", None | |
| def get_generation_progre |