本电子书开源，欢迎 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="讲个笑话")]]))

### 参考资料

1. 斯坦福问答数据集

# results matching ""

# No results matching ""