Extracting Schema for OpenAI Functions

In the code example below, we use the openai_function_call library to extract a schema to be used on OpenAIChatStream from a good old python function, so you don't need to write the schema yourself.

First you need to install the library:

pip install openai-function-call

Then simply annotate your function using @openai_function and you will be able to get a openai_schema out of it:

from typing import Literal, TypedDict

from openai_function_call import openai_function


class WeatherReturn(TypedDict):
    location: str
    forecast: str
    temperature: str

@openai_function
def get_current_weather(
    location: str, format: Literal["celsius", "fahrenheit"] = "celsius"
) -> WeatherReturn:
    """
    Gets the current weather in a given location, use this function for any questions related to the weather

    Parameters
    ----------
    location
        The city to get the weather, e.g. San Francisco. Guess the location from user messages

    format
        A string with the full content of what the given role said
    """

    return WeatherReturn(
        location=location,
        forecast="sunny",
        temperature="25 C" if format == "celsius" else "77 F",
    )


get_current_weather.openai_schema

    {'name': 'get_current_weather',
     'description': '\n    Gets the current weather in a given location, use this function for any questions related to the weather\n\n    Parameters\n    ----------\n    location\n        The city to get the weather, e.g. San Francisco. Guess the location from user messages\n\n    format\n        A string with the full content of what the given role said\n    ',
     'parameters': {'properties': {'location': {'type': 'string'},
       'format': {'default': 'celsius',
        'enum': ['celsius', 'fahrenheit'],
        'type': 'string'}},
      'required': ['format', 'location'],
      'type': 'object'}}

We can then use this schema directly on our stream:

import json
from typing import Union

from langstream import Stream, collect_final_output
from langstream.contrib import OpenAIChatStream, OpenAIChatDelta, OpenAIChatMessage

stream: Stream[str, Union[OpenAIChatDelta, WeatherReturn]] = OpenAIChatStream[
    str, OpenAIChatDelta
](
    "WeatherStream",
    lambda user_input: [
        OpenAIChatMessage(role="user", content=user_input),
    ],
    model="gpt-3.5-turbo",
    functions=[get_current_weather.openai_schema],
    temperature=0,
).map(
    lambda delta: get_current_weather(**json.loads(delta.content))
    if delta.role == "function" and delta.name == "get_current_weather"
    else delta
)

await collect_final_output(
    stream(
        "I'm in my appartment in Amsterdam, thinking... should I take an umbrella for my pet chicken?"
    )
)

    [{'location': 'Amsterdam', 'forecast': 'sunny', 'temperature': '25 C'}]

That's it, if you have any questions about this example, join our discord community and we can help you out.