LangChain plus n4n.ai: your first chat completion call

This langchain n4n.ai chat completion tutorial gets you from an empty directory to a working streaming chat call in minutes. We’ll use LangChain’s ChatOpenAI wrapper pointed at an OpenAI-compatible endpoint so you can route to hundreds of models without rewriting your client code.

Prerequisites

• Python 3.10 or newer
• pip and a virtual environment
• An API key from the gateway (exposed as N4N_API_KEY)
• Basic familiarity with Python and environment variables

If you don’t have a key yet, create one in the dashboard and export it locally:

export N4N_API_KEY="sk-..."

Do not commit this value. Use a .env file or your shell profile.

Step 1: Install the dependencies

LangChain split its provider packages in 2024. For any OpenAI-compatible server you need langchain-openai and langchain-core.

pip install langchain-openai langchain-core python-dotenv

Create a .env file to keep the key out of source control:

# .env
N4N_API_KEY=sk-your-key-here

Step 2: Your first chat completion

The fastest path is a single invoke call. ChatOpenAI accepts a base_url parameter, which is all you need to repoint the client from OpenAI to the gateway.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

llm = ChatOpenAI(
    model="openai/gpt-4o-mini",
    base_url="https://api.n4n.ai/v1",
    api_key=os.environ["N4N_API_KEY"],
    temperature=0,
)

response = llm.invoke("Return a JSON object with a single field 'hello'.")
print(response.content)

Expected output (formatted for readability):

{
  "hello": "world"
}

The model string openai/gpt-4o-mini is routed through the gateway to the correct upstream. You can swap it for anthropic/claude-3-haiku or any other supported identifier without changing the surrounding code. The temperature=0 makes the call deterministic, which is what you want for structured output.

Step 3: Multi-turn conversation

LangChain uses message objects rather than raw strings for stateful dialog. Use SystemMessage and HumanMessage to constrain behavior and preserve context.

from langchain_core.messages import SystemMessage, HumanMessage

messages = [
    SystemMessage(content="You are a terse DevOps assistant. Reply in one line."),
    HumanMessage(content="What does 'idempotent' mean in deployment scripts?"),
]

resp = llm.invoke(messages)
print(resp.content)

Expected output:

An idempotent deployment can run multiple times without changing the result beyond the first apply.

To continue the thread, append the AIMessage returned by the model and the next HumanMessage:

from langchain_core.messages import AIMessage

messages.append(AIMessage(content=resp.content))
messages.append(HumanMessage(content="Give an example command that is idempotent."))

resp2 = llm.invoke(messages)
print(resp2.content)

This pattern is the foundation for chatbots. In production you would persist the messages list in a session store rather than keeping it in memory.

Step 4: Streaming tokens

For CLI tools or live UIs, stream tokens instead of blocking on the full response. ChatOpenAI exposes .stream().

for chunk in llm.stream("List three Linux distros optimized for containers:"):
    print(chunk.content, end="", flush=True)
print()

Expected output (order preserved, no newlines between tokens):

Alpine, Flatcar Container Linux, RancherOS

Streaming respects the same base_url and auth. You get Chunk objects with .content strings; aggregate them if you need the full text. Under the hood LangChain uses Server-Sent Events, so latency to first token drops from hundreds of milliseconds to tens.

Why streaming matters

If you build a user-facing feature, blocking on invoke makes the interface feel frozen. Streaming also lets you cancel generation mid-flight when the user hits stop, saving tokens. The gateway forwards the upstream token stream without buffering, so your stream() loop sees the same chunks the provider emits.

Step 5: Reading usage and passing cache hints

Production calls need token accounting. LangChain surfaces usage in response.usage_metadata when the provider returns it.

resp = llm.invoke("Explain the difference between a pod and a container in 20 words.")
print(resp.usage_metadata)

Typical output:

{'input_tokens': 15, 'output_tokens': 22, 'total_tokens': 37}

The gateway performs per-token usage metering, so the numbers reflect what you’ll be billed. To forward provider cache-control hints (for example, to encourage ephemeral caching on supported upstreams), pass them as extra headers:

llm_with_cache = ChatOpenAI(
    model="anthropic/claude-3-haiku",
    base_url="https://api.n4n.ai/v1",
    api_key=os.environ["N4N_API_KEY"],
    extra_headers={"Cache-Control": "max-age=300"},
)

cached_resp = llm_with_cache.invoke("Repeat after me: caching saves tokens.")
print(cached_resp.content)

The header is forwarded unchanged to the upstream provider that supports it. Not every model honors cache hints, but the ones that do will reuse prompt prefixes and cut your input token count on repeated calls.

Step 6: Resilience without custom retry logic

Because the n4n.ai endpoint handles automatic fallback when a provider is rate-limited or degraded, the same ChatOpenAI instance keeps serving requests even if the primary route errors. You do not need to wrap invoke in manual try/except loops for provider 429s.

If you want explicit routing control, pass a model prefix or use extra_headers to set a client routing directive; the gateway honors it and still applies fallback on failure.

# Force a specific provider family, fallback still applies on outage
routed_llm = ChatOpenAI(
    model="openai/gpt-4o",
    base_url="https://api.n4n.ai/v1",
    api_key=os.environ["N4N_API_KEY"],
    extra_headers={"x-routing": "openai"},
)
print(routed_llm.invoke("Ping").content)

Step 7: Async and batch calls

LangChain supports ainvoke and abatch for concurrent workloads. If you serve requests from an async web framework like FastAPI, use the async path to avoid blocking the event loop.

import asyncio

async def main():
    tasks = [llm.ainvoke(f"Say {word}") for word in ["hi", "yo", "hey"]]
    results = await asyncio.gather(*tasks)
    for r in results:
        print(r.content)

asyncio.run(main())

For batch embedding or bulk summarization, llm.batch([...]) wraps the same logic synchronously and handles concurrency internally.

Step 8: Putting it together in a script

Here is a minimal runnable file that covers invoke, streaming, and usage:

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage

load_dotenv()

llm = ChatOpenAI(
    model="openai/gpt-4o-mini",
    base_url="https://api.n4n.ai/v1",
    api_key=os.environ["N4N_API_KEY"],
)

# Single shot
print(llm.invoke("Say hi in JSON").content)

# Stream
for chunk in llm.stream("Name a fast web framework:"):
    print(chunk.content, end="", flush=True)
print()

# Usage
resp = llm.invoke([SystemMessage(content="Be brief"), HumanMessage(content="HTTP vs gRPC?")])
print(resp.usage_metadata)

Run it with python main.py. You should see the JSON, the streamed framework name, and a usage dict.

Where to go next

Swap the model string to any of the 240+ identifiers the gateway exposes. Add LangChain memory or a RunnableWithMessageHistory for stateful agents. The client code stays identical; only the model route changes.

This tutorial deliberately stayed at the raw client level. Once this works, layer in LangChain chains, tools, or retrievers without touching the ChatOpenAI configuration.