LangChain流式聊天补全结合WebSocket：AsyncCallbackHandler失效排查

你遇到的核心问题是：同步调用LLM时，LangChain不会触发AsyncCallbackHandler的异步回调方法。BaseCallbackHandler是同步回调，同步调用LLM可以正常触发；但AsyncCallbackHandler需要配合LLM的异步调用接口才能工作，同步调用会跳过异步回调逻辑。

具体修改步骤 #

1. 将generate_cited_answer_stream改为异步函数
2. 替换LLM的同步调用为异步调用方法
3. 修正代码中重复定义的变量名（避免混淆）

修改后的完整代码：

async def send_message_to_room(room_group_name, message):
    print("sending message to room", room_group_name, message)
    channel_layer = get_channel_layer()
    await channel_layer.group_send(
        room_group_name,
        {
            "type": "chat_message",
            "message": message,
        }
    )

class MyCustomHandler(AsyncCallbackHandler):

    def __init__(self, room_group_name):
        self.channel_layer = get_channel_layer()
        self.room_group_name = room_group_name

        print("MyCustomHandler init")

    async def on_llm_new_token(self, token: str, **kwargs):
        print(token)
        await send_message_to_room(self.room_group_name, token)

async def generate_cited_answer_stream(roomname, question=question, texts=texts, responsetype="Simple and Pedagogical"
                                       , system_message_with_response_type=system_message_with_response_type
                                       , human_message_with_response_type=human_message_with_response_type):


    handler = MyCustomHandler(room_group_name=roomname)

    chat = ChatOpenAI(temperature=0, streaming=True, callbacks=[handler])

    # 修正重复定义的变量名
    system_message_prompt = SystemMessagePromptTemplate.from_template(system_message_with_response_type)
    human_message_prompt = HumanMessagePromptTemplate.from_template(human_message_with_response_type)

    chat_prompt = ChatPromptTemplate.from_messages([system_message_prompt, human_message_prompt])

    prompt_value = chat_prompt.format_prompt(question=question, texts=texts, responsetype=responsetype)

    # 改用异步调用方法，LangChain v0.1+用ainvoke，旧版本用acall
    await chat.ainvoke(prompt_value.to_messages())

额外注意事项 #

• 调用generate_cited_answer_stream的上下文必须是异步的：比如在Django Channels消费者中，要在异步方法内用await generate_cited_answer_stream(...)触发，不能用同步方式调用异步函数
• 确认LangChain版本：如果是旧版本（v0.1之前），异步调用方法是acall而不是ainvoke，请根据版本调整

内容的提问来源于stack exchange，提问作者Chinmay