chore: A new more advanced method

kitt/core/model.py

@@ -0,0 +1,224 @@
+import json
+import re
+import uuid
+
+from langchain.memory import ChatMessageHistory
+from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
+from langchain_core.utils.function_calling import convert_to_openai_function
+import ollama
+from pydantic import BaseModel
+from loguru import logger
+
+
+from kitt.skills import vehicle_status
+
+
+
+class FunctionCall(BaseModel):
+    arguments: dict
+    """
+    The arguments to call the function with, as generated by the model in JSON
+    format. Note that the model does not always generate valid JSON, and may
+    hallucinate parameters not defined by your function schema. Validate the
+    arguments in your code before calling your function.
+    """
+
+    name: str
+    """The name of the function to call."""
+
+schema_json = json.loads(FunctionCall.schema_json())
+HRMS_SYSTEM_PROMPT = """<|begin_of_text|>
+<|im_start|>system
+You are a function calling AI agent with self-recursion.
+You can call only one function at a time and analyse data you get from function response.
+You are provided with function signatures within <tools></tools> XML tags.
+{car_status}
+
+You may use agentic frameworks for reasoning and planning to help with user query.
+Please call a function and wait for function results to be provided to you in the next iteration.
+Don't make assumptions about what values to plug into function arguments.
+Once you have called a function, results will be fed back to you within <tool_response></tool_response> XML tags.
+Don't make assumptions about tool results if <tool_response> XML tags are not present since function hasn't been executed yet.
+Analyze the data once you get the results and call another function.
+At each iteration please continue adding the your analysis to previous summary.
+Your final response should directly answer the user query.
+
+
+Here are the available tools:
+<tools> {tools} </tools>
+If the provided function signatures doesn't have the function you must call, you may write executable python code in markdown syntax and call code_interpreter() function as follows:
+<tool_call>
+{{"arguments": {{"code_markdown": <python-code>, "name": "code_interpreter"}}}}
+</tool_call>
+Make sure that the json object above with code markdown block is parseable with json.loads() and the XML block with XML ElementTree.
+When using tools, ensure to only use the tools provided and not make up any data and do not provide any explanation as to which tool you are using and why.
+
+When asked for the weather, lookup the weather for the current location of the car. Unless the user provides a location, then use that location.
+If asked about points of interest, use the tools available to you. Do not make up points of interest.
+
+Use the following pydantic model json schema for each tool call you will make:
+{schema}
+
+At the very first turn you don't have <tool_results> so you shouldn't not make up the results.
+Please keep a running summary with analysis of previous function results and summaries from previous iterations.
+Do not stop calling functions until the task has been accomplished or you've reached max iteration of 10.
+If you plan to continue with analysis, always call another function.
+For each function call return a valid json object (using doulbe quotes) with function name and arguments within <tool_call></tool_call> XML tags as follows:
+<tool_call>
+{{"arguments": <args-dict>, "name": <function-name>}}
+</tool_call>
+<|im_end|>"""
+AI_PREAMBLE = """
+<|im_start|>assistant
+"""
+HRMS_TEMPLATE_USER = """
+<|im_start|>user
+{user_input}<|im_end|>"""
+HRMS_TEMPLATE_ASSISTANT = """
+<|im_start|>assistant
+{assistant_response}<|im_end|>"""
+HRMS_TEMPLATE_TOOL_RESULT = """
+<|im_start|>tool
+{result}
+<|im_end|>"""
+
+
+def append_message(prompt, h):     
+    if h.type == "human":
+        prompt += HRMS_TEMPLATE_USER.format(user_input=h.content)
+    elif h.type == "ai":
+        prompt += HRMS_TEMPLATE_ASSISTANT.format(assistant_response=h.content)
+    elif h.type == "tool":
+        prompt += HRMS_TEMPLATE_TOOL_RESULT.format(result=h.content)
+    return prompt
+
+
+def get_prompt(template, history, tools, schema, car_status=None):
+    if not car_status:
+        # car_status = vehicle.dict()
+        car_status = vehicle_status()[0]
+
+    # "vehicle_status": vehicle_status_fn()[0]
+    kwargs = {"history": history, "schema": schema, "tools": tools, "car_status": car_status}
+
+    
+    prompt = template.format(**kwargs).replace("{{", "{").replace("}}", "}")
+    
+    if history:
+        for h in history.messages:
+            prompt = append_message(prompt, h)
+
+    # if input:
+    #     prompt += USER_QUERY_TEMPLATE.format(user_input=input)
+    return prompt
+
+
+def use_tool(tool_call, tools):
+    func_name = tool_call["name"]
+    kwargs = tool_call["arguments"]
+    for tool in tools:
+        if tool.name == func_name:
+            return tool.invoke(input=kwargs)
+    return None
+
+
+def parse_tool_calls(text):
+    logger.debug(f"Start parsing tool_calls: {text}")
+    pattern = r'<tool_call>\s*(\{.*?\})\s*</tool_call>'
+
+    if not text.startswith("<tool_call>"):
+        return [], []
+
+    matches = re.findall(pattern, text, re.DOTALL)
+    tool_calls = []
+    errors = []
+    for match in matches:
+        try:
+            tool_call = json.loads(match)
+            tool_calls.append(tool_call)
+        except json.JSONDecodeError as e:
+            errors.append(f"Invalid JSON in tool call: {e}")
+    
+    logger.debug(f"Tool calls: {tool_calls}, errors: {errors}")
+    return tool_calls, errors
+
+
+def process_response(user_query, res, history, tools, depth):
+    """Returns True if the response contains tool calls, False otherwise."""
+    logger.debug(f"Processing response: {res}")
+    tool_calls, errors = parse_tool_calls(res)
+    # TODO: Handle errors
+    if not tool_calls:
+        return False
+    # tool_results = ""
+    tool_results = f"Agent iteration {depth} to assist with user query: {user_query}\n"
+    for tool_call in tool_calls:
+        # TODO: Extra Validation
+        # Call the function
+        try:
+            result = use_tool(tool_call, tools)
+            if type(result) == tuple:
+                result = result[1]
+            tool_results += f"<tool_response>\n{result}\n</tool_response>\n"
+        except Exception as e:
+            print(e)
+    # Currently only to mimic OpneAI's behavior
+    # But it could be used for tracking function calls
+
+    tool_results = tool_results.strip()
+    print(f"Tool results: {tool_results}")
+    tool_call_id = uuid.uuid4().hex
+    history.add_message(ToolMessage(content=tool_results, tool_call_id=tool_call_id))
+    return True
+
+
+def run_inference_step(history, tools, schema_json, dry_run=False):
+    # If we decide to call a function, we need to generate the prompt for the model
+    # based on the history of the conversation so far.
+    # not break the loop
+    openai_tools = [convert_to_openai_function(tool) for tool in tools]
+    prompt = get_prompt(HRMS_SYSTEM_PROMPT, history, openai_tools, schema_json)
+    print(f"Prompt is:{prompt + AI_PREAMBLE}\n------------------\n")
+
+    data = {
+        "prompt": prompt + AI_PREAMBLE,
+        # "streaming": False,
+        # "model": "smangrul/llama-3-8b-instruct-function-calling",
+        # "model": "elvee/hermes-2-pro-llama-3:8b-Q5_K_M",
+        # "model": "NousResearch/Hermes-2-Pro-Llama-3-8B",
+        "model": "interstellarninja/hermes-2-pro-llama-3-8b",
+        "raw": True,
+        "options": {"temperature": 0.8,
+                    # "max_tokens": 1500,
+                    "num_predict": 1500,
+                    # "num_predict": 1500,
+                # "max_tokens": 1500,
+        }
+    }
+
+    if dry_run:
+        print(prompt + AI_PREAMBLE)
+        return "Didn't really run it."
+
+    out = ollama.generate(**data)
+
+    res = out["response"]
+
+    return res
+
+
+def process_query(user_query: str, history: ChatMessageHistory, tools):
+    history.add_message(HumanMessage(content=user_query))
+    for depth in range(10):
+        out = run_inference_step(history, tools, schema_json)
+        print(f"Inference step result:\n{out}\n------------------\n")
+        history.add_message(AIMessage(content=out))
+        if not process_response(user_query, out, history, tools, depth):
+            print(f"This is the answer, no more iterations: {out}")
+            return out
+        # Otherwise, tools result is already added to history, we just need to continue the loop.
+    # If we get here something went wrong.
+    history.add_message(
+        AIMessage(content="Sorry, I am not sure how to help you with that.")
+    )
+    return "Sorry, I am not sure how to help you with that."

kitt/core/utils.py

@@ -0,0 +1,35 @@
+from typing import List, Tuple, Optional, Union
+
+
+def plot_route(points, vehicle: Union[tuple[float, float], None] = None):
+    import plotly.express as px
+    
+    lats = []
+    lons = []
+
+    for point in points:
+        lats.append(point["latitude"])
+        lons.append(point["longitude"])
+    # fig = px.line_geo(lat=lats, lon=lons)
+    # fig.update_geos(fitbounds="locations")
+
+    fig = px.line_mapbox(
+        lat=lats, lon=lons, zoom=12, height=600, color_discrete_sequence=["red"]
+    )
+
+    if vehicle:
+        fig.add_trace(
+            px.scatter_mapbox(
+                lat=[vehicle[0]],
+                lon=[vehicle[1]],
+                color_discrete_sequence=["blue"],
+            ).data[0]
+        )
+
+    fig.update_layout(
+        mapbox_style="open-street-map",
+        # mapbox_zoom=12,
+    )
+    fig.update_geos(fitbounds="locations")
+    fig.update_layout(margin={"r": 20, "t": 20, "l": 20, "b": 20})
+    return fig

kitt/skills/__init__.py

@@ -2,7 +2,7 @@ from datetime import datetime
 import inspect
 
 from .common import execute_function_call, extract_func_args, vehicle as vehicle_obj
-from .weather import get_weather, get_forecast
+from .weather import get_weather_current_location, 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

kitt/skills/routing.py

@@ -17,31 +17,6 @@ def find_coordinates(address):
     return lat, lon
 
 
-def plot_route(points):
-    import plotly.express as px
-
-    lats = []
-    lons = []
-
-    for point in points:
-        lats.append(point["latitude"])
-        lons.append(point["longitude"])
-    # fig = px.line_geo(lat=lats, lon=lons)
-    # fig.update_geos(fitbounds="locations")
-
-    fig = px.line_mapbox(
-        lat=lats, lon=lons, zoom=12, height=600, color_discrete_sequence=["red"]
-    )
-
-    fig.update_layout(
-        mapbox_style="open-street-map",
-        # mapbox_zoom=12,
-    )
-    fig.update_geos(fitbounds="locations")
-    fig.update_layout(margin={"r": 20, "t": 20, "l": 20, "b": 20})
-    return fig
-
-
 def calculate_route(origin, destination):
     """This function is called when the origin or destination is updated in the GUI. It calculates the route between the origin and destination."""
     print(f"calculate_route(origin: {origin}, destination: {destination})")
@@ -64,7 +39,7 @@ def calculate_route(origin, destination):
     data = response.json()
     points = data["routes"][0]["legs"][0]["points"]
 
-    return plot_route(points), vehicle.model_dump_json(), points
+    return vehicle.model_dump_json(), points
 
 
 def find_route_tomtom(

kitt/skills/vehicle.py

@@ -2,8 +2,11 @@ from .common import vehicle
 
 
 STATUS_TEMPLATE = """
-We are at {location}, coordinates: {lat}, {lon},
-current time: {time}, current date: {date} and our destination is: {destination}.
+The current location is:{location}
+The current Geo coordinates: {lat}, {lon}
+The current time: {time}
+The current date: {date}
+The current destination is: {destination}
 """
 
 

kitt/skills/weather.py

@@ -3,12 +3,32 @@ import requests
 from .common import config, vehicle
 
 
+def get_weather_current_location():
+    """
+    Returns the CURRENT weather in current location.
+    When responding to user, only mention the weather condition, temperature, and the temperature that it feels like, unless the user asks for more information.
+
+    Returns:
+        dict: The weather data in the specified location.
+    """
+    print(
+        f"get_weather: location is empty, using the vehicle location. ({vehicle.location})"
+    )
+    location = vehicle.location
+    return get_weather(location)
+
+
 # current weather API
 def get_weather(location: str = ""):
     """
-    Returns the CURRENT weather in a specified location.
+    Get the current weather in a specified location.
+    When responding to user, only mention the weather condition, temperature, and the temperature that it feels like, unless the user asks for more information.
+
     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). If the location is not specified, the function will return the weather in the current location.
+        location (string) : Optional. The name of the location, if empty, the vehicle location is used.
+
+    Returns:
+        dict: The weather data in the specified location.
     """
 
     if location == "":
@@ -56,10 +76,10 @@ def get_weather(location: str = ""):
 # weather forecast API
 def get_forecast(city_name: str = "", when=0, **kwargs):
     """
-    Returns the weather forecast in a specified number of days for a specified city .
+    Get the weather forecast in a specified number of days for a specified location.
     Args:
-    city_name (string) : Required. The name of the city.
-    when (int) : Required. in number of days (until the day for which we want to know the forecast) (example: tomorrow is 1, in two days is 2, etc.)
+        city_name (string) : Required. The name of the city.
+        when (int) : Required. in number of days (until the day for which we want to know the forecast) (example: tomorrow is 1, in two days is 2, etc.)
     """
 
     when += 1

main.py

@@ -11,6 +11,10 @@ from kitt.skills.routing import calculate_route
 import ollama
 
 from langchain.tools.base import StructuredTool
+from langchain.memory import ChatMessageHistory
+from langchain_core.utils.function_calling import convert_to_openai_function
+from loguru import logger
+
 
 from kitt.skills import (
     get_weather,
@@ -21,9 +25,12 @@ from kitt.skills import (
     search_along_route_w_coordinates,
     do_anything_else,
     date_time_info,
+    get_weather_current_location
 )
 from kitt.skills import extract_func_args
 from kitt.core import voice_options, tts_gradio
+from kitt.core.model import process_query
+from kitt.core import utils as kitt_utils
 
 
 global_context = {
@@ -33,6 +40,7 @@ global_context = {
 }
 
 speaker_embedding_cache = {}
+history = ChatMessageHistory()
 
 MODEL_FUNC = "nexusraven"
 MODEL_GENERAL = "llama3:instruct"
@@ -111,11 +119,12 @@ def get_vehicle_status(state):
 tools = [
     StructuredTool.from_function(get_weather),
     StructuredTool.from_function(find_route),
-    # StructuredTool.from_function(vehicle_status),
+    StructuredTool.from_function(vehicle_status_fn),
     StructuredTool.from_function(search_points_of_interests),
     StructuredTool.from_function(search_along_route),
     StructuredTool.from_function(date_time_info),
-    StructuredTool.from_function(do_anything_else),
+    StructuredTool.from_function(get_weather_current_location),
+    # StructuredTool.from_function(do_anything_else),
 ]
 
 
@@ -133,6 +142,9 @@ def run_generic_model(query):
     return out["response"]
 
 
+def clear_history():
+    history.clear()
+
 
 def run_nexusraven_model(query, voice_character):
     global_context["prompt"] = get_prompt(RAVEN_PROMPT_FUNC, query, "", tools)
@@ -169,36 +181,13 @@ def run_nexusraven_model(query, voice_character):
 
 
 def run_llama3_model(query, voice_character):
-    global_context["prompt"] = get_prompt(RAVEN_PROMPT_FUNC, query, "", tools)
-    print("Prompt: ", global_context["prompt"])
-    data = {
-        "prompt": global_context["prompt"],
-        # "streaming": False,
-        # "model": "smangrul/llama-3-8b-instruct-function-calling",
-        "model": "elvee/hermes-2-pro-llama-3:8b-Q5_K_M",
-        "raw": True,
-        "options": {"temperature": 0.5, "stop": ["\nReflection:", "\nThought:"]},
-    }
-    out = ollama.generate(**data)
-    llm_response = out["response"]
-    if "Call: " in llm_response:
-        print(f"llm_response: {llm_response}")
-        llm_response = llm_response.replace("<bot_end>", " ")
-        func_name, kwargs = extract_func_args(llm_response)
-        print(f"Function: {func_name}, Args: {kwargs}")
-        if func_name == "do_anything_else":
-            output_text = run_generic_model(query)
-        else:
-            output_text = use_tool(func_name, kwargs, tools)
-    else:
-        output_text = out["response"]
-
-    if type(output_text) == tuple:
-        output_text = output_text[0]
+    output_text = process_query(query, history, tools)
     gr.Info(f"Output text: {output_text}, generating voice output...")
+    # voice_out = tts_gradio(output_text, voice_character, speaker_embedding_cache)[0]
+    voice_out = None
     return (
         output_text,
-        tts_gradio(output_text, voice_character, speaker_embedding_cache)[0],
+        voice_out,
     )
 
 
@@ -216,22 +205,28 @@ def run_model(query, voice_character, state):
 
 
 def calculate_route_gradio(origin, destination):
-    plot, vehicle_status, points = calculate_route(origin, destination)
+    vehicle_status, points = calculate_route(origin, destination)
+    plot = kitt_utils.plot_route(points, vehicle=vehicle.location_coordinates)
     global_context["route_points"] = points
     vehicle.location_coordinates = points[0]["latitude"], points[0]["longitude"]
-    return plot, vehicle_status
+    return plot, vehicle_status, 0
 
 
-def update_vehicle_status(trip_progress):
+def update_vehicle_status(trip_progress, origin, destination):
+    if not global_context["route_points"]:
+        vehicle_status, points = calculate_route(origin, destination)
+        global_context["route_points"] = points
     n_points = len(global_context["route_points"])
-    new_coords = global_context["route_points"][
-        min(int(trip_progress / 100 * n_points), n_points - 1)
-    ]
+    index = min(int(trip_progress / 100 * n_points), n_points - 1)
+    print(f"Trip progress: {trip_progress} len: {n_points}, index: {index}")
+    new_coords = global_context["route_points"][index]
     new_coords = new_coords["latitude"], new_coords["longitude"]
     print(f"Trip progress: {trip_progress}, len: {n_points}, new_coords: {new_coords}")
     vehicle.location_coordinates = new_coords
     vehicle.location = ""
-    return vehicle.model_dump_json()
+
+    plot = kitt_utils.plot_route(global_context["route_points"], vehicle=vehicle.location_coordinates)
+    return vehicle.model_dump_json(), plot
 
 
 device = "cuda" if torch.cuda.is_available() else "cpu"
@@ -278,7 +273,7 @@ def save_and_transcribe_audio(audio):
 # What's the closest restaurant from here?
 
 
-def create_demo(tts_server: bool = False, model="llama3"):
+def create_demo(tts_server: bool = False, model="llama3", tts=True):
     print(f"Running the demo with model: {model} and TTSServer: {tts_server}")
     with gr.Blocks(theme=gr.themes.Default()) as demo:
         state = gr.State(
@@ -287,6 +282,7 @@ def create_demo(tts_server: bool = False, model="llama3"):
                 "query": "",
                 "route_points": [],
                 "model": model,
+                "tts": tts,
             }
         )
         trip_points = gr.State(value=[])
@@ -344,6 +340,8 @@ def create_demo(tts_server: bool = False, model="llama3"):
                 vehicle_status = gr.JSON(
                     value=vehicle.model_dump_json(), label="Vehicle status"
                 )
+                # Push button
+                clear_history_btn = gr.Button(value="Clear History")
             with gr.Column():
                 output_audio = gr.Audio(label="output audio", autoplay=True)
                 output_text = gr.TextArea(
@@ -355,12 +353,12 @@ def create_demo(tts_server: bool = False, model="llama3"):
         origin.submit(
             fn=calculate_route_gradio,
             inputs=[origin, destination],
-            outputs=[map_plot, vehicle_status],
+            outputs=[map_plot, vehicle_status, trip_progress],
         )
         destination.submit(
             fn=calculate_route_gradio,
             inputs=[origin, destination],
-            outputs=[map_plot, vehicle_status],
+            outputs=[map_plot, vehicle_status, trip_progress],
         )
 
         # Update time based on the time picker
@@ -375,13 +373,17 @@ def create_demo(tts_server: bool = False, model="llama3"):
 
         # Set the vehicle status based on the trip progress
         trip_progress.release(
-            fn=update_vehicle_status, inputs=[trip_progress], outputs=[vehicle_status]
+            fn=update_vehicle_status, inputs=[trip_progress, origin, destination], outputs=[vehicle_status, map_plot]
         )
 
         # Save and transcribe the audio
         input_audio.stop_recording(
             fn=save_and_transcribe_audio, inputs=[input_audio], outputs=[input_text]
         )
+
+        # Clear the history
+        clear_history_btn.click(fn=clear_history, inputs=[], outputs=[])
+
     return demo
 
 
@@ -389,7 +391,7 @@ def create_demo(tts_server: bool = False, model="llama3"):
 gr.close_all()
 
 
-demo = create_demo(False, "llama3")
+demo = create_demo(False, "llama3", tts=False)
 demo.launch(
     debug=True,
     server_name="0.0.0.0",