LangChain streaming callbacks with OpenAI-compatible APIs

Wiring up langchain streaming callbacks openai-compatible endpoints is mostly plumbing until you need token-level observability, provider failover, or usage accounting. This how-to builds a reusable pattern with LangChain’s ChatOpenAI client against any OpenAI-compatible gateway, including custom callback handlers that work in sync and async contexts.

Step 1: Install and pin the right packages

LangChain split its provider integrations into separate packages. Use langchain-openai for the ChatOpenAI wrapper; the core langchain package provides the callback base classes.

pip install "langchain>=0.2.0" langchain-openai openai

Pin versions in production. The openai SDK is a transitive dependency but importing it directly avoids surprises when you inspect raw response objects.

Step 2: Configure the OpenAI-compatible client

ChatOpenAI accepts base_url and api_key. Point it at your gateway instead of api.openai.com. If you use a gateway such as n4n.ai, the same base URL fronts 240+ models and applies automatic fallback when a upstream provider is degraded, so your client code stays identical across model swaps.

import os
from langchain_openai import ChatOpenAI

BASE_URL = os.environ["OPENAI_BASE_URL"]  # e.g. "https://api.n4n.ai/v1"
API_KEY = os.environ["OPENAI_API_KEY"]

def make_model(model_name: str, *, stream: bool = True) -> ChatOpenAI:
    return ChatOpenAI(
        model=model_name,
        base_url=BASE_URL,
        api_key=API_KEY,
        streaming=stream,
        temperature=0.2,
        max_tokens=1024,
    )

Set streaming=True at construction. LangChain ignores the flag if you later pass stream=True to .invoke, but being explicit prevents silent full-response waits.

Step 3: Implement a custom streaming callback handler

The built-in StreamingStdOutCallbackHandler prints tokens, but real apps need to route them to a queue, websocket, or UI framework. Subclass BaseCallbackHandler and override on_llm_new_token. The method receives the chunk string and the **kwargs bag containing the raw response object from the OpenAI SDK.

from langchain_core.callbacks import BaseCallbackHandler

class TokenAccumulator(BaseCallbackHandler):
    def __init__(self) -> None:
        self.tokens: list[str] = []
        self.usage: dict | None = None

    def on_llm_new_token(self, token: str, **kwargs) -> None:
        self.tokens.append(token)
        # token-level side effects: push to websocket, update tqdm, etc.
        print(token, end="", flush=True)

    def on_llm_end(self, response, **kwargs) -> None:
        # With streaming, usage often lands in the final response metadata
        if hasattr(response, "llm_output") and response.llm_output:
            self.usage = response.llm_output.get("token_usage")
        print(f"\n[usage] {self.usage}")

For async pipelines, subclass AsyncCallbackHandler and define on_llm_new_token / on_llm_end as async def. Never block the event loop inside the async variant.

Step 4: Wire callbacks into the model invocation

Pass the handler instance via the callbacks argument. Callbacks can be set per-call or per-model; per-call wins for isolation when multiple requests share a model.

handler = TokenAccumulator()
model = make_model("gpt-4o-mini")

# Streaming invoke
result = model.invoke(
    "Explain TCP slow start in three bullet points.",
    config={"callbacks": [handler]},
)
print(f"\n[final content] {result.content}")
print(f"[accumulated tokens] {len(handler.tokens)}")

The result.content is the full message, but tokens have already been flushed incrementally. If you only need the stream, use model.stream() which yields AIMessageChunk objects and still triggers callbacks.

for chunk in model.stream("Same question, streaming chunks:"):
    pass  # callbacks handle emission

Step 5: Handle async streaming

Async streaming requires AsyncCallbackHandler and ainvoke / astream. The pattern mirrors sync but uses await inside callbacks only when necessary.

from langchain_core.callbacks import AsyncCallbackHandler

class AsyncTokenSink(AsyncCallbackHandler):
    def __init__(self) -> None:
        self.count = 0

    async def on_llm_new_token(self, token: str, **kwargs) -> None:
        self.count += 1
        # await websocket.send(token)  # example non-blocking IO
        print(token, end="", flush=True)

async def main():
    handler = AsyncTokenSink()
    model = make_model("gpt-4o-mini")
    await model.ainvoke(
        "Write a haiku about distributed caches.",
        config={"callbacks": [handler]},
    )
    print(f"\n[async tokens seen] {handler.count}")

import asyncio
asyncio.run(main())

Use astream when you want to consume chunks in your own loop while the callback performs side effects. Both paths fire on_llm_end with the aggregated usage.

Step 6: Capture usage and respect routing directives

OpenAI-compatible streaming responses embed usage only in the final chunk. LangChain surfaces this in on_llm_end via response.llm_output["token_usage"]. If your gateway performs per-token usage metering, the numbers match what you see in billing—no client-side estimation required.

When you send provider-specific hints (e.g., cache control or routing directives), forward them through model_kwargs. LangChain passes unknown keys to the underlying SDK, and a compliant gateway honors client routing directives and forwards provider cache-control hints without extra headers.

model = ChatOpenAI(
    model="anthropic.claude-3-5-sonnet",
    base_url=BASE_URL,
    api_key=API_KEY,
    streaming=True,
    model_kwargs={
        "extra_headers": {"x-routing-prefer": "anthropic"},
        "user": "trace-123",
    },
)

This keeps your langchain streaming callbacks openai-compatible setup agnostic to which backend actually served the token.

Step 7: Verify the integration end to end

Run the sync script first. Success criteria:

1. Tokens appear on stdout incrementally, not all at once after a pause.
2. [usage] prints a dict with prompt_tokens, completion_tokens, total_tokens.
3. handler.tokens length equals the character or word count expectation (roughly completion_tokens * 4 for English).
4. Swap model to a different provider slug (e.g., a Mistral model) without changing callback code; streaming still works.

For async, confirm AsyncTokenSink.count is non-zero and the event loop exits cleanly. If you artificially block the gateway (e.g., bad API key), LangChain raises APIConnectionError before the first token; with a gateway that provides automatic fallback, a degraded primary provider should yield tokens from the secondary without client changes.

Gotchas that waste an afternoon

• Callback identity: Reusing one handler across concurrent requests mixes token streams. Instantiate per request.
• Streaming flag mismatch: Setting streaming=False but passing a streaming callback silently buffers. Check the flag.
• Async handler in sync call: LangChain will not await your async def; tokens vanish. Match handler type to call type.
• Usage missing: Some older OpenAI-compatible proxies omit usage in streaming. Patch by counting tokens locally with tiktoken as a fallback, but prefer a gateway that returns accurate metering.

The pattern above gives you token-level control, clean async support, and provider portability without rewriting application code when you switch models or backends.