LLM-Context und Tools

import requests

def chat(model_name: str, question: str) -> tuple[str, str]:
    url = "http://localhost:11434/api/chat"
    data = {
        "model": model_name,
        "messages": [
            {
                "role": "user",
                "content": question
            }
        ],
        "stream": False
    }

    response_chat = requests.post(url, json=data)
    content = _extract_chat_content(response_chat.json())
    return model_name, content

def _extract_chat_content(json_data: dict) -> str:
    return json_data["message"]['content']


if __name__ == '__main__':
    my_question = 'What kind of clothes do I need for a short trip to Paris? Please answer in max 5 sentences.'
    response = chat('llama3.2', my_question)
    print(response)

# definition of the supported tools
forecast_function = {
    'type': 'function',
    'function': {
        'name': 'get_forecast',
        'description': 'The weather forecast for a city',
        'parameters': {
            'type': 'object',
            'required': ['city'],
            'properties': {
                'city': {'type': 'string', 'description': 'The name of the city'},
            },
        },
    },
}

# implementation of the supported tools
def get_forecast(city: str) -> str:
    if city == 'Paris':
        forecasts = [
            'temperature: 30 celsius',
            'wind: 5 km/h',
            'precipitation: 0%',
        ]
        return "\n---\n".join(forecasts)
    elif city == 'London':
        forecasts = [
            'temperature: 20 celsius',
            'wind: 20 km/h',
            'precipitation: 80%',
        ]
        return "\n---\n".join(forecasts)
    elif city == 'Berlin':
        forecasts = [
            'temperature: 15 celsius',
            'wind: 10 km/h',
            'precipitation: 0%',
        ]
        return "\n---\n".join(forecasts)
    else:
        return "\n---\n".join([])

from via_rest.weather.with_tools.tools import forecast_function

class Context:

    def __init__(self, model_name: str):
        self.model_name = model_name
        self.data = {
            "model": model_name,
            "messages": [
            ],
            "tools": [forecast_function],
            "stream": False
        }

    def add_message(self, message):
        self.data["messages"].append(message)

    def add_request_message(self, question: str):
        message = {
            "role": "user",
            "content": question
        }
        self.add_message(message)

    def add_tool_message_with_tool_results(self, function_name: str, tool_result: str):
        message = {
            'role': 'tool',
            'content': tool_result,
            'tool_name': function_name}
        self.add_message(message)

import requests
from via_rest.weather.with_tools.context import Context
from via_rest.weather.with_tools.tools import get_forecast

def _send_request(data: dict) -> dict:
    url = "http://localhost:11434/api/chat"
    return requests.post(url, json=data).json()


available_functions = {
    'get_forecast': get_forecast,
}


def chat_with_tools(model_name: str, question: str) -> str:
    context = Context(model_name)
    context.add_request_message(question)
    first_response = _send_request(context.data)

    # check, if we should call a tool for the model
    if tool_call_requests := first_response['message']['tool_calls']:
        for tool_call_request in tool_call_requests:
            function_name = tool_call_request['function']['name']
            function_arguments = tool_call_request['function']['arguments']

            # check if the function is available
            if function_name not in available_functions:
                continue
            else:
                # call the function locally
                function_to_call = available_functions.get(function_name)
                output = function_to_call(**function_arguments)

                # add response message of model
                context.add_message(first_response['message'])

                # add function result
                context.add_tool_message_with_tool_results(function_name, str(output))

    if tool_call_requests:
        final_response = _send_request(context.data)
        return final_response['message']['content']
    else:
        return ''


if __name__ == '__main__':
    my_question = 'What kind of clothes do I need for a short trip to Paris? Please answer in max 5 sentences.'
    response = chat_with_tools('llama3.2', my_question)
    print('Final LLM response:', response)

[REQUEST] make initial request with tool definitions
  messages
    {
      "role": "user",
      "content": "What kind of clothes do I need for a short trip to Paris? Please answer in max 5 sentences."
    }
  tools
    {
      "type": "function",
      "function": {
        "name": "get_forecast",
        "description": "The weather forecast for a city",
        "parameters": {
          "type": "object",
          "required": [
            "city"
          ],
          "properties": {
            "city": {
              "type": "string",
              "description": "The name of the city"
            }
          }
        }
      }
    }

[RESPONSE] from LLM
  message:
    {
      "role": "assistant",
      "content": "",
      "tool_calls": [
        {
          "function": {
            "name": "get_forecast",
            "arguments": {
              "city": "Paris"
            }
          }
        }
      ]
    }

[TOOL] Extracted tool call request
  function_name: get_forecast
  arguments: {'city': 'Paris'}

[TOOL]
  calling function get_forecast with arguments {'city': 'Paris'}

[TOOL]
  function output:
    temperature: 30 celsius
    ---
    wind: 5 km/h
    ---
    precipitation: 0%

[REQUEST] make request with tool results
  messages
    {
      "role": "user",
      "content": "What kind of clothes do I need for a short trip to Paris? Please answer in max 5 sentences."
    },
    {
      "role": "assistant",
      "content": "",
      "tool_calls": [
        {
          "id": "call_rng1wyam",
          "function": {
            "index": 0,
            "name": "get_forecast",
            "arguments": {
              "city": "Paris"
            }
          }
        }
      ]
    },
    {
      "role": "tool",
      "content": "temperature: 30 celsius\n---\nwind: 5 km/h\n---\nprecipitation: 0%",
      "tool_name": "get_forecast"
    }

  tools
    {
      "type": "function",
      "function": {
        "name": "get_forecast",
        "description": "The weather forecast for a city",
        "parameters": {
          "type": "object",
          "required": [
            "city"
          ],
          "properties": {
            "city": {
              "type": "string",
              "description": "The name of the city"
            }
          }
        }
      }
    }

[RESPONSE]
  message:
    {
      "role": "assistant",
      "content": "Based on the forecast, Paris is expected to be warm and sunny during your trip. For a short trip, consider packing lightweight and versatile clothing such as t-shirts, tank tops, shorts, and dresses that can keep you cool in the daytime. You may also want to bring a light sweater or cardigan for cooler evenings. Comfortable walking shoes are a must for exploring the city. Don't forget to pack a hat and sunglasses for sun protection!"
    }

package viaRest.weather;

import com.google.gson.Gson;
import com.google.gson.JsonArray;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Function;

import static viaRest.weather.Tools.getForecast;

public class WeatherWithTools {

    private static final Gson gson = new Gson();
    private static final HttpClient httpClient = HttpClient.newHttpClient();
    private static final String URL = "http://localhost:11434/api/chat";

    private static JsonObject sendRequest(JsonObject data) throws Exception {
        String jsonData = gson.toJson(data);

        HttpRequest request = HttpRequest.newBuilder()
                .uri(URI.create(URL))
                .header("Content-Type", "application/json")
                .POST(HttpRequest.BodyPublishers.ofString(jsonData))
                .build();

        HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
        return gson.fromJson(response.body(), JsonObject.class);
    }

    private static final Map<String, Function<Map<String, String>, String>> availableFunctions = new HashMap<>();

    static {
        availableFunctions.put("get_forecast", args -> getForecast(args.get("city")));
    }

    public static String chatWithTools(String modelName, String question) throws Exception {
        Context context = new Context(modelName);
        context.addRequestMessage(question);
        JsonObject firstResponse = sendRequest(context.getData());

        // Check if we should call a tool for the model
        JsonElement toolCallsElement = firstResponse.getAsJsonObject("message").get("tool_calls");
        JsonArray toolCallRequests = null;

        if (toolCallsElement != null && !toolCallsElement.isJsonNull()) {
            toolCallRequests = toolCallsElement.getAsJsonArray();
        }

        if (toolCallRequests != null && toolCallRequests.size() > 0) {
            for (JsonElement toolCallElement : toolCallRequests) {
                JsonObject toolCallRequest = toolCallElement.getAsJsonObject();

                JsonObject functionObj = toolCallRequest.getAsJsonObject("function");
                String functionName = functionObj.get("name").getAsString();
                JsonObject functionArguments = functionObj.getAsJsonObject("arguments");

                Map<String, Object> argsMap = new HashMap<>();
                argsMap.put("function_name", functionName);
                argsMap.put("arguments", functionArguments);

                // Check if the function is available
                if (!availableFunctions.containsKey(functionName)) {
                    continue;
                } else {
                    // Call the function locally
                    Function<Map<String, String>, String> functionToCall = availableFunctions.get(functionName);

                    // Convert JsonObject arguments to Map<String, String>
                    Map<String, String> argMap = new HashMap<>();
                    for (String key : functionArguments.keySet()) {
                        argMap.put(key, functionArguments.get(key).getAsString());
                    }

                    String output = functionToCall.apply(argMap);

                    // Add response message of model
                    context.addMessage(firstResponse.getAsJsonObject("message"));

                    // Add function result
                    context.addToolMessageWithToolResults(functionName, output);
                }
            }
        }

        if (toolCallRequests != null && toolCallRequests.size() > 0) {
            JsonObject finalResponse = sendRequest(context.getData());
            return finalResponse.getAsJsonObject("message").get("content").getAsString();
        } else {
            return "";
        }
    }

    public static void main(String[] args) {
        try {
            String myQuestion = "What kind of clothes do I need for a short trip to Paris?";
            String response = chatWithTools("llama3.2", myQuestion);
            System.out.println("Final LLM response: " + response);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}