Weather App with Mistral Function Call

Here, we demonstrate Mistral’s function calling with a weather app. The original Mistral function calling tutorial simulates function calls, but here, we want to use a real third-party API. We will go over:

  1. User specifies tools.
  2. User specifies query.
  3. Model generates function arguments.
  4. User executes a function to get tool results + generate a final answer.

Setup

from mistralai.client import MistralClient
from mistralai.models.chat_completion import ChatMessage
import os

mistral_api_key = os.getenv('MISTRAL_API_KEY')
client = MistralClient(api_key=mistral_api_key)

User Specifies Tools

Users can define tools for their use cases. For our use case, we have one function as our one tool: get_weather to retrieve weather data given a city name. Note that this is a very simple function that will NOT be able to look up zipcode or fancier locations such as city and state. The streamlit app at the end of the article has more sophisticated code that accepts zipcode, city and state, and unit type.

def get_weather(location):
    # First, we need to get the latitude and longitude for the location
    geocoding_url = f"https://geocoding-api.open-meteo.com/v1/search?name={location}&count=1&language=en&format=json"
    geo_response = requests.get(geocoding_url)

    if geo_response.status_code != 200:
        return {"error": "Unable to find location"}

    geo_data = geo_response.json()
    if not geo_data.get("results"):
        return {"error": "Location not found"}

    lat = geo_data["results"][0]["latitude"]
    lon = geo_data["results"][0]["longitude"]

    # Now we can get the weather data
    weather_url = f"https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}&current=temperature_2m,relative_humidity_2m,weather_code&forecast_days=1"
    weather_response = requests.get(weather_url)

    if weather_response.status_code != 200:
        return {"error": "Unable to fetch weather data"}

    weather_data = weather_response.json()

    # Convert weather code to description
    weather_codes = {
        0: "Clear sky",
        1: "Mainly clear", 2: "Partly cloudy", 3: "Overcast",
        45: "Fog", 48: "Depositing rime fog",
        51: "Light drizzle", 53: "Moderate drizzle", 55: "Dense drizzle",
        61: "Slight rain", 63: "Moderate rain", 65: "Heavy rain",
        71: "Slight snow fall", 73: "Moderate snow fall", 75: "Heavy snow fall",
        77: "Snow grains",
        80: "Slight rain showers", 81: "Moderate rain showers", 82: "Violent rain showers",
        85: "Slight snow showers", 86: "Heavy snow showers",
        95: "Thunderstorm", 96: "Thunderstorm with slight hail", 99: "Thunderstorm with heavy hail"
    }

    weather_code = weather_data["current"]["weather_code"]
    weather_description = weather_codes.get(weather_code, "Unknown")

    return {
        "temperature": weather_data["current"]["temperature_2m"],
        "humidity": weather_data["current"]["relative_humidity_2m"],
        "condition": weather_description
    }

For Mistral models to understand the functions, we need specify the function with a JSON schema: type, function name, function description, function parameters, and the required parameter for the function. Since we only have one function here, let’s define one function specification in a list. Note that this function specification calls the get_weather function above, with one required parameter (location):

# Define the function schema
weather_function = {
    "type": "function",
    "function": {
      "name": "get_weather",
      "description": "Get the current weather for a location",
      "parameters": {
        "type": "object",
        "properties": {
            "location": {
                "type": "string",
                "description": "The city name, e.g. New York"
            }
        },
        "required": ["location"],
      }
    }
}

User Specifies Query

Let’s ask about the weather of a location:

# Set up the conversation
messages = [
    ChatMessage(role="user", content="What's the weather like in Tokyo?")
]

Model Generates Function Arguments

We provide Mistral models with both the user query and the tool specifications (JSON schema) to inform them about these functions.

This step will NOT run the function directly, but rather to 1) determine the appropriate function to use, 2) identify any essential information missing for a function, and 3) generate necessary arguments for the chosen function.

Note that in our case there’s only one function, get_weather. Users can use tool_choice to specify how tools are used:

  • “auto”: default mode. Model decides if it uses the tool or not.
  • “any”: forces tool use.
  • “none”: prevents tool use.
# Make the API call
response = client.chat(
    model="mistral-small-latest",
    messages=messages,
    tools=[weather_function],
    tool_choice="auto"
)

response.choices[0].message

We get the response including tool_call with the chosen function name get_weather and the arguments for this function:

ChatMessage(role='assistant', content='', name=None, tool_calls=[ToolCall(id='TxR6ZEqSM', type=<ToolType.function: 'function'>, function=FunctionCall(name='get_weather', arguments='{"location": "Tokyo"}'))], tool_call_id=None)

Let’s add the response message to the message list:

messages.append(response.choices[0].message)

User Executes Function to Get Tool Results + Model Generates Final Answer

Currently, it’s the user’s responsibility to execute these functions, and the function execution lies on the user side. The function below extracts useful information from the model response, including function_name and function_params, and creates a final prompt for Mistral to get a final answer.

# If a function was called, execute it and generate a response
function_call = response.choices[0].message.tool_calls[0].function

if function_call:
    function_name = function_call.name
    function_args = json.loads(function_call.arguments)

    if function_name == "get_weather":
        weather_data = get_weather(function_args["location"])

        # Generate a response using the weather data
        response = client.chat(
            model="mistral-small-latest",
            messages=messages + [
                ChatMessage(role="tool", content=json.dumps(weather_data), name="get_weather")
            ]
        )
        print(response.choices[0].message.content)
else:
    print(response.choices[0].message.content)

Response:

The current weather in Tokyo is 20.9 degrees Celsius with 98% humidity. It’s currently moderately drizzling.

Weather App with Streamlit

Finally let’s create a Streamlit app around the weather function calls. Here’s what the user interface looks like:

The full Python code is similar to the code above, but has more functionalities such as accepting zip code, city and state, and different temperature units:

import streamlit as st
import requests
import json
import os

from mistralai.client import MistralClient
from mistralai.models.chat_completion import ChatMessage

# Initialize the Mistral client
mistral_api_key = os.getenv('MISTRAL_API_KEY')
client = MistralClient(api_key=mistral_api_key)

def get_coordinates(location):
    # Check if the location is a zip code (assuming 5-digit US zip codes)

    if location.isdigit() and len(location) == 5:
        # Use Zippopotam.us for zip code lookup
        zip_url = f"https://api.zippopotam.us/us/{location}"
        zip_response = requests.get(zip_url)
        
        if zip_response.status_code != 200:
            return None, "Unable to find location"
        
        zip_data = zip_response.json()
        # print('zip data:', zip_data)
        
        if not zip_data:
            return None, "Location not found"
        
        lat = float(zip_data["places"][0]["latitude"])
        lon = float(zip_data["places"][0]["longitude"])

        # Extract formatted location
        formatted_location = zip_data["places"][0]["place name"]+', '+zip_data["places"][0]["state"]

    else:
        # Use Nominatim for geocoding
        nominatim_url = f"https://nominatim.openstreetmap.org/search?q={location}&format=json&limit=1"
        headers = {
            "User-Agent": "WeatherApp/1.0"  # It's good practice to identify your application
        }
        geo_response = requests.get(nominatim_url, headers=headers)
        
        if geo_response.status_code != 200:
            return None, "Unable to find location"
        
        geo_data = geo_response.json()
        # print('geo data:', geo_data)
        
        if not geo_data:
            return None, "Location not found"
        
        result = geo_data[0]
        lat = float(result["lat"])
        lon = float(result["lon"])
        
        # Extract formatted location
        formatted_location = result.get("display_name", location)

    return (lat, lon, formatted_location), None


def get_weather(location, use_fahrenheit):
    coordinates, error = get_coordinates(location)
    if error:
        return {"error": error}
    
    lat, lon, formatted_location = coordinates
    
    # Weather API call
    temperature_unit = "fahrenheit" if use_fahrenheit else "celsius"
    weather_url = f"https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}&current=temperature_2m,relative_humidity_2m,weather_code&temperature_unit={temperature_unit}&forecast_days=1"
    weather_response = requests.get(weather_url)
    
    if weather_response.status_code != 200:
        return {"error": "Unable to fetch weather data"}
    
    weather_data = weather_response.json()
    # print('weather_data:', weather_data)
    
    # Convert weather code to description
    weather_codes = {
        0: "Clear sky", 1: "Mainly clear", 2: "Partly cloudy", 3: "Overcast",
        45: "Fog", 48: "Depositing rime fog",
        51: "Light drizzle", 53: "Moderate drizzle", 55: "Dense drizzle",
        61: "Slight rain", 63: "Moderate rain", 65: "Heavy rain",
        71: "Slight snow fall", 73: "Moderate snow fall", 75: "Heavy snow fall",
        77: "Snow grains",
        80: "Slight rain showers", 81: "Moderate rain showers", 82: "Violent rain showers",
        85: "Slight snow showers", 86: "Heavy snow showers",
        95: "Thunderstorm", 96: "Thunderstorm with slight hail", 99: "Thunderstorm with heavy hail"
    }
    
    weather_code = weather_data["current"]["weather_code"]
    weather_description = weather_codes.get(weather_code, "Unknown")
    
    return {
        "temperature": weather_data["current"]["temperature_2m"],
        "humidity": weather_data["current"]["relative_humidity_2m"],
        "condition": weather_description,
        "unit": "°F" if use_fahrenheit else "°C",
        "location": formatted_location
    }


# Define the function schema
weather_function = {
    "type": "function",
    "function": {
      "name": "get_weather",
      "description": "Get the current weather for a location",
      "parameters": {
        "type": "object",
        "properties": {
            "location": {
                "type": "string",
                "description": "The city name or ZIP code, e.g. New York or 10001"
            }
        },
        "required": ["location", "use_fahrenheit"],
      }
    }
}

# Streamlit app
st.title("Weather App with Mistral AI")

# User input
location = st.text_input("Enter a location (zip code or some combination of city/state/country):")

# Temperature unit toggle
use_fahrenheit = st.toggle("Use Fahrenheit", value=False)

if st.button("Get Weather"):
    if location:
        with st.spinner("Fetching weather information..."):
            # Set up the conversation
            messages = [
                ChatMessage(role="user", content=f"What's the weather like in {location}? Please provide the temperature in {'Fahrenheit' if use_fahrenheit else 'Celsius'}.")
            ]

            # Make the API call to Mistral
            response = client.chat(
                model="mistral-large-latest",
                messages=messages,
                tools=[weather_function],
                tool_choice="auto"
            )

            # Add response
            messages.append(response.choices[0].message)

            # Extract the function call
            function_call = response.choices[0].message.tool_calls[0].function

            # If a function was called, execute it and generate a response
            if function_call:
                function_name = function_call.name
                function_args = json.loads(function_call.arguments)

                if function_name == "get_weather":
                    weather_data = get_weather(function_args["location"], use_fahrenheit)
                    
                    if "error" in weather_data:
                        st.error(f"Error: {weather_data['error']}")
                    else:
                        # Generate a response using the weather data
                        response = client.chat(
                            model="mistral-large-latest",
                            messages=messages + [
                                ChatMessage(role="tool", content=json.dumps(weather_data), name="get_weather")
                            ]
                        )

                        st.success(response.choices[0].message.content)

                        # Display raw weather data
                        st.subheader("Raw Weather Data")
                        st.json(weather_data)
            else:
                st.warning("Unable to process the weather request.")
    else:
        st.warning("Please enter a location.")

That’s all for now!