Solving async structured output issues

API
1

Hello I am looking for guidance on how I could go about using the asynchronous client with structured outputs. Within the scope of my current project I have a bunch of sequential API calls that utilize structured outputs, to reduce latency I figured that I would run these calls asynchronously. Unfortunately, I cannot seem to get it to work.

I cannot show the exact code I am working with due to project rules, but using an example this is what I am trying to accomplish:

from pydantic import BaseModel
from openai import AsyncOpenAI
import asyncio

async_client = AsyncOpenAI()
class CalendarEvent(BaseModel):
    name: str
    date: str
    participants: list[str]

async def asyncExample():
    async_client = AsyncOpenAI()
    completion = await async_client.chat.completions.parse(
        model="gpt-4o-2024-08-06",
        messages=[
            {"role": "system", "content": "Extract the event information."},
            {"role": "user", "content": "Alice and Bob are going to a science fair on Friday."},
        ],
        response_format=CalendarEvent,
    )
    event = completion.choices[0].message.parsed
    return event

However, this has been throwing me an error saying:
AttributeError: ‘AsyncCompletions’ object has no attribute ‘parse’.

Since the interface I am creating requires multiple structured output calls in a row (that do not rely upon each other), having them run asynchronously or at the same time in some capacity would do me wonders in reducing latency. As such, I was wondering if anyone knows of a way to accomplish this?

2

Here’s your class from the library:

>>>import openai
>>>client=openai.AsyncOpenAI()
>>>resp = client.chat.completions
>>>resp.__class__

<class 'openai.resources.chat.completions.completions.AsyncCompletions'>

and your method, being existing and whatnot:


>>>resp = client.chat.completions.parse(messages=[], model="gpt-4o")
>>>resp
<coroutine object AsyncCompletions.parse at 0x000000000EF46FE0>

I’ve got a “parse”, where’s yours?

So you should check the OpenAI library version:

>>>openai.__version__
'1.100.1'

and see if it isn’t instead way old, before the introduction of structured outputs support.

Then pass the reusable client into the functions.

Here’s an example of blasting off two calls in parallel using functional programming that actually has inputs and output. The text to extract against is the varying input to the function, and you’ve got a self-describing schema also. (the title is superfluous, the AI was getting the name duplicated as title by the streamer->schema->API.)

from typing import Type
from pydantic import BaseModel, Field, ConfigDict
from openai import AsyncOpenAI
import asyncio

class CalendarEvent(BaseModel):
    model_config = ConfigDict(
        title="Calendar Event",
        description="Structured representation of a single calendar event extracted from a user message."
    )
    name: str = Field(
        ...,
        title="Event Name",
        description="Concise title of the event (e.g., 'Kickoff meeting', 'Birthday dinner')."
    )
    date: str = Field(
        ...,
        title="Event Date/Time",
        description="When the event occurs; ISO 8601 (e.g., '2025-09-05') or clear natural language (e.g., 'Friday', 'next Monday 10am')."
    )
    participants: list[str] = Field(
        ...,
        title="Participants",
        description="List of attendee names mentioned for the event."
    )

async def extract_structured(
    client: AsyncOpenAI,
    user_message: str,
    schema: Type[BaseModel],
    model: str = "gpt-4o-2024-08-06",
) -> str:
    completion = await client.chat.completions.parse(
        model=model,
        messages=[
            {"role": "system", "content": "Extract the event information."},
            {"role": "user", "content": user_message},
        ],
        response_format=schema,
    )
    parsed = completion.choices[0].message.parsed
    return parsed.model_dump_json()

async def main():
    client = AsyncOpenAI()
    msgs = [
        "Kickoff meeting for Project Atlas next Monday; attendees: Alice, Bob, Charlie.",
        "Birthday dinner for Dana on 2025-09-05 with Sam and Dana.",
    ]
    tasks = [extract_structured(client, m, CalendarEvent) for m in msgs]
    responses = await asyncio.gather(*tasks)
    print(responses)

if __name__ == "__main__":
    asyncio.run(main())

Success with proper library version support, a list of the two API calls’ JSONs, parsed (strings, the result of model_dump() instead of the methods for keys):

['{"name":"Kickoff meeting for Project Atlas","date":"next Monday","participants":["Alice","Bob","Charlie"]}', '{"name":"Birthday Dinner for Dana","date":"2025-09-05","participants":["Dana","Sam"]}']

Then handle the scenario where a validation or an API call fails.

2 Likes
3

Yep, it seems like I forgot to update my OpenAI library version. I appreciate you taking the time out of your day to answer my question.

Everything seems to be working on my end now.

1 Like
4

Great! Fun trick, pop two BaseModels in there: constructing an anyOf; one to use when the AI has not found anything good to extract.

1 Like