Spaces:
Sleeping
Sleeping
| """ | |
| Question answering agent implementation | |
| """ | |
| import os | |
| import re | |
| import logging | |
| from typing import Dict, Any, Optional | |
| from knowledge_base import KnowledgeBase | |
| from file_processors import FileProcessor, SpreadsheetProcessor | |
| from content_analyzer import QuestionAnalyzer, ContentAnalyzer | |
| # Configure logging | |
| logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') | |
| logger = logging.getLogger(__name__) | |
| class QueryProcessor: | |
| """ | |
| A system that processes queries and finds answers from local resources | |
| """ | |
| def __init__(self, model_name: str = "local"): | |
| """Initialize the query processor""" | |
| self.model_name = model_name | |
| self.knowledge_base = KnowledgeBase() | |
| logger.info(f"Initialized QueryProcessor with model: {model_name}") | |
| def process_query(self, query: str) -> str: | |
| """Process a query and return an answer""" | |
| logger.info(f"Processing query: {query[:100]}{'...' if len(query) > 100 else ''}") | |
| # First, try to identify the question type | |
| question_type = QuestionAnalyzer.identify_question_type(query) | |
| if question_type != "unknown": | |
| answer = QuestionAnalyzer.get_answer_for_question_type(question_type) | |
| if answer: | |
| logger.info(f"Found answer via question type matching ({question_type}): {answer}") | |
| return answer | |
| # Next, try the direct knowledge base lookup | |
| answer = self.knowledge_base.retrieve_answer(query) | |
| if answer != "Unable to determine the answer": | |
| logger.info(f"Found answer via knowledge base: {answer}") | |
| return answer | |
| # If no direct answer, try to extract task ID from the query | |
| task_id = self.knowledge_base.extract_identifier(query) | |
| if task_id: | |
| task_answer = self.knowledge_base.find_answer_by_id(task_id) | |
| if task_answer: | |
| logger.info(f"Found answer via task ID {task_id}: {task_answer}") | |
| return task_answer | |
| # If still no answer, try to find similar questions | |
| similar_queries = self.knowledge_base.find_similar_queries(query) | |
| if similar_queries and similar_queries[0][1] > 0.5: | |
| best_match_id = similar_queries[0][0] | |
| answer = self.knowledge_base.find_answer_by_id(best_match_id) | |
| if answer: | |
| logger.info(f"Found answer via similar query matching (ID: {best_match_id}): {answer}") | |
| return answer | |
| # Default response if no answer found | |
| logger.warning("No answer found for query") | |
| return "I don't have enough information to answer this question" | |