Synchronous per-message streaming object for a single LLM response.

Returned by BaseChatModel.stream_events(version="v3"). Content-block protocol events are fed into this object and accumulated into typed projections.

Projections (always return the same cached object):

.text — iterable of str deltas; str() for full text
.reasoning — same as .text for reasoning content
.tool_calls — iterable of ToolCallChunk deltas; .get() returns list[ToolCall]
.output — blocking property, returns assembled AIMessage

Usage info is available on .output.usage_metadata once the stream has finished.

Output shape is always v1 content blocks

.output.content is always a list of v1 protocol blocks (text, reasoning, tool_call, image, …), regardless of the underlying model's output_version setting. That attribute only controls the legacy stream() / astream() / invoke() paths; ChatModelStream is built on the content-block protocol and emits v1 shapes by construction.

Raw event iteration::

for event in stream:
    print(event)  # MessagesData dicts

class

RunnableBinding

Wrap a Runnable with additional functionality.

A RunnableBinding can be thought of as a "runnable decorator" that preserves the essential features of Runnable; i.e., batching, streaming, and async support, while adding additional functionality.

Any class that inherits from Runnable can be bound to a RunnableBinding. Runnables expose a standard set of methods for creating RunnableBindings or sub-classes of RunnableBindings (e.g., RunnableRetry, RunnableWithFallbacks) that add additional functionality.

These methods include:

bind: Bind kwargs to pass to the underlying Runnable when running it.
with_config: Bind config to pass to the underlying Runnable when running it.
with_listeners: Bind lifecycle listeners to the underlying Runnable.
with_types: Override the input and output types of the underlying Runnable.
with_retry: Bind a retry policy to the underlying Runnable.
with_fallbacks: Bind a fallback policy to the underlying Runnable.

Example: bind: Bind kwargs to pass to the underlying Runnable when running it.

# Create a Runnable binding that invokes the chat model with the
# additional kwarg `stop=['-']` when running it.
from langchain_openai import ChatOpenAI

model = ChatOpenAI()
model.invoke('Say "Parrot-MAGIC"', stop=["-"])  # Should return `Parrot`
# Using it the easy way via `bind` method which returns a new
# RunnableBinding
runnable_binding = model.bind(stop=["-"])
runnable_binding.invoke('Say "Parrot-MAGIC"')  # Should return `Parrot`

Can also be done by instantiating a RunnableBinding directly (not recommended):

from langchain_core.runnables import RunnableBinding

runnable_binding = RunnableBinding(
    bound=model,
    kwargs={"stop": ["-"]},  # <-- Note the additional kwargs
)
runnable_binding.invoke('Say "Parrot-MAGIC"')  # Should return `Parrot`

class

Runnable

A unit of work that can be invoked, batched, streamed, transformed and composed.

Key Methods

invoke/ainvoke: Transforms a single input into an output.
batch/abatch: Efficiently transforms multiple inputs into outputs.
stream/astream: Streams output from a single input as it's produced.
astream_log: Streams output and selected intermediate results from an input.

Built-in optimizations:

Batch: By default, batch runs invoke() in parallel using a thread pool executor. Override to optimize batching.
Async: Methods with 'a' prefix are asynchronous. By default, they execute the sync counterpart using asyncio's thread pool. Override for native async.

All methods accept an optional config argument, which can be used to configure execution, add tags and metadata for tracing and debugging etc.

Runnables expose schematic information about their input, output and config via the input_schema property, the output_schema property and config_schema method.

Composition

Runnable objects can be composed together to create chains in a declarative way.

Any chain constructed this way will automatically have sync, async, batch, and streaming support.

The main composition primitives are RunnableSequence and RunnableParallel.

RunnableSequence invokes a series of runnables sequentially, with one Runnable's output serving as the next's input. Construct using the | operator or by passing a list of runnables to RunnableSequence.

RunnableParallel invokes runnables concurrently, providing the same input to each. Construct it using a dict literal within a sequence or by passing a dict to RunnableParallel.

For example,

from langchain_core.runnables import RunnableLambda

# A RunnableSequence constructed using the `|` operator
sequence = RunnableLambda(lambda x: x + 1) | RunnableLambda(lambda x: x * 2)
sequence.invoke(1)  # 4
sequence.batch([1, 2, 3])  # [4, 6, 8]

# A sequence that contains a RunnableParallel constructed using a dict literal
sequence = RunnableLambda(lambda x: x + 1) | {
    "mul_2": RunnableLambda(lambda x: x * 2),
    "mul_5": RunnableLambda(lambda x: x * 5),
}
sequence.invoke(1)  # {'mul_2': 4, 'mul_5': 10}

Standard Methods

All Runnables expose additional methods that can be used to modify their behavior (e.g., add a retry policy, add lifecycle listeners, make them configurable, etc.).

These methods will work on any Runnable, including Runnable chains constructed by composing other Runnables. See the individual methods for details.

For example,

from langchain_core.runnables import RunnableLambda

import random

def add_one(x: int) -> int:
    return x + 1

def buggy_double(y: int) -> int:
    """Buggy code that will fail 70% of the time"""
    if random.random() > 0.3:
        print('This code failed, and will probably be retried!')  # noqa: T201
        raise ValueError('Triggered buggy code')
    return y * 2

sequence = (
    RunnableLambda(add_one) |
    RunnableLambda(buggy_double).with_retry( # Retry on failure
        stop_after_attempt=10,
        wait_exponential_jitter=False
    )
)

print(sequence.input_schema.model_json_schema()) # Show inferred input schema
print(sequence.output_schema.model_json_schema()) # Show inferred output schema
print(sequence.invoke(2)) # invoke the sequence (note the retry above!!)

Debugging and tracing

As the chains get longer, it can be useful to be able to see intermediate results to debug and trace the chain.

You can set the global debug flag to True to enable debug output for all chains:

from langchain_core.globals import set_debug

set_debug(True)

Alternatively, you can pass existing or custom callbacks to any given chain:

from langchain_core.tracers import ConsoleCallbackHandler

chain.invoke(..., config={"callbacks": [ConsoleCallbackHandler()]})

For a UI (and much more) checkout LangSmith.

class

RunnableConfig

Configuration for a Runnable.

Note

Custom values

The TypedDict has total=False set intentionally to:

Allow partial configs to be created and merged together via merge_configs
Support config propagation from parent to child runnables via var_child_runnable_config (a ContextVar that automatically passes config down the call stack without explicit parameter passing), where configs are merged rather than replaced

Example

# Parent sets tags
chain.invoke(input, config={"tags": ["parent"]})
# Child automatically inherits and can add:
# ensure_config({"tags": ["child"]}) -> {"tags": ["parent", "child"]}

module

types

Standard, multimodal content blocks for Large Language Model I/O.

This module provides standardized data structures for representing inputs to and outputs from LLMs. The core abstraction is the Content Block, a TypedDict.

Rationale

Different LLM providers use distinct and incompatible API schemas. This module provides a unified, provider-agnostic format to facilitate these interactions. A message to or from a model is simply a list of content blocks, allowing for the natural interleaving of text, images, and other content in a single ordered sequence.

An adapter for a specific provider is responsible for translating this standard list of blocks into the format required by its API.

Extensibility

Data not yet mapped to a standard block may be represented using the NonStandardContentBlock, which allows for provider-specific data to be included without losing the benefits of type checking and validation.

Furthermore, provider-specific fields within a standard block are fully supported by default in the extras field of each block. This allows for additional metadata to be included without breaking the standard structure. For example, Google's thought signature:

AIMessage(
    content=[
        {
            "type": "text",
            "text": "J'adore la programmation.",
            "extras": {"signature": "EpoWCpc..."},  # Thought signature
        }
    ], ...
)

Note

Following widespread adoption of PEP 728, we intend to add extra_items=Any as a param to Content Blocks. This will signify to type checkers that additional provider-specific fields are allowed outside of the extras field, and that will become the new standard approach to adding provider-specific metadata.

Note

Example with PEP 728 provider-specific fields:

# Content block definition
# NOTE: `extra_items=Any`
class TextContentBlock(TypedDict, extra_items=Any):
    type: Literal["text"]
    id: NotRequired[str]
    text: str
    annotations: NotRequired[list[Annotation]]
    index: NotRequired[int]

from langchain_core.messages.content import TextContentBlock

# Create a text content block with provider-specific fields
my_block: TextContentBlock = {
    # Add required fields
    "type": "text",
    "text": "Hello, world!",
    # Additional fields not specified in the TypedDict
    # These are valid with PEP 728 and are typed as Any
    "openai_metadata": {"model": "gpt-4", "temperature": 0.7},
    "anthropic_usage": {"input_tokens": 10, "output_tokens": 20},
    "custom_field": "any value",
}

# Mutating an existing block to add provider-specific fields
openai_data = my_block["openai_metadata"]  # Type: Any

Example Usage

# Direct construction
from langchain_core.messages.content import TextContentBlock, ImageContentBlock

multimodal_message: AIMessage(
    content_blocks=[
        TextContentBlock(type="text", text="What is shown in this image?"),
        ImageContentBlock(
            type="image",
            url="https://www.langchain.com/images/brand/langchain_logo_text_w_white.png",
            mime_type="image/png",
        ),
    ]
)

# Using factories
from langchain_core.messages.content import create_text_block, create_image_block

multimodal_message: AIMessage(
    content=[
        create_text_block("What is shown in this image?"),
        create_image_block(
            url="https://www.langchain.com/images/brand/langchain_logo_text_w_white.png",
            mime_type="image/png",
        ),
    ]
)

Factory functions offer benefits such as:

Automatic ID generation (when not provided)
No need to manually specify the type field

LangChain Assistant

Menu

chat_models

Attributes

Functions

Classes

Type Aliases

Modules

Key Methods

Composition

Standard Methods

Debugging and tracing