Spaces:
Sleeping
Sleeping
| from langchain_core.tools import tool, StructuredTool | |
| from langgraph.types import interrupt, Command | |
| import requests | |
| import warnings | |
| from agent.rag.rag import init_rag, get_relevant_question | |
| warnings.filterwarnings("ignore", category=UserWarning) | |
| text_encoder_model = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2" | |
| price_url = "https://open.er-api.com/v6/latest/EGP" | |
| class GraphTools: | |
| def __init__(self,path_file:str,text_encoder_model:str=text_encoder_model,price_url:str=price_url): | |
| print("Initializing GraphTools...") | |
| self.text_encoder_model = text_encoder_model | |
| self.path_file = path_file | |
| self.price_url = price_url | |
| self.model, self.corpus_embeddings, self.corpus, self.answers = self.init_rag() | |
| ############################## | |
| ## ASK USER Interrupt Tool | |
| ############################## | |
| def ask_user(self, question: str) -> str: | |
| """Ask the user in Arabic one clear clarifying question when more information is needed. | |
| Ask only one question per call and wait for the user's answer before asking | |
| another if necessary. | |
| """ | |
| answer = interrupt({"question": question}) | |
| return answer | |
| ############################## | |
| ## EGP to usd conv. tool | |
| ############################## | |
| def get_egp_to_usd(self, egp_amount:float) -> float : | |
| """Convert an amount from Egyptian Pounds (EGP) to US Dollars (USD) | |
| using the latest available exchange rate. | |
| Use this tool whenever you found prices. | |
| Args: | |
| egp_amount: The amount in Egyptian Pounds (EGP) to convert. | |
| Returns: | |
| The equivalent amount in US Dollars (USD). | |
| Returns: | |
| - Converted USD amount on success. | |
| - None if the exchange-rate service returns an unexpected response. | |
| - -1 if the request fails or another error occurs. | |
| """ | |
| try: | |
| response = requests.get(self.price_url, timeout=8) | |
| response.raise_for_status() # raise error for bad status | |
| data = response.json() | |
| if data.get("result") != "success": | |
| print("API error:", data) | |
| return None | |
| rate = data["rates"]["USD"] | |
| usd_amount = egp_amount * rate | |
| return usd_amount | |
| except Exception as e: | |
| print("Error:", e) | |
| return -1 | |
| #################### | |
| ## RAG | |
| #################### | |
| def init_rag(self): | |
| return init_rag(self.path_file, self.text_encoder_model) | |
| def get_relevant_question(self,query:str) -> str: | |
| """ | |
| Retrieve the most relevant question-answer pair for a user's query | |
| using semantic similarity. | |
| Use this tool when the user asks a question that may already have an | |
| existing answer in the knowledge base. The tool searches semantically | |
| rather than by exact keyword matching. | |
| Args: | |
| query: The user's question or search query. | |
| Returns: | |
| A formatted string containing the most relevant question, its answer, | |
| and the similarity score if a sufficiently similar match is found. | |
| Returns None if no match meets the similarity threshold. | |
| """ | |
| return get_relevant_question(self.model, self.corpus_embeddings, self.corpus, self.answers, query) | |
| def get_tools(self): | |
| ask_user_tool = StructuredTool.from_function( | |
| func=self.ask_user, | |
| name="ask_user", | |
| description=self.ask_user.__doc__, | |
| ) | |
| dollar_tool = StructuredTool.from_function( | |
| func=self.get_egp_to_usd, | |
| name="get_egp_to_usd", | |
| description=self.get_egp_to_usd.__doc__, | |
| ) | |
| rag_tool = StructuredTool.from_function( | |
| func=self.get_relevant_question, | |
| name="get_relevant_question", | |
| description=self.get_relevant_question.__doc__, | |
| ) | |
| return [ask_user_tool,dollar_tool,rag_tool] |