cai/examples/customer_service_streaming/src/utils.py

44 lines
1.6 KiB
Python

def get_completion(client,
messages: list[dict[str, str]],
model: str = "gpt-4-0125-preview",
max_tokens=2000,
temperature=0.7,
tools=None,
stream=False,):
# Prepare the request parameters
request_params = {
"model": model,
"messages": messages,
"max_tokens": max_tokens,
"temperature": temperature,
"stream": stream,
}
if tools and isinstance(tools, list):
# Tools are already in dictionary format
request_params["tools"] = tools
# Make the API call with the possibility of streaming
if stream:
completion = client.chat.completions.create(**request_params)
# create variables to collect the stream of chunks
collected_chunks = []
collected_messages = []
for chunk in completion:
collected_chunks.append(chunk) # save the event response
# extract the message
chunk_message = chunk.choices[0].delta.content
collected_messages.append(chunk_message) # save the message
print(chunk_message, end="") # print the message
# yield chunk_message # Yield each part of the completion as it
# arrives
return collected_messages # Returns the whole completion
else:
completion = client.chat.completions.create(**request_params)
return completion.choices[0].message # Returns the whole completion
def is_dict_empty(d):
return all(not v for v in d.values())