Spaces:
Sleeping
Sleeping
File size: 2,731 Bytes
922f271 |
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 |
"""
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"
|