123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155 |
- """
- Inspired by the OpenAI example found here:
- https://platform.openai.com/docs/guides/function-calling/parallel-function-calling
- """
- import datetime
- from openai import OpenAI
- import json
- client = OpenAI(api_key="EMPTY", base_url="http://localhost:2242/v1")
- models = client.models.list()
- model = models.data[0].id
- stream = True
- def get_current_date_utc():
- print("Calling get_current_date_utc client side.")
- return datetime.datetime.now(datetime.timezone.utc).strftime(
- "The current UTC datetime is (day: %A, date (day/month/year): %d/%m/%Y, time: %H:%M)."
- )
- # Example dummy function hard coded to return the same weather
- # In production, this could be your backend API or an external API
- def get_current_weather(location, unit="fahrenheit"):
- """Get the current weather in a given location"""
- print("Calling get_current_weather client side.")
- if "tokyo" in location.lower():
- return json.dumps({
- "location": "Tokyo",
- "temperature": "10",
- "unit": unit
- })
- elif "san francisco" in location.lower():
- return json.dumps({
- "location": "San Francisco",
- "temperature": "72",
- "unit": unit
- })
- elif "paris" in location.lower():
- return json.dumps({
- "location": "Paris",
- "temperature": "22",
- "unit": unit
- })
- else:
- return json.dumps({"location": location, "temperature": "unknown"})
- def run_conversation():
- # Step 1: send the conversation and available functions to the model
- # messages = [{"role": "user", "content": "What's the weather like in San Francisco, Tokyo, and Paris?"}]
- messages = [{
- "role":
- "user",
- "content":
- "What's the weather like in San Francisco, Tokyo, and Paris ? We also need to know the current date."
- }]
- tools = [{
- "type": "function",
- "function": {
- "name": "get_current_weather",
- "description": "Get the current weather in a given location",
- "parameters": {
- "type": "object",
- "properties": {
- "location": {
- "type":
- "string",
- "description":
- "The city and state, e.g. San Francisco, CA",
- },
- "unit": {
- "type": "string",
- "enum": ["celsius", "fahrenheit"]
- },
- },
- "required": ["location"],
- },
- },
- }, {
- "type": "function",
- "function": {
- "name": "get_current_date_utc",
- "description": "Get the current UTC time",
- },
- }]
- response = client.chat.completions.create(
- model=model,
- messages=messages,
- tools=tools,
- stream=stream,
- tool_choice="auto", # auto is default, but we'll be explicit
- )
- response_message = ""
- tool_calls = None
- if stream:
- text_message = ""
- for chunk in response:
- if chunk.choices[0].finish_reason is not None:
- if chunk.choices[0].finish_reason == "tool_calls":
- tool_calls = chunk.choices[0].delta.tool_calls
- break
- if chunk.choices[0].delta.content is not None:
- text_message += chunk.choices[0].delta.content
- response_message = {"role": "assistant", "content": text_message}
- else:
- if not len(response.choices):
- return None
- response_message = response.choices[0].message
- # print(str(response_message))
- tool_calls = response_message.tool_calls
- # Step 2: check if the model wanted to call a function
- if tool_calls:
- # Step 3: call the function
- # Note: the JSON response may not always be valid; be sure to handle errors
- available_functions = {
- "get_current_weather": get_current_weather,
- "get_current_date_utc": get_current_date_utc,
- }
- messages.append(
- response_message) # extend conversation with assistant's reply
- # Step 4: send the info for each function call and function response to the model
- for tool_call in tool_calls:
- function_name = tool_call.function.name
- function_to_call = available_functions[function_name]
- if function_name == "get_current_weather":
- function_args = json.loads(tool_call.function.arguments)
- function_response = function_to_call(
- location=function_args.get("location"),
- unit=function_args.get("unit"),
- )
- else:
- function_response = function_to_call()
- messages.append({
- "tool_call_id": tool_call.id,
- "role": "tool",
- "name": function_name,
- "content": function_response,
- }) # extend conversation with function response
- second_response = client.chat.completions.create(
- model=model,
- messages=messages,
- ) # get a new response from the model where it can see the function response
- for it_msg, msg in enumerate(messages):
- print("Message %i:\n %s\n" % (it_msg, str(msg)))
- return second_response
- result = run_conversation()
- print("Final response:\n%s" % result)
|