StreamingModel Class — langchain Architecture

Architecture documentation for the StreamingModel class in test_base.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  d868244d_12b4_d836_9e7c_9800c66d6934["StreamingModel"]
  22e0bcf0_7794_65ab_0d0e_c3696c97310d["NoStreamingModel"]
  d868244d_12b4_d836_9e7c_9800c66d6934 -->|extends| 22e0bcf0_7794_65ab_0d0e_c3696c97310d
  8830054d_ac1e_daa9_c6c5_ff55b10d0bf3["test_base.py"]
  d868244d_12b4_d836_9e7c_9800c66d6934 -->|defined in| 8830054d_ac1e_daa9_c6c5_ff55b10d0bf3
  8ad312e7_e338_bc74_5d4b_c04756215bc6["_stream()"]
  d868244d_12b4_d836_9e7c_9800c66d6934 -->|method| 8ad312e7_e338_bc74_5d4b_c04756215bc6

Relationship Graph

Source Code

libs/core/tests/unit_tests/language_models/chat_models/test_base.py lines 368–379

class StreamingModel(NoStreamingModel):
    streaming: bool = False

    @override
    def _stream(
        self,
        messages: list[BaseMessage],
        stop: list[str] | None = None,
        run_manager: CallbackManagerForLLMRun | None = None,
        **kwargs: Any,
    ) -> Iterator[ChatGenerationChunk]:
        yield ChatGenerationChunk(message=AIMessageChunk(content="stream"))