Spaces:
Running
Running
File size: 3,997 Bytes
75e2b6c |
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 |
from datetime import datetime, timezone
from typing import Optional, Dict, Any
from yake import KeywordExtractor
from app.services.chathistory import ChatSession
from app.services.llm_model import Model
from app.services.environmental_condition import EnvironmentalData
from app.services.prompts import *
from app.services.MagicConvert import MagicConvert
class Report:
def __init__(self, token: str, session_id: Optional[str] = None):
self.token = token
self.session_id = session_id
self.chat_session = ChatSession(token, session_id)
self.user_city = self.chat_session.get_city()
city = self.user_city if self.user_city else ''
self.environment_data = EnvironmentalData(city)
self.markitdown = MagicConvert()
def extract_keywords_yake(self, text: str, max_ngram_size: int = 2, num_keywords: int = 4) -> list:
kw_extractor = KeywordExtractor(
lan="en",
n=max_ngram_size,
top=num_keywords,
features=None
)
keywords = kw_extractor.extract_keywords(text)
return [kw[0] for kw in keywords]
def ensure_valid_session(self, title: str = None) -> str:
if not self.session_id or not self.session_id.strip():
self.chat_session.create_new_session(title=title)
self.session_id = self.chat_session.session_id
else:
try:
if not self.chat_session.validate_session(self.session_id, title=title):
self.chat_session.create_new_session(title=title)
self.session_id = self.chat_session.session_id
except ValueError:
self.chat_session.create_new_session(title=title)
self.session_id = self.chat_session.session_id
return self.session_id
def process_chat(self, query: str, report_file: str, file_type: Optional[str] = None) -> Dict[str, Any]:
try:
profile = self.chat_session.get_name_and_age()
self.chat_session.update_title(self.session_id, query)
self.session_id = self.ensure_valid_session(title=query)
language = self.chat_session.get_language().lower()
language_prompt = LANGUAGE_RESPONSE_PROMPT.format(language=language)
if not report_file or not file_type:
return {
"error": "Report file or file type missing",
"query": query,
"response": "Sorry, report file or file type is missing.",
"timestamp": datetime.now(timezone.utc).isoformat()
}
report_file_name = report_file + " (File Uploaded)"
conversion_result = self.markitdown.magic(report_file)
report_text = conversion_result.text_content
prompt = MEDICAL_REPORT_ANALYSIS_PROMPT.format(
report=report_text,
current_query=query
)
response = Model().response = Model().llm(prompt + "\n" + language_prompt , query)
keywords = self.extract_keywords_yake(response)
chat_data = {
"query": report_file_name + "\n" +query,
"response": response,
"references": "",
"page_no": "",
"keywords": keywords,
"images": "",
"context": report_text,
"timestamp": datetime.now(timezone.utc).isoformat(),
"session_id": self.chat_session.session_id
}
if not self.chat_session.save_chat(chat_data):
raise ValueError("Failed to save chat message")
return chat_data
except Exception as e:
return {
"error": str(e),
"query": query,
"response": "Sorry, there was an error processing your request.",
"timestamp": datetime.now(timezone.utc).isoformat()
} |