run-llama/llama_index · error · NotImplementedError

stream_call is not supported by default.

Error message

stream_call is not supported by default.

What it means

PydanticProgram (the base class for structured-output programs) implements __call__ and acall, but streaming a structured program is not generically possible, so the default stream_call() raises NotImplementedError. Only specific subclasses (e.g. OpenAIPydanticProgram with streaming support) override it.

Source

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

    @property
    @abstractmethod
    def output_cls(self) -> Type[Model]:
        pass

    @abstractmethod
    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"

View on GitHub (pinned to afd0fef371)

Solutions

  1. Use the non-streaming API: output = program(input=..., description=...).
  2. Switch to a subclass that supports streaming (e.g. OpenAIPydanticProgram) if you truly need incremental structured output.
  3. Feature-detect before calling: hasattr check / isinstance check against streaming-capable subclasses.

Example fix

# before
 chunks = list(program.stream_call(input='extract...'))  # NotImplementedError

# after
 result = program(input='extract...')  # non-streaming
 # or use a streaming-capable subclass:
 # from llama_index.core.program import OpenAIPydanticProgram
Defensive patterns

Strategy: fallback

Validate before calling

def supports_streaming(program) -> bool:
    # only subclasses that override stream_call can stream
    return type(program).stream_call is not PydanticProgram.stream_call

Type guard

from llama_index.core.program import PydanticProgram

def is_streamable_program(program) -> bool:
    return type(program).stream_call is not PydanticProgram.stream_call

Try / catch

try:
    chunks = list(program.stream_call(input=...))
except NotImplementedError:
    result = program(input=...)  # graceful fallback to sync call
else:
    result = merge(chunks)

Prevention

When it happens

Trigger: Calling program.stream_call(...) on a pydantic program that does not override stream_call - typical with LLMTextCompletionProgram, GuidanceProgram, or function-calling programs used through generic code that prefers streaming when available.

Common situations: Writing UI code that tries stream_call first and falls back to __call__; switching a working non-streaming program to streaming without checking the subclass supports it.

Related errors


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