本电子书开源，欢迎 star 🌟，关注《LLM 应用开发实践笔记》
我的新书《LangChain编程从入门到实践》 已经开售！推荐正在学习AI应用开发的朋友购买阅读！
LangChain模块之Callbacks
回调模块允许接到LLM应用程序的各个阶段，鉴于LLM的幻觉问题，这对于日志记录、监视、流式处理和其他任务非常有用，现在也有专用的工具Helicone，Arize AI等产品可用，具体看LLM应用生态初创公司说明
自定义回调对象
所有的回调对象都是基于这个基类来声明的
class BaseCallbackHandler:
"""Base callback handler that can be used to handle callbacks from langchain."""
def on_llm_start(
self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
) -> Any:
"""Run when LLM starts running."""
def on_chat_model_start(
self, serialized: Dict[str, Any], messages: List[List[BaseMessage]], **kwargs: Any
) -> Any:
"""Run when Chat Model starts running."""
def on_llm_new_token(self, token: str, **kwargs: Any) -> Any:
"""Run on new LLM token. Only available when streaming is enabled."""
def on_llm_end(self, response: LLMResult, **kwargs: Any) -> Any:
"""Run when LLM ends running."""
def on_llm_error(
self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any
) -> Any:
"""Run when LLM errors."""
def on_chain_start(
self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any
) -> Any:
"""Run when chain starts running."""
def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> Any:
"""Run when chain ends running."""
def on_chain_error(
self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any
) -> Any:
"""Run when chain errors."""
def on_tool_start(
self, serialized: Dict[str, Any], input_str: str, **kwargs: Any
) -> Any:
"""Run when tool starts running."""
def on_tool_end(self, output: str, **kwargs: Any) -> Any:
"""Run when tool ends running."""
def on_tool_error(
self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any
) -> Any:
"""Run when tool errors."""
def on_text(self, text: str, **kwargs: Any) -> Any:
"""Run on arbitrary text."""
def on_agent_action(self, action: AgentAction, **kwargs: Any) -> Any:
"""Run on agent action."""
def on_agent_finish(self, finish: AgentFinish, **kwargs: Any) -> Any:
"""Run on agent end."""
使用回调的两种方式
- 构造函数时定义回调：在构造函数中定义，例如
LLMChain(callbacks=[handler], tags=['a-tag'])，它将被用于对该对象的所有调用，并且将只针对该对象，例如，如果你向LLMChain构造函数传递一个handler，它将不会被附属于该链的Model使用。 - 请求函数时传入回调：定义在用于发出请求的call()/run()/apply()方法中，例如
chain.call(inputs, callbacks=[handler])，它将仅用于该特定请求，以及它所包含的所有子请求（例如，对LLMChain的调用会触发对Model的调用，Model会使用call()方法中传递的相同 handler）。
下面这是采用构造函数定义回调的例子：
class MyCustomSyncHandler(BaseCallbackHandler):
def on_llm_new_token(self, token: str, **kwargs) -> None:
print(f"同步回调被调用: token: {token}")
class MyCustomAsyncHandler(AsyncCallbackHandler):
async def on_llm_start(
self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
) -> None:
"""Run when chain starts running."""
print("LLM调用开始....")
await asyncio.sleep(0.3)
print("Hi! I just woke up. Your llm is starting")
async def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
"""Run when chain ends running."""
print("LLM调用结束....")
await asyncio.sleep(0.3)
print("Hi! I just woke up. Your llm is ending")
if __name__ == "__main__":
chat = ChatOpenAI(
max_tokens=25,
streaming=True,
callbacks=[MyCustomSyncHandler(), MyCustomAsyncHandler()],
)
asyncio.run(chat.agenerate([[HumanMessage(content="讲个笑话")]]))