Async programming¶
If you have a large number of text messages to process, querying the API for them one by one might take a while and most the time would be spent waiting for the responses.
Luckily, the API providers, including OpenAI, allow sending multiple queries simultaneously to the API. Note that different users might have different rate limits. This feature means that while you are waiting for the response, you can process other text messages, greatly saving your time.
Here is my own experience. I created a tool to filter the new papers on arXiv everyday to save my time. It runs the abstracts of new papers through a GPT model and has it classify them. The old implementation does this paper by paper in order, taking a few minutes to finish about 100 papers. Although this is not slow at all, I decided to give async programming a try. With the new implementation, it only takes a few seconds to process all the papers now, over x50 faster!
There are two ways to do this in Python: threading and async programming. This notebook walks through both; ready-to-use template scripts are linked at the end.
Although both approaches achieve the same goal, I personally recommend the threading approach because it is easier to implement. Specifically, once you have a sync implementation, you can easily convert it to a threading implementation without much effort. But turning the sync implementation to async requires a lot of changes. Below I provide a detailed explanation of the two approaches and you can see the differences clearly.
⚠️ Warning: Async programming and threading can be complicated, and you might run into all sorts of issues if you don't know what you are doing. My suggestion is that you should only consider this approach when you absolutely need it. If you don't have that many text messages to process, you can just use a for loop. If you are not in a rush, you should also consider the new batch API, which is much easier to handle and costs only half the price.
Setup¶
Install the dependencies and configure your API key.
On Google Colab, store your key in Colab Secrets first: click the key icon in the left sidebar, add a secret named OPENAI_API_KEY, and enable notebook access. Running locally, put the key in a .env file or in the OPENAI_API_KEY environment variable (see handling API keys).
%pip install -q openai pydantic python-dotenv
import os
from dotenv import load_dotenv
load_dotenv() # loads a .env file from the working directory, if there is one
if "OPENAI_API_KEY" not in os.environ:
try:
# On Google Colab, read the key from Colab Secrets
from google.colab import userdata
os.environ["OPENAI_API_KEY"] = userdata.get("OPENAI_API_KEY")
except ImportError:
raise RuntimeError("Set OPENAI_API_KEY in your environment or in a .env file")
Detailed explanation¶
First, let's define the text messages, the prompt, and the output schema. Assuming we have a list of text messages that we want to evaluate the sentiment of:
from pydantic import BaseModel, Field
text_messages = [
"The service here is very good!",
"The service here is good.",
"The service here is ok.",
"The service here is not very good.",
"The service here is terrible!",
]
system_prompt = "You are an expert on sentiment analysis. Your job is to evaluate the sentiment of the given text message."
user_instruction = """
Given the following text message: '{text_message}', please evaluate its sentiment by giving a score in the range of -1 to 1, where -1 means negative and 1 means positive.
Also explain why.
"""
class Sentiment(BaseModel):
score: float = Field(
ge=-1,
le=1,
description="Sentiment score in the range of -1 to 1, where -1 means negative and 1 means positive.",
)
explanation: str = Field(description="Explanation of the sentiment score.")
Sync implementation¶
Let's define a sync function to query the API:
from openai import OpenAI
client = OpenAI()
def process_text_message(text_message):
print(f"Working on text message: {text_message}")
response = client.responses.parse(
model="gpt-5.6-luna",
text_format=Sentiment,
instructions=system_prompt,
input=user_instruction.format(text_message=text_message),
)
senti_score_result = response.output_parsed
result = {
"text_message": text_message,
"chatgpt_response": senti_score_result.model_dump(),
}
return result
Running it with a for loop is straight forward (we also time it for later comparison):
import time
start_time = time.perf_counter()
sync_results = []
for text_message in text_messages:
sync_result = process_text_message(text_message)
sync_results.append(sync_result)
end_time = time.perf_counter()
print(f"Sync method done in {end_time - start_time:.2f} seconds.")
for result in sync_results:
print(result)
Working on text message: The service here is very good!
Working on text message: The service here is good.
Working on text message: The service here is ok.
Working on text message: The service here is not very good.
Working on text message: The service here is terrible!
Sync method done in 9.74 seconds.
{'text_message': 'The service here is very good!', 'chatgpt_response': {'score': 0.9, 'explanation': 'The message expresses strong positive sentiment, praising the service as “very good.”'}}
{'text_message': 'The service here is good.', 'chatgpt_response': {'score': 0.8, 'explanation': 'The message expresses a clearly positive opinion, describing the service as good. The sentiment is favorable, though not extremely enthusiastic.'}}
{'text_message': 'The service here is ok.', 'chatgpt_response': {'score': 0.1, 'explanation': 'The phrase "ok" conveys a neutral to mildly positive assessment. It indicates the service is acceptable, but not especially good or enthusiastic.'}}
{'text_message': 'The service here is not very good.', 'chatgpt_response': {'score': -0.7, 'explanation': 'The message expresses dissatisfaction with the service, describing it as not very good. This indicates a clearly negative sentiment, though the wording is relatively mild rather than strongly hostile.'}}
{'text_message': 'The service here is terrible!', 'chatgpt_response': {'score': -0.95, 'explanation': 'The message expresses strong dissatisfaction with the service, using the emphatic negative word “terrible.”'}}
Threading implementation¶
Now, let's use threading to process the text messages.
We can reuse the sync function above and run it in a thread pool from the concurrent.futures module.
Note that we define a N_THREADS variable to set the number of threads.
Increasing the number of threads will speed up the processing, but it will also increase the memory usage.
And the marginal improvement will decrease as the number of threads increases.
from concurrent.futures import ThreadPoolExecutor
N_THREADS = 3
print(f"Threading method with {N_THREADS} threads:")
start_time = time.perf_counter()
with ThreadPoolExecutor(max_workers=N_THREADS) as executor:
threading_results = list(executor.map(process_text_message, text_messages))
end_time = time.perf_counter()
print(f"Threading method done in {end_time - start_time:.2f} seconds.")
for result in threading_results:
print(result)
Threading method with 3 threads: Working on text message: The service here is very good! Working on text message: The service here is good. Working on text message: The service here is ok.
Working on text message: The service here is not very good.
Working on text message: The service here is terrible!
Threading method done in 3.66 seconds.
{'text_message': 'The service here is very good!', 'chatgpt_response': {'score': 0.9, 'explanation': 'The message expresses strong positive sentiment, praising the service as “very good.”'}}
{'text_message': 'The service here is good.', 'chatgpt_response': {'score': 0.8, 'explanation': 'The message expresses a clearly positive opinion, stating that the service is good. The sentiment is favorable, though the wording is straightforward rather than highly enthusiastic.'}}
{'text_message': 'The service here is ok.', 'chatgpt_response': {'score': 0.1, 'explanation': 'The phrase "ok" indicates a neutral to mildly positive opinion. It suggests the service is acceptable, but not especially good or enthusiastic.'}}
{'text_message': 'The service here is not very good.', 'chatgpt_response': {'score': -0.7, 'explanation': 'The message expresses dissatisfaction with the service, indicating a clearly negative sentiment, though the wording is relatively mild rather than strongly hostile.'}}
{'text_message': 'The service here is terrible!', 'chatgpt_response': {'score': -0.95, 'explanation': 'The message expresses strong dissatisfaction with the service, using the emphatic negative word “terrible.”'}}
Async implementation¶
Now, let's try the async version. First, we need to define an async function to query the API:
from openai import AsyncOpenAI
async_client = AsyncOpenAI()
async def process_text_message_async(text_message):
print(f"Working on text message: {text_message}")
response = await async_client.responses.parse(
model="gpt-5.6-luna",
text_format=Sentiment,
instructions=system_prompt,
input=user_instruction.format(text_message=text_message),
)
senti_score_result = response.output_parsed
result = {
"text_message": text_message,
"chatgpt_response": senti_score_result.model_dump(),
}
return result
The code is similar to the sync function above.
The main difference is that the async_client is an AsyncOpenAI object instead of an OpenAI object.
And we add the async keyword to the function name and the await keyword to the function call.
We can no longer use for loop to process the text messages with the async function any more. Instead, we will use asyncio.gather.
Note: Jupyter and Colab notebooks already run an event loop, so we
awaitthe coroutine directly here. In a standalone script you would useasyncio.run(async_main())instead — see the async template script.
import asyncio
async def async_main():
async_results = await asyncio.gather(
*[process_text_message_async(text_message) for text_message in text_messages]
)
return async_results
start_time = time.perf_counter()
async_results = await async_main()
end_time = time.perf_counter()
print(f"Async method done in {end_time - start_time:.2f} seconds.")
for result in async_results:
print(result)
Working on text message: The service here is very good! Working on text message: The service here is good. Working on text message: The service here is ok. Working on text message: The service here is not very good. Working on text message: The service here is terrible!
Async method done in 2.10 seconds.
{'text_message': 'The service here is very good!', 'chatgpt_response': {'score': 0.9, 'explanation': 'The message expresses strong positive sentiment, praising the service as “very good.”'}}
{'text_message': 'The service here is good.', 'chatgpt_response': {'score': 0.8, 'explanation': 'The message expresses a clearly positive opinion, stating that the service is good.'}}
{'text_message': 'The service here is ok.', 'chatgpt_response': {'score': 0.1, 'explanation': 'The message expresses a mildly positive or neutral opinion. “Ok” indicates the service is acceptable, but not particularly good or enthusiastic.'}}
{'text_message': 'The service here is not very good.', 'chatgpt_response': {'score': -0.7, 'explanation': 'The message expresses clear dissatisfaction with the quality of the service, making its sentiment moderately negative.'}}
{'text_message': 'The service here is terrible!', 'chatgpt_response': {'score': -0.95, 'explanation': 'The message expresses strong dissatisfaction with the service, using the emphatic word “terrible,” indicating a highly negative sentiment.'}}
In the run above, the async implementation finished in about 2 seconds, versus about 10 seconds for the sync version; the threading version with 3 threads took about 4 seconds. Exact timings vary between runs, and the improvement grows with the number of text messages to process.
Template scripts¶
To help you get started with your own projects, the repository provides two template scripts:
Both templates allow you to set a max number of concurrent requests to the API and a timeout for each request. They also use tqdm to show the progress of the requests. You can use the scripts as a starting point for your own implementation. You might still need to add more code to handle other errors to make your application more robust, though.