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
- Use the non-streaming API: output = program(input=..., description=...).
- Switch to a subclass that supports streaming (e.g. OpenAIPydanticProgram) if you truly need incremental structured output.
- 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
- Feature-detect streaming support before calling stream_call.
- Design call sites with a non-streaming fallback path.
- Pin and test the exact program subclass you rely on for streaming.
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
- astream_call is not supported by default.
- Output parser is not supported for streaming.
- stream_complete is not supported by default.
- astream_complete is not supported by default.
- Could not extract final answer from input text: {input_text}
AI-assisted analysis of run-llama/llama_index@afd0fef371 (2026-08-15).
Data as JSON: /api/errors/f016515a79895a4e.
Report an issue: GitHub.