LangChain streaming with Claude 3.5 Sonnet via n4n.ai

Getting langchain streaming claude 3.5 sonnet n4n.ai working requires pointing LangChain’s OpenAI-compatible chat client at the gateway and enabling token callbacks. This tutorial gives a copy-paste path from install to verified stream, including a custom handler that surfaces usage metadata. You’ll avoid the common pitfalls around stdout buffering and async event loops.

Step 1: Install the correct LangChain packages

LangChain split its provider integrations out of the core package in 2024. Use langchain-openai for the ChatOpenAI class and langchain-core for callback primitives.

pip install langchain-openai==0.1.7 langchain-core==0.2.0 python-dotenv

Pin versions in production. The streaming contract has changed across minor releases; unpinned installs will break your callback code silently.

Step 2: Configure the ChatOpenAI client for Claude 3.5 Sonnet

Set your gateway API key as OPENAI_API_KEY. The model name must match what the gateway expects; for Claude 3.5 Sonnet that is typically claude-3-5-sonnet or a dated snapshot. Point base_url at the n4n.ai OpenAI-compatible endpoint so the request hits the right backend.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

llm = ChatOpenAI(
    model="claude-3-5-sonnet",
    base_url="https://api.n4n.ai/v1",
    api_key=os.environ["OPENAI_API_KEY"],
    streaming=True,
    temperature=0.2,
    max_tokens=1024,
)

Claude 3.5 Sonnet has a large context window, but the gateway may cap max_tokens per request. Set it conservatively to avoid truncated streams. Temperature near zero reduces variance for eval harnesses. If you omit streaming=True, LangChain buffers the full response and your callbacks fire once. That defeats the purpose of langchain streaming claude 3.5 sonnet n4n.ai.

Step 3: Implement a streaming callback handler

The default StreamingStdOutCallbackHandler prints tokens but gives you no hook for accumulation or usage. Subclass BaseCallbackHandler to capture tokens and react on completion.

from langchain_core.callbacks import BaseCallbackHandler

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

    def on_llm_new_token(self, token: str, **kwargs) -> None:
        self.tokens.append(token)
        print(token, end="", flush=True)

    def on_llm_end(self, response, **kwargs) -> None:
        # response holds the final LLMResult with metadata
        self.usage = response.llm_output.get("usage") if response.llm_output else None
        print("\n--- stream closed ---")

The flush=True is mandatory on some platforms. Without it, Python buffers stdout and you see nothing until the process exits. If you run concurrent requests, instantiate a new handler per call. The handler is not thread-safe by default; storing tokens on a shared list will interleave outputs.

Step 4: Run a synchronous streaming invocation

Use invoke with a config carrying your handler. LangChain still returns the full AIMessage, but tokens arrive incrementally through the callback.

handler = TokenAccumulator()
msg = llm.invoke(
    "Explain the difference between TCP and UDP in three sentences.",
    config={"callbacks": [handler]},
)
print("\nFinal content length:", len(msg.content))
print("Accumulated tokens:", len(handler.tokens))

For tighter control, use the stream generator:

for chunk in llm.stream("List three uses for a bloom filter."):
    print(chunk.content, end="", flush=True)

The generator yields AIMessageChunk objects. Concatenate .content raw; do not strip whitespace until the end, or you will corrupt token boundaries.

Step 5: Run async streaming without blocking

Many web frameworks (FastAPI, Starlette) need astream. Wrap the call in an async function and use async for.

import asyncio

async def main():
    handler = TokenAccumulator()
    async for chunk in llm.astream(
        "Write a haiku about distributed locks.",
        config={"callbacks": [handler]},
    ):
        print(chunk.content, end="", flush=True)
    print(f"\nUsage: {handler.usage}")

asyncio.run(main())

Do not mix invoke and astream on the same client inside a running event loop. Recreate the ChatOpenAI instance per request if you see RuntimeError: Event loop is closed.

To plug this into FastAPI:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/ask")
async def ask(q: str):
    async def event_gen():
        async for chunk in llm.astream(q):
            yield chunk.content
    return StreamingResponse(event_gen(), media_type="text/plain")

This keeps the HTTP connection open and tokens flowing without blocking the worker.

Step 6: Capture per-token usage metadata

The gateway returns OpenAI-style usage in the final chunk. LangChain exposes it via response_metadata on the AIMessage or inside llm_output in the callback.

# after invoke
print(msg.response_metadata.get("usage"))
# after astream, handler.usage is populated by on_llm_end
print(handler.usage)

Expect a dict with prompt_tokens, completion_tokens, and total_tokens. Per-token metering means you pay only for delivered tokens. If a stream disconnects mid-way, usage reflects partial completion. Log the usage dict for audit before shipping to billing.

Step 7: Verify success

A correct setup shows these behaviors:

1. Tokens print one by one with no multi-second pause before first character.
2. handler.tokens length is greater than 1 and roughly equals completion_tokens.
3. usage is non-null and total_tokens == prompt_tokens + completion_tokens.
4. Async runs complete without event-loop errors.

If tokens arrive in one blob, check streaming=True and flush=True. If usage is None, inspect msg.response_metadata raw to confirm the gateway returned it; some model snapshots omit usage on streaming.

Step 8: Forward cache and routing hints

Claude 3.5 Sonnet supports prompt caching via cache-control headers. The OpenAI-compatible endpoint forwards provider cache-control hints when passed as extra_headers.

llm.invoke(
    "Summarize the attached RFC.",
    extra_headers={"anthropic-cache-control": "ephemeral"},
)

Client routing directives (e.g., x-routing-key) are honored identically. This matters when you run langchain streaming claude 3.5 sonnet n4n.ai across multiple regions and want sticky routing. If you omit routing headers, the gateway may apply automatic fallback when a provider is rate-limited or degraded—useful for uptime, but it can shift latency characteristics.

Common failure modes

• Wrong model string: Gateway returns 404 if claude-3-5-sonnet isn’t registered. Use the exact id from the gateway’s model list.
• Callback not attached: Forgetting config={"callbacks": [...]} silently disables streaming output.
• Sync client in async context: Use astream or you’ll block the loop.
• Double streaming: Setting both streaming=True and using stream is fine, but wrapping invoke in asyncio.run from within an async frame crashes.

Minimal end-to-end script

import os
from langchain_openai import ChatOpenAI
from langchain_core.callbacks import BaseCallbackHandler

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

llm = ChatOpenAI(
    model="claude-3-5-sonnet",
    base_url="https://api.n4n.ai/v1",
    api_key=os.environ["OPENAI_API_KEY"],
    streaming=True,
)

llm.invoke(
    "Describe backpressure in stream processing.",
    config={"callbacks": [PrintHandler()]},
)

Run it with python script.py. You should see the explanation token-by-token, then a clean exit. That is the entire langchain streaming claude 3.5 sonnet n4n.ai loop, ready to drop into a FastAPI route or CLI tool.