Implementation:Langchain ai Langchain BaseChatModel Stream

Knowledge Sources	LangChain
Domains	Streaming, NLP
Last Updated	2026-02-11 00:00 GMT

Overview

Concrete tool for streaming chat model responses as iterators of AIMessageChunk objects provided by langchain-core.

Description

The BaseChatModel.stream() method converts input, sets up callback managers, checks if streaming is available (via _should_stream()), and either delegates to the provider's _stream() method or falls back to invoke() yielding the full response as a single chunk. The astream() method is the async equivalent.

Usage

Call stream() or astream() on any chat model that implements _stream() or _astream().

Code Reference

Source Location

• Repository: langchain
• File: libs/core/langchain_core/language_models/chat_models.py
• Lines: L480-560 (stream), L562-654 (astream)

Signature

@override
def stream(
    self,
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> Iterator[AIMessageChunk]:

@override
async def astream(
    self,
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> AsyncIterator[AIMessageChunk]:

Import

from langchain_openai import ChatOpenAI
from langchain_core.messages import AIMessageChunk

I/O Contract

Inputs

Name	Type	Required	Description
input	LanguageModelInput	Yes	User input (str, messages, or PromptValue)
config	RunnableConfig or None	No	Run configuration with callbacks
stop	list[str] or None	No	Stop sequences

Outputs

Name	Type	Description
return (sync)	Iterator[AIMessageChunk]	Incremental chunks of the response
return (async)	AsyncIterator[AIMessageChunk]	Async incremental chunks

Usage Examples

Sync Streaming

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

for chunk in llm.stream("Write a haiku about coding"):
    print(chunk.content, end="", flush=True)

Async Streaming

import asyncio
from langchain_openai import ChatOpenAI

async def main():
    llm = ChatOpenAI(model="gpt-4o-mini")
    async for chunk in llm.astream("Write a haiku"):
        print(chunk.content, end="", flush=True)

asyncio.run(main())

Related Pages

Implements Principle

• Principle:Langchain_ai_Langchain_Streaming_Method_Invocation