From raw OpenAI streaming to LangChain's streaming API

Migrating from raw OpenAI streaming to LangChain’s streaming API is a common refactor when your LLM logic outgrows a single script. The raw SDK gives you control; LangChain gives you composability, callbacks, and a consistent interface across providers. This guide walks through the migration end to end, preserving token-by-token streaming behavior while adopting LangChain’s abstractions.

Step 1: Understand the raw streaming pattern you’re replacing

Before adding dependencies, isolate what your current code actually does. A typical raw OpenAI streaming loop looks like this:

from openai import OpenAI

client = OpenAI()

def stream_completion(prompt: str):
    stream = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )
    for chunk in stream:
        delta = chunk.choices[0].delta
        if delta.content:
            yield delta.content

This yields strings as they arrive. The caller consumes them via for token in stream_completion("..."): print(token, end="", flush=True). Key behaviors to preserve: incremental yields, no buffering, and graceful handling of the final chunk.choices[0].finish_reason.

Step 2: Add LangChain and the OpenAI integration

Install the minimal packages. You need langchain-core for the base abstractions and langchain-openai for the ChatOpenAI implementation.

pip install langchain-core langchain-openai

Avoid installing the meta langchain package unless you need the legacy chains and agents. The core + integration pattern keeps your dependency graph smaller and upgrades cleaner.

Step 3: Replace the client with ChatOpenAI

LangChain’s ChatOpenAI wraps the same HTTP calls but exposes a .stream() method that returns an iterator of AIMessageChunk objects. The migration is nearly one-to-one:

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4o-mini",
    streaming=True,          # enables .stream()
    temperature=0,
)

def stream_completion(prompt: str):
    for chunk in llm.stream(prompt):
        if chunk.content:
            yield chunk.content

The streaming=True flag is required; without it, .stream() falls back to a single blocking call. The chunk.content field mirrors delta.content from the raw SDK.

Step 4: Handle message history and system prompts

Raw SDK code often builds message lists manually. LangChain expects a sequence of BaseMessage objects. Convert your existing list:

from langchain_core.messages import HumanMessage, SystemMessage, AIMessage

messages = [
    SystemMessage(content="You are a concise assistant."),
    HumanMessage(content="Explain streaming in two sentences."),
]

for chunk in llm.stream(messages):
    if chunk.content:
        print(chunk.content, end="", flush=True)

If you already have a list of dicts ([{"role": "user", "content": "..."}]), use convert_to_messages from langchain_core.messages to avoid rewriting your data layer.

Step 5: Adopt the callback system for side effects

The real payoff of the openai streaming to langchain streaming migration is callbacks. Instead of sprinkling print() or websocket.send() inside your generator, attach a handler:

from langchain_core.callbacks import BaseCallbackHandler
from typing import Any

class TokenPrinter(BaseCallbackHandler):
    def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
        print(token, end="", flush=True)

llm = ChatOpenAI(
    model="gpt-4o-mini",
    streaming=True,
    callbacks=[TokenPrinter()],
)

# Now the call site is clean
llm.invoke([HumanMessage(content="Hello")])

on_llm_new_token fires for every token. Other useful hooks: on_llm_start, on_llm_end, on_llm_error. This decouples streaming logic from transport — useful when you later swap stdout for a WebSocket, Server-Sent Events, or a queue.

Step 6: Preserve tool calling and structured output

If your raw code uses tools or response_format, the migration stays similar. Define tools as LangChain BaseTool subclasses or use the @tool decorator, then bind them:

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"{city}: 72°F, sunny"

llm = ChatOpenAI(model="gpt-4o-mini", streaming=True)
llm_with_tools = llm.bind_tools([get_weather])

for chunk in llm_with_tools.stream([HumanMessage(content="Weather in Tokyo?")]):
    if chunk.content:
        print(chunk.content, end="", flush=True)
    if chunk.tool_calls:
        print(f"\n[tool call: {chunk.tool_calls}]")

Tool call chunks arrive as AIMessageChunk with tool_calls populated. The streaming contract is identical — you still iterate and yield incrementally.

Step 7: Configure timeouts, retries, and provider fallback

Production code needs resilience. LangChain delegates HTTP concerns to the underlying openai client, but you can pass client-level options via model_kwargs or a custom openai.Client:

import httpx
from openai import OpenAI
from langchain_openai import ChatOpenAI

http_client = httpx.Client(timeout=httpx.Timeout(connect=5.0, read=30.0))
openai_client = OpenAI(http_client=http_client, max_retries=2)

llm = ChatOpenAI(
    model="gpt-4o-mini",
    streaming=True,
    openai_client=openai_client,
)

If you route through a gateway that supports automatic fallback (for example, n4n.ai honors x-provider routing directives and forwards provider cache-control hints), configure the base URL on the OpenAI client instead of changing LangChain code:

openai_client = OpenAI(
    base_url="https://api.n4n.ai/v1",
    api_key="YOUR_GATEWAY_KEY",
    http_client=http_client,
)

The rest of your streaming logic remains untouched.

Step 8: Verify streaming behavior end to end

Write a small verification script that exercises the full path: token emission, callback firing, and final message assembly.

# verify_streaming.py
import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain_core.callbacks import BaseCallbackHandler
from typing import Any, List

class CaptureHandler(BaseCallbackHandler):
    def __init__(self):
        self.tokens: List[str] = []
    
    def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
        self.tokens.append(token)

async def main():
    handler = CaptureHandler()
    llm = ChatOpenAI(
        model="gpt-4o-mini",
        streaming=True,
        callbacks=[handler],
        temperature=0,
    )
    
    # Streaming invoke
    result = await llm.ainvoke([HumanMessage(content="Count to 5")])
    
    print("Callback tokens:", "".join(handler.tokens))
    print("Final result:  ", result.content)
    assert "".join(handler.tokens) == result.content, "Stream mismatch"

if __name__ == "__main__":
    asyncio.run(main())

Run it: python verify_streaming.py. Success criteria: the callback token concatenation matches result.content exactly, and tokens appear in real time (you’ll see them print incrementally if you add print(token, end="", flush=True) in the handler).

Step 9: Migrate async call sites

If your application uses asyncio, swap .stream() for .astream() and .invoke() for .ainvoke(). The callback interface stays the same:

async def stream_async(prompt: str):
    async for chunk in llm.astream(prompt):
        if chunk.content:
            yield chunk.content

LangChain’s async implementation uses the OpenAI client’s native async methods under the hood, so backpressure and cancellation behave correctly.

Step 10: Clean up the old raw SDK code

Once verification passes, remove the raw openai import and any manual SSE parsing. Keep the openai package only if you still use it for non-chat endpoints (embeddings, moderations, fine-tuning). Your requirements.txt should now reflect:

langchain-core>=0.2.0
langchain-openai>=0.1.0
openai>=1.30.0   # only if used elsewhere

Run your test suite. Pay attention to any integration tests that asserted on raw chunk structure — they’ll need updating to assert on AIMessageChunk fields instead.

Common pitfalls

Forgetting streaming=True. Without it, .stream() returns a single chunk containing the full response. The callback on_llm_new_token never fires.

Mixing sync and async. Calling .stream() inside an async function blocks the event loop. Use .astream() consistently in async contexts.

Dropping finish_reason. The final chunk includes response_metadata with finish_reason. If your old code branched on stop vs length, migrate that logic:

for chunk in llm.stream(messages):
    if chunk.content:
        yield chunk.content
    if chunk.response_metadata.get("finish_reason") == "length":
        logger.warning("Response truncated by max_tokens")

Callback ordering. on_llm_new_token fires before the chunk is yielded from .stream(). If you need strict ordering between callback side effects and your generator consumer, use a single consumer (either the callback or the loop, not both) for token processing.

What you gain

After this migration, your streaming code is provider-agnostic at the call site. Swapping ChatOpenAI for ChatAnthropic or ChatVertexAI changes only the import and instantiation. The callback handlers, tool bindings, and streaming loops stay identical. That’s the point of the abstraction — not to hide the HTTP, but to make the streaming contract portable.