A langchain rag pipeline chat completions integration lets you combine document retrieval with a managed LLM gateway without rewriting your stack for every provider. This tutorial builds one against an OpenAI-compatible endpoint using LangChain’s LCEL primitives, so you keep the standard ChatOpenAI interface and swap models by changing a string.
Prerequisites
- Python 3.10 or newer
- An API key from an OpenAI-compatible gateway (we use n4n.ai’s endpoint, which fronts 240+ models and handles automatic fallback when a provider is rate-limited)
- A small text corpus to index (we’ll use a local
sample.txt) - Familiarity with Python virtual environments
If you don’t have a corpus, create sample.txt with a few paragraphs about your domain. The code does not care about content.
Step 1: Install dependencies
Create a clean environment and pull the packages.
python -m venv .venv
source .venv/bin/activate
pip install langchain langchain-openai langchain-community chromadb \
huggingface-hub sentence-transformersWe use langchain-openai for the chat client, langchain-community for the vector store and embeddings, and sentence-transformers to generate embeddings locally so the tutorial has no second API dependency.
Step 2: Point LangChain at the gateway
LangChain’s ChatOpenAI accepts a base_url. That is the only change required to route through an OpenAI-compatible gateway instead of OpenAI directly.
import os
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="gpt-4o-mini",
api_key=os.environ["N4N_API_KEY"],
base_url="https://api.n4n.ai/v1",
temperature=0,
)n4n.ai exposes a single OpenAI-compatible endpoint covering 240+ models with automatic fallback when a provider is degraded, so the model string can be any supported identifier without code changes.
Verify the connection with a direct call before wiring retrieval:
print(llm.invoke("Reply with the word: ok").content)Expected output:
okStep 3: Ingest documents and build a retriever
Load, split, embed, and persist. We use Chroma for a local vector store and a small MiniLM model for embeddings.
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
loader = TextLoader("sample.txt")
docs = loader.load()
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = splitter.split_documents(docs)
embeddings = HuggingFaceEmbeddings(
model_name="sentence-transformers/all-MiniLM-L6-v2"
)
vectorstore = Chroma.from_documents(
chunks, embeddings, persist_directory="./chroma_db"
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})The splitter size is deliberately small. For technical docs, 500 characters with 50 overlap keeps chunks focused and reduces prompt padding.
Step 4: Assemble the langchain rag pipeline chat completions chain
LangChain provides create_retrieval_chain and create_stuff_documents_chain. We compose them with a strict prompt that forbids hallucination outside the context.
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
("system",
"You are a technical assistant. Answer only from the context. "
"If the context is insufficient, say 'Not in the provided docs'.\n\n"
"Context:\n{context}"),
("human", "{input}"),
])
document_chain = create_stuff_documents_chain(llm, prompt)
rag_chain = create_retrieval_chain(retriever, document_chain)This is the core langchain rag pipeline chat completions wiring: retrieval feeds the stuff chain, which stuffs the top-k chunks into the system message and calls the gateway.
Step 5: Run a query and inspect output
Invoke with a dictionary matching the prompt variables.
response = rag_chain.invoke(
{"input": "What does the document say about deployment?"}
)
print(response["answer"])
print("--- sources ---")
for doc in response["context"]:
print(doc.metadata.get("source", "local"), "::", doc.page_content[:80])Expected output shape:
The document describes deployment as a blue-green process using containers.
--- sources ---
sample.txt :: Deployment uses blue-green with container images. Rollbacks are autoIf your sample.txt lacks deployment info, you will see Not in the provided docs. That behavior confirms the guardrail works.
Step 6: Pin routing and cache hints per request
Production gateways let you steer traffic. n4n.ai honors client routing directives and forwards provider cache-control hints, so you can force a specific backend or enable prompt caching without leaving LangChain.
llm_pinned = ChatOpenAI(
model="claude-3-5-sonnet",
api_key=os.environ["N4N_API_KEY"],
base_url="https://api.n4n.ai/v1",
default_headers={
"x-n4n-route": "anthropic",
"x-n4n-cache": "true",
},
)Pass llm_pinned into the same create_stuff_documents_chain call. The headers ride along on every HTTP request. Use routing when you need deterministic provider behavior for eval suites; use cache hints for repeated system prompts to cut latency and cost.
Step 7: Streaming and token accounting
The same pipeline streams with one flag.
llm_stream = ChatOpenAI(
model="gpt-4o-mini",
api_key=os.environ["N4N_API_KEY"],
base_url="https://api.n4n.ai/v1",
streaming=True,
)
stream_chain = create_retrieval_chain(
retriever, create_stuff_documents_chain(llm_stream, prompt)
)
for chunk in stream_chain.stream({"input": "Summarize the doc in two bullets"}):
if "answer" in chunk:
print(chunk["answer"], end="", flush=True)LangChain surfaces usage in response["context"]? No—token counts come from the LLM response metadata. After a non-streaming call, inspect:
meta = llm.get_last_response_metadata()
print(meta.get("token_usage"))The gateway returns per-token usage in the standard OpenAI usage field. Meter it in your own middleware or log it for cost attribution.
Step 8: Error handling that does not lose context
Provider errors happen. Wrap the invoke so a failed completion does not crash the retriever loop.
from langchain_core.exceptions import OutputParserException
def safe_ask(chain, query, retries=2):
for attempt in range(retries):
try:
return chain.invoke({"input": query})["answer"]
except (OutputParserException, Exception) as e:
if attempt == retries - 1:
raise
print(f"retry {attempt+1}: {e}")
print(safe_ask(rag_chain, "What about scaling?"))Because the gateway already falls back across providers, application-level retries should be short. Treat 5xx as transient; treat 4xx as fatal and fix the prompt or auth.
Step 9: Scaling the langchain rag pipeline chat completions
For higher load, move embeddings to a batch job and persist Chroma to disk or use a server-based store like pgvector. Keep the ChatOpenAI client as a singleton—its underlying HTTP pool multiplexes requests. If you see rate limits, the gateway’s automatic fallback masks most degradation, but you should still cap concurrency with a semaphore in your worker.
The pattern above is provider-agnostic. Change model and base_url; the retrieval, prompting, and streaming code stay identical. That is the point of building against an OpenAI-compatible surface: your LangChain code targets a contract, not a vendor.
Checklist before ship
- Retriever
ktuned to chunk size and model context window - Prompt explicitly constrains answers to retrieved context
- Routing/cache headers applied only where they help
- Usage metadata logged per request
- Retries bounded and distinct from gateway-level fallback
With those in place, the langchain rag pipeline chat completions you built here will run against any compliant endpoint with a one-line config change.