如何创建自定义回调处理器

先决条件

本指南假定您熟悉以下概念：

回调

LangChain 包含一些内置的回调处理器，但您通常会希望创建具有自定义逻辑的处理器。

要创建自定义回调处理器，我们需要确定希望回调处理器处理的事件以及当事件触发时我们希望回调处理器做什么。然后，我们只需将回调处理器附加到对象上，例如通过构造函数或在运行时。

在下面的示例中，我们将通过自定义处理器实现流式传输。

在我们的自定义回调处理器 MyCustomHandler 中，我们实现了 on_llm_new_token 来打印我们刚刚收到的 token。然后，我们将自定义处理器作为构造函数回调附加到模型对象。

from langchain_anthropic import ChatAnthropic
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.prompts import ChatPromptTemplate


class MyCustomHandler(BaseCallbackHandler):
    def on_llm_new_token(self, token: str, **kwargs) -> None:
        print(f"My custom handler, token: {token}")


prompt = ChatPromptTemplate.from_messages(["Tell me a joke about {animal}"])

# To enable streaming, we pass in `streaming=True` to the ChatModel constructor
# Additionally, we pass in our custom handler as a list to the callbacks parameter
model = ChatAnthropic(
    model="claude-3-sonnet-20240229", streaming=True, callbacks=[MyCustomHandler()]
)

chain = prompt | model

response = chain.invoke({"animal": "bears"})

API Reference:ChatAnthropic | BaseCallbackHandler | ChatPromptTemplate

My custom handler, token: Here
My custom handler, token: 's
My custom handler, token:  a
My custom handler, token:  bear
My custom handler, token:  joke
My custom handler, token:  for
My custom handler, token:  you
My custom handler, token: :
My custom handler, token: 

Why
My custom handler, token:  di
My custom handler, token: d the
My custom handler, token:  bear
My custom handler, token:  dissol
My custom handler, token: ve
My custom handler, token:  in
My custom handler, token:  water
My custom handler, token: ?
My custom handler, token: 
Because
My custom handler, token:  it
My custom handler, token:  was
My custom handler, token:  a
My custom handler, token:  polar
My custom handler, token:  bear
My custom handler, token: !

您可以在此参考页面查看您可以处理的事件列表。请注意，handle_chain_* 事件会针对大多数 LCEL 可运行对象运行。

后续步骤

您现在已经学会了如何创建自己的自定义回调处理程序。

接下来，请查看本节中的其他操作指南，例如如何将回调附加到可运行对象。

后续步骤​

后续步骤