File: gpt-oss-tools-stream.py

package info (click to toggle)
ollama-python 0.5.3-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 376 kB
  • sloc: python: 3,056; makefile: 5
file content (77 lines) | stat: -rw-r--r-- 2,247 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
import random
from typing import Iterator

from ollama import chat
from ollama._types import ChatResponse


def get_weather(city: str) -> str:
  """
  Get the current temperature for a city

  Args:
      city (str): The name of the city

  Returns:
      str: The current temperature
  """
  temperatures = list(range(-10, 35))

  temp = random.choice(temperatures)

  return f'The temperature in {city} is {temp}°C'


def get_weather_conditions(city: str) -> str:
  """
  Get the weather conditions for a city

  Args:
      city (str): The name of the city

  Returns:
      str: The current weather conditions
  """
  conditions = ['sunny', 'cloudy', 'rainy', 'snowy', 'foggy']
  return random.choice(conditions)


available_tools = {'get_weather': get_weather, 'get_weather_conditions': get_weather_conditions}

messages = [{'role': 'user', 'content': 'What is the weather like in London? What are the conditions in Toronto?'}]


model = 'gpt-oss:20b'
# gpt-oss can call tools while "thinking"
# a loop is needed to call the tools and get the results
final = True
while True:
  response_stream: Iterator[ChatResponse] = chat(model=model, messages=messages, tools=[get_weather, get_weather_conditions], stream=True)

  for chunk in response_stream:
    if chunk.message.content:
      if not (chunk.message.thinking or chunk.message.thinking == '') and final:
        print('\nFinal result: ')
        final = False
      print(chunk.message.content, end='', flush=True)
    if chunk.message.thinking:
      print(chunk.message.thinking, end='', flush=True)

  print()

  if chunk.message.tool_calls:
    for tool_call in chunk.message.tool_calls:
      function_to_call = available_tools.get(tool_call.function.name)
      if function_to_call:
        print('\nCalling tool: ', tool_call.function.name, 'with arguments: ', tool_call.function.arguments)
        result = function_to_call(**tool_call.function.arguments)
        print('Tool result: ', result + '\n')

        messages.append(chunk.message)
        messages.append({'role': 'tool', 'content': result, 'tool_name': tool_call.function.name})
      else:
        print(f'Tool {tool_call.function.name} not found')

  else:
    # no more tool calls, we can stop the loop
    break