from langchain.tools import tool from langchain.prompts import PromptTemplate from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode, tools_condition,InjectedState from langchain_core.messages import ( SystemMessage, HumanMessage, AIMessage, ToolMessage, ) from langgraph.types import Command, interrupt from langgraph.checkpoint.memory import MemorySaver from langchain_core.tools.base import InjectedToolCallId #structuring import ast from dataclasses import dataclass from typing_extensions import TypedDict from typing import Annotated, Literal #getting current location import geocoder import os import requests import json from dotenv import load_dotenv from os import listdir from os.path import isfile, join load_dotenv() GOOGLE_API_KEY=os.getenv('google_api_key') class State(TypedDict): """ A dictionnary representing the state of the agent. """ messages: Annotated[list, add_messages] #location data latitude: str longitude: str address: str #results from place search places: dict def get_current_location_node(state: State): current_location = geocoder.ip("me") if current_location.latlng: latitude, longitude = current_location.latlng address = current_location.address return {'latitude':latitude, 'longitude':longitude, 'address':address} else: return None @tool def get_current_location_tool(tool_call_id: Annotated[str, InjectedToolCallId]): """ Tool to get the current location of the user. agrs: none """ current_location = geocoder.ip("me") if current_location.latlng: latitude, longitude = current_location.latlng address = current_location.address return Command(update={'messages':[ToolMessage(F'The current location is: address:{address}, longitude:{longitude},lattitude:{latitude}', tool_call_id=tool_call_id)], 'latitude':latitude, 'longitude':longitude, 'address':address}) else: return None @tool def find_places_near_me(query:str,state: Annotated[dict, InjectedState],tool_call_id: Annotated[str, InjectedToolCallId]): """ Use this tool to find locations near me. args: query - has to be one of the following "car_dealer", "car_rental", "car_repair", "car_wash", "electric_vehicle_charging_station", "gas_station", "parking", "rest_stop", "corporate_office", "farm", "ranch", "art_gallery", "art_studio", "auditorium", "cultural_landmark", "historical_place", "monument", "museum", "performing_arts_theater", "sculpture", "library", "preschool", "primary_school", "school", "secondary_school", "university", "adventure_sports_center", "amphitheatre", "amusement_center", "amusement_park", "aquarium", "banquet_hall", "barbecue_area", "botanical_garden", "bowling_alley", "casino", "childrens_camp", "comedy_club", "community_center", "concert_hall", "convention_center", "cultural_center", "cycling_park", "dance_hall", "dog_park", "event_venue", "ferris_wheel", "garden", "hiking_area", "historical_landmark", "internet_cafe", "karaoke", "marina", "movie_rental", "movie_theater", "national_park", "night_club", "observation_deck", "off_roading_area", "opera_house", "park", "philharmonic_hall", "picnic_ground", "planetarium", "plaza", "roller_coaster", "skateboard_park", "state_park", "tourist_attraction", "video_arcade", "visitor_center", "water_park", "wedding_venue", "wildlife_park", "wildlife_refuge", "zoo", "public_bath", "public_bathroom", "stable", "accounting", "atm", "bank", "acai_shop", "afghani_restaurant", "african_restaurant", "american_restaurant", "asian_restaurant", "bagel_shop", "bakery", "bar", "bar_and_grill", "barbecue_restaurant", "brazilian_restaurant", "breakfast_restaurant", "brunch_restaurant", "buffet_restaurant", "cafe", "cafeteria", "candy_store", "cat_cafe", "chinese_restaurant", "chocolate_factory", "chocolate_shop", "coffee_shop", "confectionery", "deli", "dessert_restaurant", "dessert_shop", "diner", "dog_cafe", "donut_shop", "fast_food_restaurant", "fine_dining_restaurant", "food_court", "french_restaurant", "greek_restaurant", "hamburger_restaurant", "ice_cream_shop", "indian_restaurant", "indonesian_restaurant", "italian_restaurant", "japanese_restaurant", "juice_shop", "korean_restaurant", "lebanese_restaurant", "meal_delivery", "meal_takeaway", "mediterranean_restaurant", "mexican_restaurant", "middle_eastern_restaurant", "pizza_restaurant", "pub", "ramen_restaurant", "restaurant", "sandwich_shop", "seafood_restaurant", "spanish_restaurant", "steak_house", "sushi_restaurant", "tea_house", "thai_restaurant", "turkish_restaurant", "vegan_restaurant", "vegetarian_restaurant", "vietnamese_restaurant", "wine_bar", "administrative_area_level_1", "administrative_area_level_2", "country", "locality", "postal_code", "school_district", "city_hall", "courthouse", "embassy", "fire_station", "government_office", "local_government_office", "neighborhood_police_station", "police", "post_office", "chiropractor", "dental_clinic", "dentist", "doctor", "drugstore", "hospital", "massage", "medical_lab", "pharmacy", "physiotherapist", "sauna", "skin_care_clinic", "spa", "tanning_studio", "wellness_center", "yoga_studio", "apartment_building", "apartment_complex", "condominium_complex", "housing_complex", "bed_and_breakfast", "budget_japanese_inn", "campground", "camping_cabin", "cottage", "extended_stay_hotel", "farmstay", "guest_house", "hostel", "hotel", "inn", "japanese_inn", "mobile_home_park", "motel", "private_guest_room", "resort_hotel", "rv_park", "beach", "church", "hindu_temple", "mosque", "synagogue", "astrologer", "barber_shop", "beautician", "beauty_salon", "body_art_service", "catering_service", "cemetery", "child_care_agency", "consultant", "courier_service", "electrician", "florist", "food_delivery", "foot_care", "funeral_home", "hair_care", "hair_salon", "insurance_agency", "laundry", "lawyer", "locksmith", "makeup_artist", "moving_company", "nail_salon", "painter", "plumber", "psychic", "real_estate_agency", "roofing_contractor", "storage", "summer_camp_organizer", "tailor", "telecommunications_service_provider", "tour_agency", "tourist_information_center", "travel_agency", "veterinary_care", "asian_grocery_store", "auto_parts_store", "bicycle_store", "book_store", "butcher_shop", "cell_phone_store", "clothing_store", "convenience_store", "department_store", "discount_store", "electronics_store", "food_store", "furniture_store", "gift_shop", "grocery_store", "hardware_store", "home_goods_store", "home_improvement_store", "jewelry_store", "liquor_store", "market", "pet_store", "shoe_store", "shopping_mall", "sporting_goods_store", "store", "supermarket", "warehouse_store", "wholesaler", "arena", "athletic_field", "fishing_charter", "fishing_pond", "fitness_center", "golf_course", "gym", "ice_skating_rink", "playground", "ski_resort", "sports_activity_location", "sports_club", "sports_coaching", "sports_complex", "stadium", "swimming_pool", "airport", "airstrip", "bus_station", "bus_stop", "ferry_terminal", "heliport", "international_airport", "light_rail_station", "park_and_ride", "subway_station", "taxi_stand", "train_station", "transit_depot", "transit_station", "truck_stop" """ try: my_longitude=state['longitude'] my_latitude=state['latitude'] response=requests.get(f'https://maps.googleapis.com/maps/api/place/nearbysearch/json?location={my_latitude}%2C{my_longitude}&radius=500&type={query}&key={GOOGLE_API_KEY}') data=response.json() places={} for place in data['results']: try: name=place['name'] rating=place['rating'] id=place['place_id'] response=requests.get(f'https://places.googleapis.com/v1/places/{id}?fields=googleMapsLinks.placeUri&key={GOOGLE_API_KEY}') data=response.json() link=data['googleMapsLinks']['placeUri'] places[name]= {'rating':rating, 'google_maps_link':link, } except Exception as e: f'Error: {e}' return Command(update={'places':places, 'messages':[ToolMessage(f'I found {len(places)} places', tool_call_id=tool_call_id)]}) except: return Command(update={'messages':[ToolMessage('Could not find places based on the query', tool_call_id=tool_call_id)]}) @tool def look_for_places(query: str, tool_call_id: Annotated[str, InjectedToolCallId]): """ Tool to look for places based on the user query and location. Use this tool for more complex user queries like sentences, and if the location is specified in the query. Places includes restaurants, bars, speakeasy, games, anything. args: query - the query has to be in this format eg.Spicy%20Vegetarian%20Food%20in%20Sydney%20Australia. """ try: response=requests.get(f'https://maps.googleapis.com/maps/api/place/textsearch/json?query={query}?&key={GOOGLE_API_KEY}') data=response.json() places={} for place in data['results']: try: name=place['name'] rating=place['rating'] id=place['place_id'] price_level=place['price_level'] address=place['formatted_address'] lattitude=place['geometry']['location']['lat'] longitude=place['geometry']['location']['lng'] response=requests.get(f'https://places.googleapis.com/v1/places/{id}?fields=googleMapsLinks.placeUri&key={GOOGLE_API_KEY}') data=response.json() link=data['googleMapsLinks']['placeUri'] places[name]= {'address': address, 'rating':rating, 'Price_level':price_level, 'google_maps_link':link, 'longitude':longitude, 'latitude':lattitude} except Exception as e: f'Error: {e}' return Command(update={'places':places, 'messages':[ToolMessage(f'I found {len(places)} places', tool_call_id=tool_call_id)]}) except Exception as e: return f'Error: error' @tool def show_places_found(state: Annotated[dict, InjectedState]): """ Tool to get the places found by previous tool calls and to show/display them. It has links within that can also be used for directions always show the links args: none """ return state['places'] class maps_agent: def __init__(self,llm: any): self.agent=self._setup(llm) def _setup(self,llm): langgraph_tools=[get_current_location_tool,look_for_places, find_places_near_me,show_places_found] graph_builder = StateGraph(State) # Modification: tell the LLM which tools it can call llm_with_tools = llm.bind_tools(langgraph_tools) tool_node = ToolNode(tools=langgraph_tools) def chatbot(state: State): """ travel assistant that answers user questions about their trip. Depending on the request, leverage which tools to use if necessary.""" return {"messages": [llm_with_tools.invoke(state['messages'])]} graph_builder.add_node("chatbot", chatbot) graph_builder.add_node('current_location',get_current_location_node) graph_builder.add_node("tools", tool_node) # Any time a tool is called, we return to the chatbot to decide the next step graph_builder.set_entry_point("current_location") graph_builder.add_edge('current_location','chatbot') graph_builder.add_edge("tools", "chatbot") graph_builder.add_conditional_edges( "chatbot", tools_condition, ) memory=MemorySaver() graph=graph_builder.compile(checkpointer=memory) return graph def get_state(self, state_val:str): config = {"configurable": {"thread_id": "1"}} return self.agent.get_state(config).values[state_val] def stream(self,input:str): config = {"configurable": {"thread_id": "1"}} input_message = HumanMessage(content=input) for event in self.agent.stream({"messages": [input_message]}, config, stream_mode="values"): event["messages"][-1].pretty_print() def chat(self,input:str): config = {"configurable": {"thread_id": "1"}} response=self.agent.invoke({'messages':HumanMessage(content=str(input))},config) return response['messages'][-1].content