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
|
from __future__ import annotations
import rich
from pydantic import BaseModel
import openai
from openai import OpenAI
class GetWeather(BaseModel):
city: str
country: str
client = OpenAI()
with client.chat.completions.stream(
model="gpt-4o-2024-08-06",
messages=[
{
"role": "user",
"content": "What's the weather like in SF and New York?",
},
],
tools=[
# because we're using `.parse_stream()`, the returned tool calls
# will be automatically deserialized into this `GetWeather` type
openai.pydantic_function_tool(GetWeather, name="get_weather"),
],
parallel_tool_calls=True,
) as stream:
for event in stream:
if event.type == "tool_calls.function.arguments.delta" or event.type == "tool_calls.function.arguments.done":
rich.get_console().print(event, width=80)
print("----\n")
rich.print(stream.get_final_completion())
|
