Spaces:
Sleeping
Sleeping
File size: 4,314 Bytes
effa16c | 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 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 | 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] |