run-llama/llama_index · error · NotImplementedError

astream_call is not supported by default.

Error message

astream_call is not supported by default.

What it means

The async twin of stream_call: PydanticProgram.astream_call() raises NotImplementedError by default because the base class cannot stream structured output. Subclasses must explicitly provide an async streaming implementation; most do not.

Source

Thrown at llama-index-core/llama_index/core/types.py:134

    def __call__(self, *args: Any, **kwargs: Any) -> Union[Model, List[Model]]:
        pass

    async def acall(self, *args: Any, **kwargs: Any) -> Union[Model, List[Model]]:
        return self(*args, **kwargs)

    def stream_call(
        self, *args: Any, **kwargs: Any
    ) -> Generator[
        Union[Model, List[Model], "FlexibleModel", List["FlexibleModel"]], None, None
    ]:
        raise NotImplementedError("stream_call is not supported by default.")

    async def astream_call(
        self, *args: Any, **kwargs: Any
    ) -> AsyncGenerator[
        Union[Model, List[Model], "FlexibleModel", List["FlexibleModel"]], None
    ]:
        raise NotImplementedError("astream_call is not supported by default.")


class PydanticProgramMode(str, Enum):
    """Pydantic program mode."""

    DEFAULT = "default"
    OPENAI = "openai"
    LLM = "llm"
    FUNCTION = "function"
    GUIDANCE = "guidance"
    LM_FORMAT_ENFORCER = "lm-format-enforcer"


class Thread(threading.Thread):
    """
    A wrapper for threading.Thread that copies the current context and uses the copy to run the target.
    """

View on GitHub (pinned to afd0fef371)

Solutions

1. Fall back to await program.acall(...) or program(...).
2. Use a program subclass that overrides astream_call (check the class or docs before relying on it).
3. Wrap in a capability check: if type(program).astream_call is PydanticProgram.astream_call: use acall instead.

Example fix

# before
 async for chunk in program.astream_call(input='...'):  # NotImplementedError
     print(chunk)

# after
 result = await program.acall(input='...')  # or program(...)
 print(result)

async def run_program(program, **kwargs):
    if type(program).astream_call is PydanticProgram.astream_call:
        return await program.acall(**kwargs)  # no async streaming support
    return [chunk async for chunk in program.astream_call(**kwargs)]

from llama_index.core.program import PydanticProgram

def is_astreamable_program(program) -> bool:
    return type(program).astream_call is not PydanticProgram.astream_call

try:
    async for chunk in program.astream_call(input=...):
        yield chunk
except NotImplementedError:
    yield await program.acall(input=...)

When it happens

Trigger: Awaiting/iterating program.astream_call(...) in an async agent or FastAPI handler on a program subclass that only implements __call__/acall. Common when code was written against a subclass that supported streaming and then the program type was swapped.

Common situations: Async agent pipelines (AgentWorkflow, async chat engines) that uniformly call astream_call on attached programs; migrating from OpenAIPydanticProgram to LLMTextCompletionProgram and losing streaming support silently.

Related errors

• astream_complete is not supported by default.
• stream_call is not supported by default.
• Could not parse output: {output}
• Streaming is not enabled. Please use achat() instead.
• achat_stream is None. Cannot asynchronously write to history

AI-assisted analysis of run-llama/llama_index@afd0fef371 (2026-08-15). Data as JSON: /api/errors/2e45d5a75e654256. Report an issue: GitHub.