原

LangChain 回调 (Callback)

5559人阅读 2023/10/5 16:08 总访问：3143449 评论：0 收藏：0 手机

分类: 云产品

![](https://img.tnblog.net/arcimg/hb/dcbb080269134178b1048148e5315d8f.png)
>#LangChain 回调 (Callback)
[TOC]

## 简介

tn2>`Callback` 是 `LangChain` 提供的回调机制，允许我们在 `LLM` 应用程序的各个阶段使用 `Hook`（钩子）。这对于记录日志、监控、流式传输等任务非常有用。这些任务的执行逻辑由回调处理器（`CallbackHandler`）定义。<br/>

## 回调处理器

tn2>在 `Python` 程序中， 回调处理器通过继承 `BaseCallbackHandler` 来实现。`BaseCallbackHandler` 接口对每一个可订阅的事件定义了一个回调函数。`BaseCallbackHandler` 的子类可以实现这些回调函数来处理事件。当事件触发时，`LangChain` 的回调管理器 `CallbackManager` 会调用相应的回调函数。<br/>
以下是 `BaseCallbackHandler` 的定义。请参考[源代码](https://github.com/langchain-ai/langchain/blob/v0.0.235/langchain/callbacks/base.py#L225)。


```python
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."""
```

tn2>`LangChain` 内置支持了一系列回调处理器，我们也可以按需求自定义处理器，以实现特定的业务。

## 内置处理器

tn2>`StdOutCallbackHandler` 是 `LangChain` 所支持的最基本的处理器。它将所有的回调信息打印到标准输出。这对于调试非常有用。<br/>
`LangChain` 链的基类 `Chain` 提供了一个 `callbacks` 参数来指定要使用的回调处理器。请参考[`Chain源码`](https://github.com/langchain-ai/langchain/blob/v0.0.235/langchain/chains/base.py#L63)，其中代码片段为：

```python
class Chain(Serializable, ABC):
    """Abstract base class for creating structured sequences of calls to components.
    ...
    callbacks: Callbacks = Field(default=None, exclude=True)
    """Optional list of callback handlers (or callback manager). Defaults to None.
    Callback handlers are called throughout the lifecycle of a call to a chain,
    starting with on_chain_start, ending with on_chain_end or on_chain_error.
    Each custom chain can optionally call additional callback methods, see Callback docs
    for full details."""
```

tn2>用法如下：

```python
from langchain.callbacks import StdOutCallbackHandler
from langchain.chains import LLMChain
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate

handler = StdOutCallbackHandler()
llm = OpenAI()
prompt = PromptTemplate.from_template("Who is {name}?")
chain = LLMChain(llm=llm, prompt=prompt, callbacks=[handler])
chain.run(name="Super Mario")
```


tn2>你应该期望如下输出：

![](https://img.tnblog.net/arcimg/hb/8bf23d81db7a4131bc8412d677e790e5.png)

## 自定义处理器

tn2>我们可以通过继承 `BaseCallbackHandler` 来实现自定义的回调处理器。下面是一个简单的例子，`TimerHandler` 将跟踪 `Chain` 或 `LLM` 交互的起止时间，并统计每次交互的处理耗时。


```python
from langchain.callbacks.base import BaseCallbackHandler
import time

class TimerHandler(BaseCallbackHandler):

    def __init__(self) -> None:
        super().__init__()
        self.previous_ms = None
        self.durations = []

    def current_ms(self):
        return int(time.time() * 1000 + time.perf_counter() % 1 * 1000)

    def on_chain_start(self, serialized, inputs, **kwargs) -> None:
        self.previous_ms = self.current_ms()

    def on_chain_end(self, outputs, **kwargs) -> None:
        if self.previous_ms:
          duration = self.current_ms() - self.previous_ms
          self.durations.append(duration)

    def on_llm_start(self, serialized, prompts, **kwargs) -> None:
        self.previous_ms = self.current_ms()

    def on_llm_end(self, response, **kwargs) -> None:
        if self.previous_ms:
          duration = self.current_ms() - self.previous_ms
          self.durations.append(duration)

llm = OpenAI()
timerHandler = TimerHandler()
prompt = PromptTemplate.from_template("What is the HEX code of color {color_name}?")
chain = LLMChain(llm=llm, prompt=prompt, callbacks=[timerHandler])
response = chain.run(color_name="blue")
print(response)
response = chain.run(color_name="purple")
print(response)

timerHandler.durations
```

tn2>你应该期望如下输出：

![](https://img.tnblog.net/arcimg/hb/c4213b3c28c549f794a3450e5ceba6c9.png)

## 回调处理器的适用场景

tn2>通过 `LLMChain` 的构造函数参数设置 `callbacks` 仅仅是众多适用场景之一。接下来我们简明地列出其他使用场景和示例代码。<br/>
对于 `Model`，`Agent`， `Tool`，以及 `Chain` 都可以通过以下方式设置回调处理器：

### 1. 构造函数参数 `callbacks` 设置

tn2>关于 `Chain`，以 `LLMChain` 为例，请参考本讲上一部分内容。注意在 `Chain` 上的回调器监听的是 `chain` 相关的事件，因此回调器的如下函数会被调用：
- on_chain_start
- on_chain_end
- on_chain_error
`Agent`， `Tool`，以及 `Chain` 上的回调器会分别被调用相应的回调函数。
下面分享关于 `Model` 与 `callbacks` 的使用示例：


```python
timerHandler = TimerHandler()
llm = OpenAI(callbacks=[timerHandler])
response = llm.predict("What is the HEX code of color BLACK?")
print(response)

timerHandler.durations
```

tn2>你应该期望看到类似如下的输出：

![](https://img.tnblog.net/arcimg/hb/3fb22feaf7e340f297239eb8bb8a0f2d.png)

### 2. 通过运行时的函数调用

tn2>`Model`，`Agent`， `Tool`，以及 `Chain` 的请求执行函数都接受 `callbacks` 参数，比如 `LLMChain` 的 `run` 函数，`OpenAI` 的 `predict` 函数，等都能接受 `callbacks` 参数，在运行时指定回调处理器。
以 `OpenAI` 模型类为例：


```python
timerHandler = TimerHandler()
llm = OpenAI()
response = llm.predict("What is the HEX code of color BLACK?", callbacks=[timerHandler])
print(response)

timerHandler.durations
```

tn2>你应该同样期望如下输出：

![](https://img.tnblog.net/arcimg/hb/965f968d5b474a26a62781fa910dcbb0.png)

tn2>关于 `Agent`，`Tool` 等的使用，请参考官方文档API。

欢迎加群讨论技术，1群：677373950(满了，可以加，但通过不了)，2群：656732739

👈{{preArticle.title}}

👉{{nextArticle.title}}

评价

尘叶心繁

这一世以无限游戏为使命！

博主信息

排名

文章

粉丝

文章类别

.net后台框架 184篇

linux 18篇

linux中cve 1篇

windows中cve 0篇

资源分享 10篇

Win32 3篇

前端 28篇

传说中的c 5篇

Xamarin 9篇

docker 15篇

容器编排 101篇

grpc 4篇

Go 15篇

yaml模板 1篇

理论 2篇

LangChain 回调 (Callback)

{{titleitem}} {{titleitem}}

{{titleitem}} {{titleitem}}