chore: Update vehicle destination in calculate_route function

kitt.py

@@ -4,19 +4,132 @@ import requests
 import skills
 from skills.common import config, vehicle
 from skills.routing import calculate_route
+import ollama
+
+### LLM Stuff ###
+from langchain_community.llms import Ollama
+from langchain.tools.base import StructuredTool
+
+from skills import (
+    get_weather,
+    find_route,
+    get_forecast,
+    vehicle_status as vehicle_status_fn,
+    search_points_of_interests,
+    search_along_route_w_coordinates,
+    do_anything_else,
+    date_time_info
+)
+from skills import extract_func_args
+
+
+global_context = {
+    "vehicle": vehicle,
+    "query": "How is the weather?",
+    "route_points": [],
+}
+
+
+MODEL_FUNC = "nexusraven"
+MODEL_GENERAL = "llama3:instruct"
+
+RAVEN_PROMPT_FUNC = """You are a helpful AI assistant in a car (vehicle), that follows instructions extremely well. \
+Answer questions concisely and do not mention what you base your reply on."
+
+{raven_tools}
+
+{history}
+
+User Query: Question: {input}<human_end>
+"""
+
+def get_prompt(template, input, history, tools):
+    # "vehicle_status": vehicle_status_fn()[0]
+    kwargs = {"history": history, "input": input}
+    prompt = "<human>:\n"
+    for tool in tools:
+        func_signature, func_docstring = tool.description.split(" - ", 1)
+        prompt += f'Function:\n<func_start>def {func_signature}<func_end>\n<docstring_start>\n"""\n{func_docstring}\n"""\n<docstring_end>\n'
+    kwargs["raven_tools"] = prompt
+
+    if history:
+        kwargs["history"] = f"Previous conversation history:{history}\n"
+
+    return template.format(**kwargs).replace("{{", "{").replace("}}", "}")
+
+def use_tool(func_name, kwargs, tools):
+    for tool in tools:
+        if tool.name == func_name:
+            return tool.invoke(input=kwargs)
+    return None
+
+tools = [
+    StructuredTool.from_function(get_weather),
+    StructuredTool.from_function(find_route),
+    # StructuredTool.from_function(vehicle_status),
+    StructuredTool.from_function(search_points_of_interests),
+    StructuredTool.from_function(search_along_route_w_coordinates),
+    StructuredTool.from_function(date_time_info),
+    StructuredTool.from_function(do_anything_else),
+]
+# llm = Ollama(model="nexusraven", stop=["\nReflection:", "\nThought:"], keep_alive=60*10)
 
 
 # Generate options for hours (00-23)
 hour_options = [f"{i:02d}:00" for i in range(24)]
 
+
 def set_time(time_picker):
     vehicle.time = time_picker
     return vehicle.model_dump_json()
 
+
 def get_vehicle_status(state):
     return state.value["vehicle"].model_dump_json()
 
 
+def run_generic_model(query):
+    print(f"Running the generic model with query: {query}")
+    data = {
+        "prompt": query,
+        "model": MODEL_GENERAL,
+        "options": {
+            # "temperature": 0.1,
+            # "stop":["\nReflection:", "\nThought:"]
+        }
+    }
+    out = ollama.generate(**data)
+    return out["response"]
+
+
+def run_model(query):
+    print("Query: ", query)
+    global_context["query"] = query
+    global_context["prompt"] = get_prompt(RAVEN_PROMPT_FUNC, query, "", tools)
+    print("Prompt: ", global_context["prompt"])
+    data = {
+        "prompt": global_context["prompt"],
+        # "streaming": False,
+        "model": "nexusraven",
+        # "model": "smangrul/llama-3-8b-instruct-function-calling",
+        "raw": True,
+        "options": {
+            "temperature": 0.5,
+            "stop":["\nReflection:", "\nThought:"]
+        }
+    }
+    out = ollama.generate(**data)
+    llm_response = out["response"]
+    if "Call: " in llm_response:
+        func_name, kwargs = extract_func_args(llm_response)
+        print(f"Function: {func_name}, Args: {kwargs}")
+        if func_name == "do_anything_else":
+            return run_generic_model(query)
+    
+        return use_tool(func_name, kwargs, tools)
+    return out["response"]
+
+
 # to be able to use the microphone on chrome, you will have to go to chrome://flags/#unsafely-treat-insecure-origin-as-secure and enter http://10.186.115.21:7860/
 # in "Insecure origins treated as secure", enable it and relaunch chrome
 
@@ -25,27 +138,28 @@ def get_vehicle_status(state):
 # What's the closest restaurant from here?
 
 
-model_answer = ""
-general_context = ""
-# Define the initial state with some initial context.
-print(general_context)
-initial_state = {"context": general_context}
-initial_context = initial_state["context"]
-# Create the Gradio interface.
-
-
 with gr.Blocks(theme=gr.themes.Default()) as demo:
     state = gr.State(
-        value={"context": initial_context, "query": "", "vehicle": vehicle, "route_points": []}
+        value={
+            # "context": initial_context,
+            "query": "",
+            "route_points": [],
+        }
     )
 
     with gr.Row():
         with gr.Column(scale=1, min_width=300):
             time_picker = gr.Dropdown(
-                choices=hour_options, label="What time is it? (HH:MM)", value="08:00", interactive=True
+                choices=hour_options,
+                label="What time is it? (HH:MM)",
+                value="08:00",
+                interactive=True,
             )
             history = gr.Radio(
-                ["Yes", "No"], label="Maintain the conversation history?", value="No", interactive=True
+                ["Yes", "No"],
+                label="Maintain the conversation history?",
+                value="No",
+                interactive=True,
             )
             voice_character = gr.Radio(
                 choices=[
@@ -76,7 +190,7 @@ with gr.Blocks(theme=gr.themes.Default()) as demo:
 
         with gr.Column(scale=2, min_width=600):
             map_plot = gr.Plot()
-            
+
             # map_if = gr.Interface(fn=plot_map, inputs=year_input, outputs=map_plot)
 
     with gr.Row():
@@ -87,14 +201,12 @@ with gr.Blocks(theme=gr.themes.Default()) as demo:
             input_text = gr.Textbox(
                 value="How is the weather?", label="Input text", interactive=True
             )
-            vehicle_status = gr.Textbox(
-                value=get_vehicle_status(state), label="Vehicle status", interactive=False
+            vehicle_status = gr.JSON(
+                value=vehicle.model_dump_json(), label="Vehicle status"
             )
         with gr.Column():
             output_audio = gr.Audio(label="output audio")
-            output_text = gr.TextArea(
-                value="", label="Output text", interactive=False
-            )
+            output_text = gr.TextArea(value="", label="Output text", interactive=False)
     # iface = gr.Interface(
     #     fn=transcript,
     #     inputs=[
@@ -113,19 +225,28 @@ with gr.Blocks(theme=gr.themes.Default()) as demo:
 
     # Update plot based on the origin and destination
     # Sets the current location and destination
-    origin.submit(fn=calculate_route, inputs=[origin, destination], outputs=[map_plot, vehicle_status])
-    destination.submit(fn=calculate_route, inputs=[origin, destination], outputs=[map_plot, vehicle_status])
+    origin.submit(
+        fn=calculate_route,
+        inputs=[origin, destination],
+        outputs=[map_plot, vehicle_status],
+    )
+    destination.submit(
+        fn=calculate_route,
+        inputs=[origin, destination],
+        outputs=[map_plot, vehicle_status],
+    )
 
     # Update time based on the time picker
     time_picker.select(fn=set_time, inputs=[time_picker], outputs=[vehicle_status])
 
+    # Run the model if the input text is changed
+    input_text.submit(fn=run_model, inputs=[input_text], outputs=[output_text])
+
 # close all interfaces open to make the port available
 gr.close_all()
 # Launch the interface.
 
 if __name__ == "__main__":
-    demo.launch(
-        debug=True, server_name="0.0.0.0", server_port=7860, ssl_verify=False
-    )
+    demo.launch(debug=True, server_name="0.0.0.0", server_port=7860, ssl_verify=False)
 
 # iface.launch(debug=True, share=False, server_name="0.0.0.0", server_port=7860, ssl_verify=False)

skills/__init__.py

@@ -1,12 +1,29 @@
+from datetime import datetime
 import inspect
 
-from .common import execute_function_call, extract_func_args, vehicle
+from .common import execute_function_call, extract_func_args, vehicle as vehicle_obj
 from .weather import get_weather, get_forecast
 from .routing import find_route
 from .poi import search_points_of_interests, search_along_route_w_coordinates
 from .vehicle import vehicle_status
 
 
+
+def date_time_info():
+    """Get the current date and time."""
+    time = getattr(vehicle_obj, "time")
+    date = getattr(vehicle_obj, "date")
+    datetime_obj = datetime.fromisoformat(f"{date}T{time}")
+    human_readable_datetime = datetime_obj.strftime("%I:%M %p %A, %B %d, %Y")
+    return f"It is {human_readable_datetime}."
+
+
+def do_anything_else():
+    """If the user wants to do anything else call this function. If the question doesn't match any of the functions use this one."""
+    return True
+
+
+
 def format_functions_for_prompt_raven(*functions):
     """Format functions for use in Prompt Raven.
     

skills/routing.py

@@ -109,8 +109,8 @@ def find_route_tomtom(
 
 
 def find_route(destination=""):
-    """
-    Find a route and return the distance and the estimated time to go to a specific destination from the current location.
+    """This function finds a route to a destination and returns the distance and the estimated time to go to a specific destination\
+ from the current location.
     :param destination (string): Required. The destination
     """
     # lat, lon, city = check_city_coordinates(lat_depart,lon_depart,city_depart)
@@ -153,7 +153,5 @@ def find_route(destination=""):
     arrival_hour_display = arrival_time.strftime("%H:%M")
 
     # return the distance and time
-    return (
-        f"The route to {destination} is {distance_km:.2f} km and {time_display}. Leaving now, the arrival time is estimated at {arrival_hour_display}.",
-        raw_response["routes"][0]["legs"][0]["points"],
-    )
+    return f"The route to {destination} is {distance_km:.2f} km which takes {time_display}. Leaving now, the arrival time is estimated at {arrival_hour_display}."
+        # raw_response["routes"][0]["legs"][0]["points"]

skills/vehicle.py

@@ -1,7 +1,9 @@
 from .common import vehicle
 
 
-STATUS_TEMPLATE = """We are at {location}, current time: {time}, current date: {date} and our destination is: {destination}.
+STATUS_TEMPLATE = """
+We are at {location}, coordinates: {lat}, {lon},
+current time: {time}, current date: {date} and our destination is: {destination}.
 """
 
 
@@ -19,13 +21,15 @@ def vehicle_status() -> tuple[str, dict[str, str]]:
             "destination": "Kirchberg Campus, Kirchberg"
         }
     """
-    vs = {
-        "location": "Luxembourg Gare, Luxembourg",
-        "lat": 49.6000,
-        "lon": 6.1333,
-        "date": "2025-03-29",
-        "time": "08:00:20",
-        "destination": "Kirchberg Campus, Kirchberg"
-    }
+    # vs = {
+    #     "location": "Luxembourg Gare, Luxembourg",
+    #     "lat": 49.6000,
+    #     "lon": 6.1333,
+    #     "date": "2025-03-29",
+    #     "time": "08:00:20",
+    #     "destination": "Kirchberg Campus, Luxembourg"
+    # }
     vs = vehicle.dict()
+    vs["lat"] = vs["location_coordinates"][0]
+    vs["lon"] = vs["location_coordinates"][1]
     return  STATUS_TEMPLATE.format(**vs), vs

skills/weather.py

@@ -1,14 +1,19 @@
 import requests
 
-from .common import config
+from .common import config, vehicle
 
 #current weather API
-def get_weather(location:str= "", **kwargs):
+def get_weather(location:str= ""):
     """
     Returns the CURRENT weather in a specified location.
     Args:
-    location (string) : Required. The name of the location, could be a city or lat/longitude in the following format latitude,longitude (example: 37.7749,-122.4194).
+    location (string) : Required. The name of the location, could be a city or lat/longitude in the following format latitude,longitude (example: 37.7749,-122.4194). If the location is not specified, the function will return the weather in the current location.
     """
+
+    if location == "":
+        print(f"get_weather: location is empty, using the vehicle location. ({vehicle.location})")
+        location = vehicle.location
+
     # The endpoint URL provided by WeatherAPI
     url = f"http://api.weatherapi.com/v1/current.json?key={config.WEATHER_API_KEY}&q={location}&aqi=no"
     print(url)