自定义回调处理程序
要创建自定义回调处理程序,首先需要确定我们希望它处理的 事件,以及当这些事件被触发时我们希望它做什么。然后只需将回调处理程序附加到对象上——作为构造函数回调或请求回调(参见 回调类型)。
在下面的示例中,我们将用自定义处理程序实现流式处理。
在自定义回调处理程序 MyCustomHandler 中,我们实现了 on_llm_new_token 来打印刚接收到的 token;随后将自定义处理程序作为构造函数回调附加到模型对象上。
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
class MyCustomHandler(BaseCallbackHandler):
def on_llm_new_token(self, token: str, **kwargs) -> None:
print(f"我的自定义处理程序,token: {token}")
prompt = ChatPromptTemplate.from_messages(["给我讲一个关于 {animal} 的笑话"])
# 要启用流式处理,我们在 ChatModel 构造函数中传入 `streaming=True`
# 另外,我们将自定义处理程序以列表形式传入 callbacks 参数
model = ChatOpenAI(streaming=True, callbacks=[MyCustomHandler()])
chain = prompt | model
response = chain.invoke({"animal": "bears"})我的自定义处理程序,token:
我的自定义处理程序,token: 为什么
我的自定义处理程序,token: 熊
我的自定义处理程序,token: 有
我的自定义处理程序,token: 浓密的
我的自定义处理程序,token: 毛
我的自定义处理程序,token: ?
我的自定义处理程序,token: 为了
我的自定义处理程序,token: 保
我的自定义处理程序,token: 护
我的自定义处理程序,token: !
我的自定义处理程序,token: