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

先决条件

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

回调

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 参考: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 可运行对象。

下一步

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

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

下一步​

下一步