Django Channels for real-time LLM chat over WebSockets

Wiring django channels websockets llm chat into a Django app means handling bidirectional streaming without blocking the event loop. This tutorial builds a minimal but production-shaped chat backend that streams tokens from an OpenAI-compatible LLM API to the browser over WebSockets. You’ll get runnable code, routing config, and a JS snippet to test it.

Prerequisites

Python 3.11 or newer
Django 5.0+ and Channels 4.1+
openai Python package (v1.x) for async streaming
An API key from any OpenAI-compatible provider
Basic comfort with async/await

Set up a virtual environment and install dependencies:

python -m venv venv && source venv/bin/activate
pip install django channels openai

Scaffold the project:

django-admin startproject chatproj .
python manage.py startapp chat

Add chat and channels to INSTALLED_APPS in chatproj/settings.py.

Configure Channels

Channels replaces the default WSGI handler with an ASGI stack. In settings.py:

INSTALLED_APPS = [
    "django.contrib.admin",
    "django.contrib.auth",
    "django.contrib.contenttypes",
    "django.contrib.sessions",
    "django.contrib.messages",
    "django.contrib.staticfiles",
    "channels",
    "chat",
]

ASGI_APPLICATION = "chatproj.asgi.application"

CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels.layers.InMemoryChannelLayer"
    }
}

For local dev the in-memory layer is fine. In production use Redis:

CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {"hosts": [("127.0.0.1", 6379)]},
    }
}

ASGI routing

Create chatproj/asgi.py if it doesn’t exist. Wire HTTP to Django and WebSocket to our consumer.

import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
import chat.routing

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "chatproj.settings")

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": AuthMiddlewareStack(
        URLRouter(
            chat.routing.websocket_urlpatterns
        )
    ),
})

Create chat/routing.py:

from django.urls import re_path
from . import consumers

websocket_urlpatterns = [
    re_path(r"ws/chat/$", consumers.LLMChatConsumer.as_asgi()),
]

The streaming consumer

The core of django channels websockets llm chat is the AsyncWebsocketConsumer. It accepts the socket, accumulates conversation history, and proxies the LLM’s token stream back to the client.

import os
import json
from channels.generic.websocket import AsyncWebsocketConsumer
from openai import AsyncOpenAI

# Point at n4n.ai if you want automatic fallback across providers and per-token metering.
client = AsyncOpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url="https://api.openai.com/v1",
)

class LLMChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        await self.accept()
        self.history = []

    async def disconnect(self, close_code):
        # Cleanup if you track per-user state in a group
        pass

    async def receive(self, text_data):
        data = json.loads(text_data)
        user_msg = data.get("message", "")
        if not user_msg:
            return
        self.history.append({"role": "user", "content": user_msg})
        await self.send(text_data=json.dumps({"type": "start"}))

        try:
            stream = await client.chat.completions.create(
                model=data.get("model", "gpt-3.5-turbo"),
                messages=self.history,
                stream=True,
            )
            assistant_msg = ""
            async for chunk in stream:
                delta = chunk.choices[0].delta.content or ""
                if delta:
                    assistant_msg += delta
                    await self.send(text_data=json.dumps(
                        {"type": "token", "token": delta}
                    ))
            self.history.append({"role": "assistant", "content": assistant_msg})
            await self.send(text_data=json.dumps({"type": "done"}))
        except Exception as e:
            await self.send(text_data=json.dumps(
                {"type": "error", "message": str(e)}
            ))

The stream=True flag makes the OpenAI client yield incremental delta objects. We forward each non-empty delta immediately. This keeps latency low and gives the browser a typing effect.

If you swap base_url to https://api.n4n.ai/v1, the same code gets automatic fallback when a provider is rate-limited or degraded, and the gateway forwards provider cache-control hints. That’s the only change needed to make the chat resilient across 240+ models.

Frontend harness

You don’t need React for a smoke test. Drop this HTML file in chat/templates/chat/index.html and serve it via a plain Django view.

<!DOCTYPE html>
<html>
<body>
  <div id="out"></div>
  <input id="msg" /><button onclick="send()">Send</button>
  <script>
    const ws = new WebSocket("ws://" + location.host + "/ws/chat/");
    const out = document.getElementById("out");
    ws.onmessage = (e) => {
      const m = JSON.parse(e.data);
      if (m.type === "token") out.innerHTML += m.token;
      if (m.type === "start") out.innerHTML += "<br>> ";
      if (m.type === "done") out.innerHTML += "<br><br>";
    };
    function send() {
      const v = document.getElementById("msg").value;
      ws.send(JSON.stringify({message: v}));
      document.getElementById("msg").value = "";
    }
  </script>
</body>
</html>

Add a view and URL:

# chat/views.py
from django.shortcuts import render
def index(request):
    return render(request, "chat/index.html")

# chat/urls.py
from django.urls import path
from . import views
urlpatterns = [path("", views.index)]

Include chat.urls in root urls.py.

Run and verify

Start the dev server:

python manage.py runserver

Open http://127.0.0.1:8000/. Type “Explain asyncio in one sentence”. Expected browser output:

> asyncio lets you write concurrent code using coroutines that yield control instead of blocking threads.

Server-side you’ll see no errors, and the WS connection stays open for follow-up messages. The history array maintains context, so a subsequent “Now compare it to threads” will get a contextual answer.

Test without a browser

Using the websockets package from a Python shell validates the backend quickly:

import asyncio, json, websockets

async def main():
    async with websockets.connect("ws://127.0.0.1:8000/ws/chat/") as ws:
        await ws.send(json.dumps({"message": "Hi"}))
        async for msg in ws:
            m = json.loads(msg)
            if m["type"] == "token":
                print(m["token"], end="")
            elif m["type"] == "done":
                break

asyncio.run(main())

Expected stdout is the streamed assistant response with no newlines between tokens.

Production considerations

The in-memory channel layer does not work across multiple Django instances. Use Redis and run at least two Daphne or Uvicorn workers behind a reverse proxy that supports WebSocket upgrade (nginx with proxy_set_header Upgrade $http_upgrade).

Cap self.history to avoid unbounded memory growth. A simple turn limit:

if len(self.history) > 20:
    self.history = self.history[-20:]

If you need multi-user rooms, leverage channel_layers.group_add and broadcast tokens to a group instead of a single socket. Authenticate users via AuthMiddlewareStack and read self.scope["user"] inside connect.

Why this shape

Django Channels gives you the ASGI primitives; the LLM call is just another async coroutine. The mistake engineers make is buffering the entire model response before sending. Streaming token-by-token is what makes django channels websockets llm chat feel responsive. Keep the consumer thin, push errors to the client, and let the gateway handle provider routing.