pathakDev10/EstateGuru
0
1import uuid2import threading3import asyncio4import json5import re6import random7import time8import pickle9import numpy as np10import requests # For llama.cpp server calls11from datetime import datetime12from fastapi import FastAPI, WebSocket, WebSocketDisconnect13from langchain_core.messages import AIMessage, HumanMessage, SystemMessage14from langgraph.graph import StateGraph, START, END15import faiss16from sentence_transformers import SentenceTransformer17from tools import extract_json_from_response, apply_filters_partial, rule_based_extract, structured_property_data, estateKeywords, sendTokenViaSocket18from langchain_core.prompts import ChatPromptTemplate19from langchain_core.tools import tool20from langchain_core.callbacks import StreamingStdOutCallbackHandler, CallbackManager21from langchain_core.callbacks.base import BaseCallbackHandler22 23# ------------------------ Model Inference Wrapper ------------------------24 25class ChatQwen:26 """27 A chat wrapper for Qwen using llama.cpp.28 This class can work in two modes:29 - Local: Using a llama-cpp-python binding (gguf model file loaded locally).30 - Server: Calling a remote llama.cpp server endpoint.31 """32 def __init__(33 self,34 temperature=0.3,35 streaming=False,36 max_new_tokens=512,37 callbacks=None,38 use_server=False,39 model_path: str = None,40 server_url: str = None41 ):42 self.temperature = temperature43 self.streaming = streaming44 self.max_new_tokens = max_new_tokens45 self.callbacks = callbacks46 self.use_server = use_server47 48 if self.use_server:49 # Use remote llama.cpp server – provide its URL.50 self.server_url = server_url or "http://localhost:8000"51 else:52 # For local inference, a model_path must be provided.53 if not model_path:54 raise ValueError("Local mode requires a valid model_path to the gguf file.")55 from llama_cpp import Llama # assumes llama-cpp-python is installed56 self.model = Llama(57 model_path=model_path,58 temperature=self.temperature,59 # n_ctx=512,60 n_ctx=2048,61 n_threads=4, # Adjust as needed62 batch_size=512,63 )64 65 def build_prompt(self, messages: list) -> str:66 """Build Qwen-compatible prompt with special tokens."""67 prompt = ""68 for msg in messages:69 role = msg["role"]70 content = msg["content"]71 if role == "system":72 prompt += f"<|im_start|>system\n{content}<|im_end|>\n"73 elif role == "user":74 prompt += f"<|im_start|>user\n{content}<|im_end|>\n"75 elif role == "assistant":76 prompt += f"<|im_start|>assistant\n{content}<|im_end|>\n"77 prompt += "<|im_start|>assistant\n"78 return prompt79 80 def generate_text(self, messages: list) -> str:81 prompt = self.build_prompt(messages)82 stop_tokens = ["<|im_end|>", "\n"] # Qwen's stop sequences83 84 if self.use_server:85 payload = {86 "prompt": prompt,87 "max_tokens": self.max_new_tokens,88 "temperature": self.temperature,89 "stream": self.streaming,90 "stop": stop_tokens # Add stop tokens to server request91 }92 if self.streaming:93 response = requests.post(f"{self.server_url}/generate", json=payload, stream=True)94 generated_text = ""95 for line in response.iter_lines():96 if line:97 token = line.decode("utf-8")98 # Check for stop tokens in stream99 if any(stop in token for stop in stop_tokens):100 break101 generated_text += token102 if self.callbacks:103 for callback in self.callbacks:104 callback.on_llm_new_token(token)105 return generated_text106 else:107 response = requests.post(f"{self.server_url}/generate", json=payload)108 return response.json().get("generated_text", "")109 else:110 # Local llama.cpp inference111 if self.streaming:112 stream = self.model.create_completion(113 prompt=prompt,114 max_tokens=self.max_new_tokens,115 temperature=self.temperature,116 stream=True,117 stop=stop_tokens118 )119 generated_text = ""120 for token_chunk in stream:121 token_text = token_chunk["choices"][0]["text"]122 # Stop early if we detect end token123 if any(stop in token_text for stop in stop_tokens):124 break125 generated_text += token_text126 if self.callbacks:127 for callback in self.callbacks:128 callback.on_llm_new_token(token_text)129 return generated_text130 else:131 result = self.model.create_completion(132 prompt=prompt,133 max_tokens=self.max_new_tokens,134 temperature=self.temperature,135 stop=stop_tokens136 )137 return result["choices"][0]["text"]138 139 def invoke(self, messages: list, config: dict = None) -> AIMessage:140 config = config or {}141 callbacks = config.get("callbacks", self.callbacks)142 original_callbacks = self.callbacks143 self.callbacks = callbacks144 145 output_text = self.generate_text(messages)146 self.callbacks = original_callbacks147 148 # In streaming mode we return an empty content as tokens are being sent via callbacks.149 if self.streaming:150 return AIMessage(content="")151 else:152 return AIMessage(content=output_text)153 154 def __call__(self, messages: list) -> AIMessage:155 return self.invoke(messages)156 157# ------------------------ Callback for WebSocket Streaming ------------------------158 159class WebSocketStreamingCallbackHandler(BaseCallbackHandler):160 def __init__(self, connection_id: str, loop):161 self.connection_id = connection_id162 self.loop = loop163 164 def on_llm_new_token(self, token: str, **kwargs):165 asyncio.run_coroutine_threadsafe(166 manager_socket.send_message(self.connection_id, token),167 self.loop168 )169 170# ------------------------ Instantiate the LLM ------------------------171# Choose one mode: local (set use_server=False) or server (set use_server=True).172model_path="qwen2.5-1.5b-instruct-q4_k_m.gguf"173llm = ChatQwen(174 temperature=0.3,175 streaming=True,176 max_new_tokens=512,177 use_server=False,178 model_path=model_path,179 # server_url="http://localhost:8000" # Uncomment and set if using server mode.180)181 182# ------------------------ FAISS and Sentence Transformer Setup ------------------------183 184index = faiss.read_index("./faiss.index")185with open("./metadata.pkl", "rb") as f:186 docs = pickle.load(f)187st_model = SentenceTransformer('all-MiniLM-L6-v2')188 189def make_system_prompt(suffix: str) -> str:190 return (191 "You are EstateGuru, a real estate expert developed by Abhishek Pathak at SwavishTek. "192 "Your role is to help customers buy properties using only the provided data—do not invent any details. "193 "The default currency is AED; if a query mentions another currency, convert the amount to AED "194 "(for example, convert $10k to 36726.50 AED and $1 to 3.67 AED). "195 "If a customer is interested in a property or needs to contact an agent, instruct them to call +91 8766268285. "196 "Keep your answers short, clear, and concise."197 f"\n{suffix}"198 )199 200general_query_prompt = make_system_prompt(201 "You are EstateGuru, a helpful real estate assistant. "202 "Please respond only in English. "203 "Convert any prices to USD before answering. "204 "Provide a brief, direct answer without extra details."205)206 207# ------------------------ Tool Definitions ------------------------208 209@tool210def extract_filters(query: str) -> dict:211 """Extract filters from the query."""212 llm_local = ChatQwen(temperature=0.3, streaming=False, use_server=False, model_path=model_path)213 system = (214 "You are an expert in extracting filters from property-related queries. Your task is to extract and return only the keys explicitly mentioned in the query as a valid JSON object (starting with '{' and ending with '}'). Include only those keys that are directly present in the query.\n\n"215 "The possible keys are:\n"216 " - 'projectName': The name of the project.\n"217 " - 'developerName': The developer's name.\n"218 " - 'relationshipManager': The relationship manager.\n"219 " - 'propertyAddress': The property address.\n"220 " - 'surroundingArea': The area or nearby landmarks.\n"221 " - 'propertyType': The type or configuration of the property.\n"222 " - 'amenities': Any amenities mentioned.\n"223 " - 'coveredParking': Parking availability.\n"224 " - 'petRules': Pet policies.\n"225 " - 'security': Security details.\n"226 " - 'occupancyRate': Occupancy information.\n"227 " - 'constructionImpact': Construction or its impact.\n"228 " - 'propertySize': Size of the property.\n"229 " - 'propertyView': View details.\n"230 " - 'propertyCondition': Condition of the property.\n"231 " - 'serviceCharges': Service or maintenance charges.\n"232 " - 'ownershipType': Ownership type.\n"233 " - 'totalCosts': A cost threshold or cost amount.\n"234 " - 'paymentPlans': Payment or financing plans.\n"235 " - 'expectedRentalYield': Expected rental yield.\n"236 " - 'rentalHistory': Rental history.\n"237 " - 'shortTermRentals': Short-term rental information.\n"238 " - 'resalePotential': Resale potential.\n"239 " - 'uniqueId': A unique identifier.\n\n"240 "Important instructions regarding cost thresholds:\n"241 " - If the query contains phrases like 'under 10k', 'below 2m', or 'less than 5k', interpret these as cost thresholds.\n"242 " - Convert any shorthand cost values to pure numbers (for example, '10k' becomes 10000, '2m' becomes 2000000) and assign them to the key 'totalCosts'.\n"243 " - Do not use 'propertySize' for cost thresholds.\n\n"244 " - Default currency is AED, if user query have different currency symbol then convert to equivalent AED amount (eg. $10k becomes 36726.50, $1 becomes 3.67).\n\n"245 "Example:\n"246 " For the query: \"properties near dubai mall under 43k\"\n"247 " The expected output should be:\n"248 " { \"surroundingArea\": \"dubai mall\", \"totalCosts\": 43000 }\n\n"249 "Return ONLY a valid JSON object with the extracted keys and their corresponding values, with no additional text."250 )251 252 human_str = f"Here is the query:\n{query}"253 filter_prompt = [254 {"role": "system", "content": system},255 {"role": "user", "content": human_str},256 ]257 response = llm_local.invoke(messages=filter_prompt)258 response_text = response.content if isinstance(response, AIMessage) else str(response)259 try:260 model_filters = extract_json_from_response(response_text)261 except Exception as e:262 print(f"JSON parsing error: {e}")263 model_filters = {}264 rule_filters = rule_based_extract(query)265 print("Rule-based extraction:", rule_filters)266 final_filters = {**model_filters, **rule_filters}267 print("Final extraction:", final_filters)268 return {"filters": final_filters}269 270 271@tool272def determine_route(query: str) -> dict:273 """Determine the route (search, suggest, detail, general, out_of_domain) for the query."""274 real_estate_keywords = estateKeywords275 pattern = re.compile("|".join(re.escape(keyword) for keyword in real_estate_keywords), re.IGNORECASE)276 positive_signal = bool(pattern.search(query))277 278 llm_local = ChatQwen(temperature=0.3, streaming=False, use_server=False, model_path=model_path)279 transform_suggest_to_list = query.lower().replace("suggest ", "list ", -1)280 system = """281 Classify the user query as:282 283 - **"search"**: if it requests property listings with specific filters (e.g., location, price, property type like "2bhk", service charges, pet policies, etc.).284 - **"suggest"**: if it asks for property suggestions without filters.285 - **"detail"**: if it is asking for more information about a previously provided property (for example, "tell me more about property 5" or "I want more information regarding 4BHK").286 - **"general"**: for all other real estate-related questions.287 - **"out_of_domain"**: if the query is not related to real estate (for example, tourist attractions, restaurants, etc.).288 289 Keep in mind that queries mentioning terms like "service charge", "allow pets", "pet rules", etc., are considered real estate queries.290 291 Return only the keyword: search, suggest, detail, general, or out_of_domain.292 """293 human_str = f"Here is the query:\n{transform_suggest_to_list}"294 router_prompt = [295 {"role": "system", "content": system},296 {"role": "user", "content": human_str},297 ]298 299 response = llm_local.invoke(messages=router_prompt)300 response_text = response.content if isinstance(response, AIMessage) else str(response)301 route_value = str(response_text).strip().lower()302 303 # --- NEW: Force 'detail' if query explicitly mentions a specific property (e.g., "property 2") ---304 property_detail_pattern = re.compile(r"property\s+\d+", re.IGNORECASE)305 if property_detail_pattern.search(query):306 route_value = "detail"307 308 # Fallback override if query appears detailed.309 detail_phrases = [310 "more information", "tell me more", "more details", "give me more details",311 "i need more details", "can you provide more details", "additional details",312 "further information", "expand on that", "explain further", "elaborate more",313 "more specifics", "i want to know more", "could you elaborate", "need more info",314 "provide more details", "detail it further", "in-depth information", "break it down further",315 "further explanation", "property 1", "property1", "first property", "about the 2nd", "regarding number 3"316 ]317 if any(phrase in query.lower() for phrase in detail_phrases):318 route_value = "detail"319 320 if route_value not in {"search", "suggest", "detail", "general", "out_of_domain"}:321 route_value = "general"322 if route_value == "out_of_domain" and positive_signal:323 route_value = "general"324 if route_value == "out_of_domain":325 route_value = "general" if positive_signal else "out_of_domain"326 327 return {"route": route_value}328 329# ------------------------ Workflow Setup ------------------------330 331workflow = StateGraph(state_schema=dict)332 333def route_query(state: dict) -> dict:334 new_state = state.copy()335 try:336 new_state["route"] = determine_route.invoke(new_state.get("query", "")).get("route", "general")337 print(new_state["route"])338 except Exception as e:339 print(f"Routing error: {e}")340 new_state["route"] = "general"341 return new_state342 343def hybrid_extract(state: dict) -> dict:344 new_state = state.copy()345 new_state["filters"] = extract_filters.invoke(new_state.get("query", "")).get("filters", {})346 return new_state347 348def search_faiss(state: dict) -> dict:349 new_state = state.copy()350 # Preserve previous properties until new ones are fetched:351 new_state.setdefault("current_properties", state.get("current_properties", []))352 query_embedding = st_model.encode([state["query"]])353 _, indices = index.search(query_embedding.astype(np.float32), 5)354 new_state["faiss_results"] = [docs[idx] for idx in indices[0] if idx < len(docs)]355 return new_state356 357def apply_filters(state: dict) -> dict:358 new_state = state.copy()359 new_state["final_results"] = apply_filters_partial(state["faiss_results"], state.get("filters", {}))360 return new_state361 362def suggest_properties(state: dict) -> dict:363 new_state = state.copy()364 new_state["suggestions"] = random.sample(docs, 5)365 # Explicitly update current_properties only when new listings are fetched366 new_state["current_properties"] = new_state["suggestions"]367 return new_state368 369def handle_out_of_domain(state: dict) -> dict:370 new_state = state.copy()371 new_state["response"] = "I only handle real estate inquiries. Please ask a question related to properties."372 return new_state373 374 375 376def generate_response(state: dict) -> dict:377 new_state = state.copy()378 messages = []379 380 # Add the general query prompt.381 messages.append({"role": "system", "content": general_query_prompt})382 383 # For detail queries (specific property queries), add extra instructions.384 if new_state.get("route", "general") == "detail":385 messages.append({386 "role": "system",387 "content": (388 "The user is asking about a specific property from the numbered list below. "389 "Properties are listed as 1, 2, 3, etc. Use ONLY the corresponding property details. "390 "For example, if the user says 'property 2', respond using only the details from the second entry. Never invent data."391 )392 })393 394 if new_state.get("current_properties"):395 # Format properties with indices starting at 1396 property_context = format_property_data_with_indices(new_state["current_properties"])397 messages.append({"role": "system", "content": "Available Properties:\n" + property_context})398 messages.append({"role": "system", "content": "When responding, use only the provided property details."})399 400 # for msg in state.get("messages", []): // todo: remove previous listing data and keep only last401 # if(msg["role"] == "system" and msg["content"].in)402 403 404 # Add conversation history405 # Truncate conversation history (last 2 exchanges)406 truncated_history = state.get("messages", [])[-4:] # Last 2 user+assistant pairs407 for msg in truncated_history:408 messages.append({"role": msg["role"], "content": msg["content"]})409 410 connection_id = state.get("connection_id")411 loop = state.get("loop")412 if connection_id and loop:413 print("Using WebSocket streaming")414 callback_manager = [WebSocketStreamingCallbackHandler(connection_id, loop)]415 _ = llm.invoke(416 messages,417 config={"callbacks": callback_manager}418 )419 new_state["response"] = ""420 else:421 callback_manager = [StreamingStdOutCallbackHandler()]422 response = llm.invoke(423 messages,424 config={"callbacks": callback_manager}425 )426 new_state["response"] = response.content if isinstance(response, AIMessage) else str(response)427 428 return new_state429 430 431def format_property_data_with_indices(properties: list) -> str:432 formatted = []433 for idx, prop in enumerate(properties, 1):434 cost = prop.get("totalCosts", "N/A")435 cost_str = f"{cost:,}" if isinstance(cost, (int, float)) else cost436 formatted.append(437 f"{idx}. Type: {prop['propertyType']}, Cost: AED {cost_str}, "438 f"Size: {prop.get('propertySize', 'N/A')}, Amenities: {', '.join(prop.get('amenities', []))}, "439 f"Rental Yield: {prop.get('expectedRentalYield', 'N/A')}, "440 f"Ownership: {prop.get('ownershipType', 'N/A')}"441 )442 return "\n".join(formatted)443 444 445def format_final_response(state: dict) -> dict:446 new_state = state.copy()447 448 if state.get("route") in ["search", "suggest"]:449 if "final_results" in state:450 new_state["current_properties"] = state["final_results"]451 elif "suggestions" in state:452 new_state["current_properties"] = state["suggestions"]453 elif "current_properties" in new_state:454 new_state["current_properties"] = state["current_properties"]455 456 457 # print("state: ", json.dumps(new_state), "\n\n")458 # Format the property details if available.459 # if new_state.get("current_properties"):460 if state.get("route") in ["search", "suggest"] and new_state.get("current_properties"):461 formatted = structured_property_data(state=new_state)462 463 # for idx, prop in enumerate(new_state["current_properties"], 1):464 # cost = prop.get("totalCosts", "N/A")465 # cost_str = f"{cost:,}" if isinstance(cost, (int, float)) else cost466 # formatted.append(467 # f"{idx}. Type: {prop['propertyType']}, Cost: AED {cost_str}, "468 # f"Size: {prop.get('propertySize', 'N/A')}, Amenities: {', '.join(map(str, prop.get('amenities', []))) if prop.get('amenities') else 'N/A'}, "469 # f"Rental Yield: {prop.get('expectedRentalYield', 'N/A')}, "470 # f"Ownership: {prop.get('ownershipType', 'N/A')}\n"471 # )472 aggregated_response = "Here are the property details:\n" + "\n".join(formatted)473 # print(aggregated_response)474 475 connection_id = state.get("connection_id")476 loop = state.get("loop")477 if connection_id and loop:478 import time479 tokens = aggregated_response.split(" ")480 for token in tokens:481 asyncio.run_coroutine_threadsafe(482 manager_socket.send_message(connection_id, token + " "),483 loop484 )485 time.sleep(0.05)486 new_state["response"] = ""487 else:488 new_state["response"] = aggregated_response489 elif "response" in new_state:490 connection_id = state.get("connection_id")491 loop = state.get("loop")492 if connection_id and loop:493 import time494 tokens = str(new_state["response"]).split(" ")495 for token in tokens:496 asyncio.run_coroutine_threadsafe(497 manager_socket.send_message(connection_id, token + " "),498 loop499 )500 time.sleep(0.05)501 new_state["response"] = str(new_state["response"])502 503 return new_state504 505 506 507nodes = [508 ("route_query", route_query),509 ("hybrid_extract", hybrid_extract),510 ("faiss_search", search_faiss),511 ("apply_filters", apply_filters),512 ("suggest_properties", suggest_properties),513 ("handle_out_of_domain", handle_out_of_domain),514 ("generate_response", generate_response),515 ("format_response", format_final_response)516]517 518for name, node in nodes:519 workflow.add_node(name, node)520 521workflow.add_edge(START, "route_query")522workflow.add_conditional_edges(523 "route_query",524 lambda state: state.get("route", "general"),525 {526 "search": "hybrid_extract", 527 "suggest": "suggest_properties", 528 "detail": "generate_response", 529 "general": "generate_response", 530 "out_of_domain": "handle_out_of_domain"531 }532)533workflow.add_edge("hybrid_extract", "faiss_search")534workflow.add_edge("faiss_search", "apply_filters")535workflow.add_edge("apply_filters", "format_response")536workflow.add_edge("suggest_properties", "format_response")537workflow.add_edge("generate_response", "format_response")538workflow.add_edge("handle_out_of_domain", "format_response")539workflow.add_edge("format_response", END)540 541workflow_app = workflow.compile()542 543# ------------------------ Conversation Manager ------------------------544 545class ConversationManager:546 def __init__(self):547 # Each connection gets its own conversation history and state.548 self.conversation_history = []549 # current_properties stores the current property listing.550 self.current_properties = []551 552 def _add_message(self, role: str, content: str):553 self.conversation_history.append({554 "role": role,555 "content": content,556 "timestamp": datetime.now().isoformat()557 })558 559 def process_query(self, query: str) -> str:560 # For greeting messages, reset history/state. // post request561 if query.strip().lower() in {"hi", "hello", "hey"}:562 self.conversation_history = []563 self.current_properties = []564 greeting_response = "Hello! How can I assist you today with your real estate inquiries?"565 self._add_message("assistant", greeting_response)566 return greeting_response567 568 try:569 self._add_message("user", query)570 initial_state = {571 "messages": self.conversation_history.copy(),572 "query": query,573 "route": "general",574 "filters": {},575 "current_properties": self.current_properties576 }577 for event in workflow_app.stream(initial_state, stream_mode="values"):578 final_state = event579 # Only update property listings if a new listing is fetched580 # if 'final_results' in final_state:581 # self.current_properties = final_state['final_results']582 # elif 'suggestions' in final_state:583 # self.current_properties = final_state['suggestions']584 self.current_properties = final_state.get("current_properties", [])585 586 if final_state.get("route") == "general":587 response_text = final_state.get("response", "")588 self._add_message("assistant", response_text)589 return response_text590 else:591 response = final_state.get("response", "I couldn't process that request.")592 self._add_message("assistant", response)593 return response594 except Exception as e:595 print(f"Processing error: {e}")596 return "Sorry, I encountered an error processing your request."597 598 599 600conversation_managers = {}601 602# ------------------------ FastAPI Backend with WebSockets ------------------------603 604app = FastAPI()605 606class ConnectionManager:607 def __init__(self):608 self.active_connections = {}609 610 async def connect(self, websocket: WebSocket):611 await websocket.accept()612 connection_id = str(uuid.uuid4())613 self.active_connections[connection_id] = websocket614 print(f"New connection: {connection_id}")615 return connection_id616 617 def disconnect(self, connection_id: str):618 if connection_id in self.active_connections:619 del self.active_connections[connection_id]620 print(f"Disconnected: {connection_id}")621 622 async def send_message(self, connection_id: str, message: str):623 websocket = self.active_connections.get(connection_id)624 if websocket:625 await websocket.send_text(message)626 627manager_socket = ConnectionManager()628 629def stream_query(query: str, connection_id: str, loop):630 conv_manager = conversation_managers.get(connection_id)631 if conv_manager is None:632 print(f"No conversation manager found for connection {connection_id}")633 return634 635 if query.strip().lower() in {"hi", "hello", "hey"}:636 conv_manager.conversation_history = []637 conv_manager.current_properties = []638 greeting_response = "Hello! How can I assist you today with your real estate inquiries?"639 conv_manager._add_message("assistant", greeting_response)640 sendTokenViaSocket(641 state={"connection_id": connection_id, "loop": loop},642 manager_socket=manager_socket,643 message=greeting_response644 )645 # asyncio.run_coroutine_threadsafe(646 # manager_socket.send_message(connection_id, greeting_response),647 # loop648 # )649 return650 651 conv_manager._add_message("user", query)652 initial_state = {653 "messages": conv_manager.conversation_history.copy(),654 "query": query,655 "route": "general",656 "filters": {},657 "current_properties": conv_manager.current_properties,658 "connection_id": connection_id,659 "loop": loop660 }661 # try:662 # workflow_app.invoke(initial_state)663 # except Exception as e:664 # error_msg = f"Error processing query: {str(e)}"665 # asyncio.run_coroutine_threadsafe(666 # manager_socket.send_message(connection_id, error_msg),667 # loop668 # )669 try:670 # Capture all states during execution671 # final_state = None672 # for event in workflow_app.stream(initial_state, stream_mode="values"):673 # final_state = event674 675 # # Update conversation manager with final state676 # if final_state:677 # conv_manager.current_properties = final_state.get("current_properties", [])678 # if final_state.get("response"):679 # conv_manager._add_message("assistant", final_state["response"])680 final_state = None681 for event in workflow_app.stream(initial_state, stream_mode="values"):682 final_state = event683 684 if final_state:685 # Always update current_properties from final state686 conv_manager.current_properties = final_state.get("current_properties", [])687 # Keep conversation history bounded688 conv_manager.conversation_history = conv_manager.conversation_history[-6:] # Last 3 exchanges689 690 except Exception as e:691 error_msg = f"Error processing query: {str(e)}"692 asyncio.run_coroutine_threadsafe(693 manager_socket.send_message(connection_id, error_msg),694 loop695 )696 697 698 699@app.websocket("/ws")700async def websocket_endpoint(websocket: WebSocket):701 connection_id = await manager_socket.connect(websocket)702 # Each connection maintains its own conversation manager.703 conversation_managers[connection_id] = ConversationManager()704 try:705 while True:706 query = await websocket.receive_text()707 loop = asyncio.get_event_loop()708 threading.Thread(709 target=stream_query,710 args=(query, connection_id, loop),711 daemon=True712 ).start()713 except WebSocketDisconnect:714 conv_manager = conversation_managers.get(connection_id)715 if conv_manager:716 filename = f"conversations/conversation_{connection_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"717 with open(filename, "w") as f:718 json.dump(conv_manager.conversation_history, f, indent=4)719 del conversation_managers[connection_id]720 manager_socket.disconnect(connection_id)721 722@app.post("/query")723async def post_query(query: str):724 conv_manager = ConversationManager()725 response = conv_manager.process_query(query)726 return {"response": response}727