Problems with AsyncCallbackHandler in WebSocket streaming using langchain

Other Questions langchain
1

I’m facing some challenges with the AsyncCallbackHandler while trying to stream responses in a WebSocket environment with langchain. The BaseCallbackHandler functions correctly and displays tokens in the console, but when I try to use AsyncCallbackHandler, I don’t see any activity after the initialization stage. It seems that the callback methods are not executing as expected.

async def distribute_message_to_room(room_name, message_content):
    print("distributing message to the room", room_name, message_content)
    channel_layer = get_channel_layer()
    await channel_layer.group_send(
        room_name,
        {
            "type": "message_event",
            "content": message_content,
        }
    )

class CustomStreamingHandler(AsyncCallbackHandler):

    def __init__(self, room_name):
        self.channel_layer = get_channel_layer()
        self.room_name = room_name
        print("CustomStreamingHandler initialized")

    async def on_llm_new_token(self, token_value: str, **kwargs):
        print(token_value)
        await distribute_message_to_room(self.room_name, token_value)

def stream_response(room_id, inquiry=inquiry, context=context_data, response_type="Informative"
                    , system_message_format=system_message_format
                    , user_message_format=user_message_format):

    streaming_handler = CustomStreamingHandler(room_name=room_id)
    
    chat_model = ChatOpenAI(temperature=0, streaming=True, callbacks=[streaming_handler])
    
    system_message = SystemMessagePromptTemplate.from_template(system_message_format)
    user_message = HumanMessagePromptTemplate.from_template(user_message_format)
    
    complete_prompt_template = ChatPromptTemplate.from_messages([system_message, user_message])
    
    formatted_prompt = complete_prompt_template.format_prompt(inquiry=inquiry, context=context_data, response_type=response_type)
    
    chat_model(formatted_prompt.to_messages())

Does anyone have insights on why the asynchronous callbacks are not being triggered? The print statement during initialization runs, but nothing else happens afterward.

2

Beyond the sync/async mismatch, langchain sometimes needs explicit callback setup. I hit this building a document processing pipeline with streaming outputs. The AsyncCallbackHandler needs proper event loop integration in the langchain context.

Try wrapping your model call with asyncio.create_task() or make sure you’re running in an existing event loop. Also check that your Django channels layer is configured for async operations. My callbacks were silently failing because the channel layer wasn’t initialized correctly in async context.

Add error handling around your distribute_message_to_room function and see if exceptions are getting swallowed. Sometimes the callback runs but fails during websocket send without proper error reporting.

3

yeah, this is def a context issue like emparker said. also, check if ur websocket connection stays open while streaming - I had similar probs where callbacks worked but the websocket closed unexpectedly. add logging to your distribute_message_to_room func to see if it gets called.

4

This callback nightmare is exactly why I ditched langchain’s streaming quirks. The async handler registration is flaky and Django channels just makes it worse.

Hit the same wall building a live support chat. Instead of debugging langchain’s broken callback system, I moved everything to Latenode. You can build a workflow that handles OpenAI streaming directly and pushes tokens to your WebSocket room - no callback hell.

With Latenode, streaming becomes a simple workflow. OpenAI node connects to WebSocket node, tokens flow automatically. No AsyncCallbackHandler headaches, no sync/async nightmares, no Django channels debugging.

Got the same streaming chat running in 10 minutes. The workflow handles async naturally and you just call it from Django. Way cleaner than forcing langchain callbacks to work with Django channels.

Check it out: https://latenode.com

5

Had this exact same issue building a real-time chat app with langchain. It’s not just the sync/async mismatch - langchain’s callback registration is also buggy.

What I found is ChatOpenAI’s constructor doesn’t properly bind AsyncCallbackHandlers when you create them in sync context, even if you await the calls later. The handler registers but the event loop reference goes stale.

Create your ChatOpenAI model inside an async function instead of passing the handler during construction. Build the model after you’re in async context, then add callbacks with add_handler() if it’s available.

Also check your Django channels config - make sure CHANNEL_LAYERS uses Redis or InMemory backend that supports async. I wasted hours on silent callbacks only to discover my channel layer defaulted to sync-only and was dropping async messages without any errors.

6

You’re mixing sync and async contexts, so your async callbacks never get called.

Your stream_response function is synchronous but you’re using AsyncCallbackHandler. Sync functions can’t execute async callbacks properly.

I hit this exact problem last year building a similar streaming setup. Here’s what worked:

async def stream_response(room_id, inquiry=inquiry, context=context_data, response_type="Informative"
                    , system_message_format=system_message_format
                    , user_message_format=user_message_format):

    streaming_handler = CustomStreamingHandler(room_name=room_id)
    
    chat_model = ChatOpenAI(temperature=0, streaming=True, callbacks=[streaming_handler])
    
    system_message = SystemMessagePromptTemplate.from_template(system_message_format)
    user_message = HumanMessagePromptTemplate.from_template(user_message_format)
    
    complete_prompt_template = ChatPromptTemplate.from_messages([system_message, user_message])
    
    formatted_prompt = complete_prompt_template.format_prompt(inquiry=inquiry, context=context_data, response_type=response_type)
    
    await chat_model.agenerate([formatted_prompt.to_messages()])

Three key changes:

  1. Make stream_response async
  2. Use agenerate() instead of calling the model directly
  3. Await the result

Now your async callbacks will run in the right context. Without this, langchain just skips async callbacks entirely.

7

Been fighting this callback nightmare for months across different projects. It’s not just sync/async - Django screws up the callback execution context.

Django channels creates your AsyncCallbackHandler in the main thread but tries to execute it in the channel worker thread. The event loop context gets lost.

Here’s what actually worked:

class CustomStreamingHandler(AsyncCallbackHandler):
    def __init__(self, room_name):
        self.room_name = room_name
        print("CustomStreamingHandler initialized")

    async def on_llm_new_token(self, token_value: str, **kwargs):
        # Force execution in current event loop
        loop = asyncio.get_event_loop()
        await loop.create_task(distribute_message_to_room(self.room_name, token_value))

Make sure your Django view calling stream_response has @sync_to_async if it’s sync, or just make it async.

The callbacks run but Django channels silently drops them when the context is wrong. Add exception handling everywhere - you’ll see connection errors that were getting swallowed.

Took weeks to figure this out on a customer support chat. Callbacks worked fine in dev but died in production because of thread context differences.