Skip to content

Multi-Task and Streaming

Using an Iterable lets you extract multiple structured objects from a single LLM call, streaming them as they arrive. This is useful for entity extraction, multi-task outputs, and more.

We recommend using the create_iterable method for most use cases. It's simpler and less error-prone than manually specifying Iterable[...] and stream=True.

Here's a simple example showing how to extract multiple users from a single sentence. You can use either the recommended create_iterable method or the create method with Iterable[User]:

import instructor
from pydantic import BaseModel

client = instructor.from_provider("openai/gpt-4.1-mini")


class User(BaseModel):
    name: str
    age: int


resp = client.create_iterable(
    messages=[
        {
            "role": "user",
            "content": "Ivan is 28, lives in Moscow and his friends are Alex, John and Mary who are 25, 30 and 27 respectively",
        }
    ],
    response_model=User,
)

for user in resp:
    print(user)
    #> name='Ivan' age=28
    #> name='Alex' age=25
    #> name='John' age=30
    #> name='Mary' age=27
Recommended for most use cases. Handles streaming and iteration for you.

import instructor
from pydantic import BaseModel
from typing import Iterable

client = instructor.from_provider("openai/gpt-4.1-mini")


class User(BaseModel):
    name: str
    age: int


resp = client.create(
    messages=[
        {
            "role": "user",
            "content": "Ivan is 28, lives in Moscow and his friends are Alex, John and Mary who are 25, 30 and 27 respectively",
        }
    ],
    response_model=Iterable[User],
)

for user in resp:
    print(user)
    #> name='Ivan' age=28
    #> name='Alex' age=25
    #> name='John' age=30
    #> name='Mary' age=27
Use this if you need more manual control or compatibility with legacy code.

We also support more complex extraction patterns such as Unions as you'll see below out of the box.

Warning

Unions don't work with Gemini because the AnyOf is not supported in the current response schema.

Synchronous Usage

import instructor
from typing import Iterable, Union, Literal
from pydantic import BaseModel


class Weather(BaseModel):
    location: str
    units: Literal["imperial", "metric"]


class GoogleSearch(BaseModel):
    query: str


client = instructor.from_provider("openai/gpt-4.1-mini", mode=instructor.Mode.TOOLS)

results = client.create(
    messages=[
        {"role": "system", "content": "You must always use tools"},
        {
            "role": "user",
            "content": "What is the weather in toronto and dallas and who won the super bowl?",
        },
    ],
    response_model=Iterable[Union[Weather, GoogleSearch]],
    stream=True,
)

for item in results:
    print(item)
    #> location='Toronto' units='metric'
    #> location='Dallas' units='imperial'
    #> query='Super Bowl winner'

import instructor
from typing import Union, Literal
from pydantic import BaseModel


class Weather(BaseModel):
    location: str
    units: Literal["imperial", "metric"]


class GoogleSearch(BaseModel):
    query: str


client = instructor.from_provider("openai/gpt-4.1-mini", mode=instructor.Mode.TOOLS)

results = client.create_iterable(
    messages=[
        {"role": "system", "content": "You must always use tools"},
        {
            "role": "user",
            "content": "What is the weather in toronto and dallas and who won the super bowl?",
        },
    ],
    response_model=Union[Weather, GoogleSearch],
)

for item in results:
    print(item)
    #> location='Toronto' units='metric'
    #> location='Dallas' units='imperial'
    #> query='Super Bowl winner'

Stopping Iteration Early

Once iteration has started, close the iterable if you stop before consuming all results. Put synchronous iteration in a try/finally block and call resp.close() in finally. For asynchronous iteration, use await resp.aclose() instead. A break statement alone does not close a Python generator.

The shared iterable parser releases a closable upstream source when parsing ends, including validation failures. OpenAI chat-completion and Responses modes use this path. This cleanup does not cover partial-model streaming or providers with their own iterable parsers.

See Also

Asynchronous Usage

import instructor
from typing import Iterable, Union, Literal
from pydantic import BaseModel
import asyncio


class Weather(BaseModel):
    location: str
    units: Literal["imperial", "metric"]


class GoogleSearch(BaseModel):
    query: str


aclient = instructor.from_provider(
    "openai/gpt-4.1-mini", async_client=True, mode=instructor.Mode.TOOLS
)


async def main():
    results = await aclient.create(
        messages=[
            {"role": "system", "content": "You must always use tools"},
            {
                "role": "user",
                "content": "What is the weather in toronto and dallas and who won the super bowl?",
            },
        ],
        response_model=Iterable[Union[Weather, GoogleSearch]],
        stream=True,
    )
    async for item in results:
        print(item)
        #> location='Toronto' units='metric'
        #> location='Dallas' units='imperial'
        #> query='Super Bowl winner'


asyncio.run(main())

import asyncio
from typing import Literal, Union

import instructor
from pydantic import BaseModel


class Weather(BaseModel):
    location: str
    units: Literal["imperial", "metric"]


class GoogleSearch(BaseModel):
    query: str


aclient = instructor.from_provider(
    "openai/gpt-4.1-mini", async_client=True, mode=instructor.Mode.TOOLS
)


async def iter_results():
    async for item in aclient.create_iterable(
        messages=[
            {"role": "system", "content": "You must always use tools"},
            {
                "role": "user",
                "content": "What is the weather in toronto and dallas and who won the super bowl?",
            },
        ],
        response_model=Union[Weather, GoogleSearch],
    ):
        yield item


async def main():
    async for item in iter_results():
        print(item)
        #> location='Toronto' units='metric'
        #> location='Dallas' units='imperial'
        #> query='Super Bowl winner'


asyncio.run(main())

Markdown output and multiple tool calls

Native tool streams yield complete items as they arrive. Interleaved OpenAI tool calls are assembled by call index. In Markdown mode, a fenced JSON block takes precedence over JSON quoted in the surrounding prose. Fenced items can be emitted immediately. When the response has no fence, Instructor emits all bare JSON candidates in order at the end of the stream.

Code that calls extract_json_from_stream or its async counterpart directly with a known bare JSON stream can pass prefer_fenced=False to consume each completed value immediately.