OpenAI API Python Sample Code and Example Output (GPT-4o Compatible)
In this article, we introduce sample code for using the OpenAI API.
The OpenAI API is not limited to OpenAI's own service; other inference engines such as vLLM also expose OpenAI-compatible API servers, and it is becoming the de facto standard for LLM serving APIs. So it is worth getting familiar with the coding conventions.
A simple code sample for accessing the API of OpenAI's GPT series looks like the following. Let's receive the generated output incrementally via streaming.
Sample Code: Quick Start
import asyncio
import os
import traceback
from openai import AsyncOpenAI
async def main() -> None:
try:
# Specify the model name
# model="gpt-4-turbo" # $10.00/MTok for input ,$30.00/MTok for output
# model="gpt-4o" # $5.00/MTok for input ,$15.00/MTok for output
model = "gpt-3.5-turbo-0125" #
# Get the API key from an environment variable
api_key = "your api key"
client = AsyncOpenAI(
api_key=api_key
)
stream = await client.chat.completions.create(
model=model,
stream=True,
messages=[
{"role": "system", "content": "You are a sincere and helpful assistant."},
{"role": "user", "content": "Hello"}
],
stream_options={"include_usage": True}, # Output usage (number of input/output tokens)
)
async for chunk in stream:
print(f"chunk__{chunk}")
except Exception as e:
print(f"An unexpected error occurred: {e}\n{traceback.format_exc()}")
finally:
pass
asyncio.run(main())
Example Output
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content='', function_call=None, role='assistant', tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content='Hello', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content=' there', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content='!', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content=' How', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content=' can', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content=' I', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content=' be', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content=' of', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content=' help', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content=' to', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content=' you', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content=' today', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content='?', function_call=None, role=None, tool_calls=None), finish_reason=None, index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[Choice(delta=ChoiceDelta(content=None, function_call=None, role=None, tool_calls=None), finish_reason='stop', index=0, logprobs=None)], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=None)
ChatCompletionChunk(id='chatcmpl-9iGtdyZ43HFebZV22QOaZPIzgeStA', choices=[], created=1234567890, model='gpt-3.5-turbo-0125', object='chat.completion.chunk', system_fingerprint=None, usage=CompletionUsage(completion_tokens=14, prompt_tokens=31, total_tokens=45))
Sample Code: Parsing the Streamed Chunk Contents
Let's parse each chunk and extract the various pieces of data it contains.
import asyncio
import traceback
from openai import AsyncOpenAI
async def main() -> None:
try:
model = "gpt-3.5-turbo-0125"
api_key = "your api key"
client = AsyncOpenAI(api_key=api_key)
stream = await client.chat.completions.create(
model=model,
stream=True,
messages=[
{"role": "system", "content": "You are a sincere and helpful assistant."},
{"role": "user", "content": "Hello"}
],
stream_options={"include_usage": True}, # Output usage
)
first_chunk = None
last_chunk = None
finish_reason = None
full_content = ""
role = None
created = None
model = None
completion_id = None
async for chunk in stream:
object_type = chunk.object
if object_type == "chat.completion.chunk":
if first_chunk is None:
# On the first chunk
first_chunk = chunk
model = first_chunk.model
created = first_chunk.created
completion_id = first_chunk.id
if chunk.choices:
first_choice = chunk.choices[0]
role = first_choice.delta.role
# Information available from the first chunk
print(f"completion_id: {completion_id}")
print(f"created: {created}")
# Information available only from the first chunk
print(f"model: {model}")
print(f"role: {role}")
print("streaming text: ", end="", flush=True)
last_chunk = chunk
if chunk.choices:
first_choice = chunk.choices[0]
if first_choice.delta.content:
# Text generated in this iteration
delta_str = first_choice.delta.content
print(delta_str, end="", flush=True) # Print the generated text incrementally
full_content += delta_str # Append to the full text
if finish_reason is None:
finish_reason = first_choice.finish_reason
print()
if last_chunk:
# Process the data from the last chunk
usage = last_chunk.usage
print(f"Full Content: {full_content}")
print(f"Finish Reason: {finish_reason}")
if usage:
print(f"ttl tokens: {usage.total_tokens}")
print(f"num input tokens:: {usage.prompt_tokens}")
print(f"num output tokens: {usage.completion_tokens}")
else:
print("Usage information not available")
except Exception as e:
print(f"An unexpected error occurred: {e}\n{traceback.format_exc()}")
finally:
pass
asyncio.run(main())
Execution Result
completion_id: chatcmpl-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
created: 123456789
model: gpt-3.5-turbo-0125
role: assistant
streaming text: Hello there! How can I be of help to you today?
Full Content: Hello there! How can I be of help to you today?
Finish Reason: stop
ttl tokens: 51
num input tokens:: 31
num output tokens: 20