How do you stream individual message chunks using AIMessageChunk and how do you aggregate them into a complete AIMessage?

Q: How do you stream individual message chunks using AIMessageChunk and how do you aggregate them into a complete AIMessage?

When streaming responses from a chat model using `.stream()` or `astream_events`, the model yields `AIMessageChunk` objects. These chunks represent partial content. To reconstruct the complete `AIMessage`, you can accumulate them by adding chunks together using the `+` operator, which is overloaded to merge content and tool call deltas appropriately. This aggregation is automatically handled when you collect all chunks into a list and sum them, or you can manually accumulate. For tool calls, chunk merging requires special handling because tool call arguments may arrive in multiple fragments; `AIMessageChunk` provides a `tool_calls` field that aggregates as chunks are added. ```python from langchain_openai import ChatOpenAI model = ChatOpenAI(model="gpt-4o", streaming=True) # Method 1: Accumulate chunks manually full_message = None for chunk in model.stream("Tell me a joke."): print(chunk.content, end="", flush=True) if full_message is None: full_message = chunk else: full_message += chunk # Method 2: Collect in list and sum chunks = list(model.stream("Tell me a joke.")) complete_message = sum(chunks[1:], chunks[0]) # Overloaded + works # For tool calls model_with_tools = model.bind_tools([get_weather]) chunks = list(model_with_tools.stream("What's the weather in Paris?")) complete = sum(chunks[1:], chunks[0]) print(complete.tool_calls) # Aggregated tool call arguments ``` When using LangGraph, you can also receive messages via `astream_events` with event types `on_chat_model_stream`. Each event contains a chunk of type `AIMessageChunk`. You can accumulate these in the same way.

Difficulty: 5/10

streaming, message aggregation, LangChain

Stream individual message chunks using `AIMessageChunk` objects that are emitted by streaming LLMs, and aggregate them by accumulating the `content` fields and merging `tool_calls` chunks using the `AIMessageChunk`'s built-in `+` operator or custom aggregation logic.

When streaming responses from a chat model using .stream() or astream_events, the model yields AIMessageChunk objects. These chunks represent partial content. To reconstruct the complete AIMessage, you can accumulate them by adding chunks together using the + operator, which is overloaded to merge content and tool call deltas appropriately. This aggregation is automatically handled when you collect all chunks into a list and sum them, or you can manually accumulate. For tool calls, chunk merging requires special handling because tool call arguments may arrive in multiple fragments; AIMessageChunk provides a tool_calls field that aggregates as chunks are added.

Streaming and Aggregating AIMessageChunks

When using LangGraph, you can also receive messages via astream_events with event types on_chat_model_stream. Each event contains a chunk of type AIMessageChunk. You can accumulate these in the same way.

Scenario Questions

0-2 years experience

2-5 years experience

How do you stream individual message chunks using AIMessageChunk and how do you aggregate them into a complete AIMessage?

Stream individual message chunks using AIMessageChunk objects that are emitted by streaming LLMs, and aggregate them by accumulating the content fields and merging tool_calls chunks using the AIMessageChunk's built-in + operator or custom aggregation logic.

Scenario Questions

Follow-up Questions

Stream individual message chunks using `AIMessageChunk` objects that are emitted by streaming LLMs, and aggregate them by accumulating the `content` fields and merging `tool_calls` chunks using the `AIMessageChunk`'s built-in `+` operator or custom aggregation logic.