【AI大模型应用开发】【LangChain系列】6. LangChain的Callbacks模块：监控调试程序的重要手段_langchain callbacks

大家好，我是【同学小张】。持续学习，持续干货输出，关注我，跟我一起学AI大模型技能。

LangChain提供了一个回调系统，允许您挂接到LLM应用程序的各个阶段。这对于日志记录、监视、流式传输和其他任务非常有用。

0. LangChain Callbacks模块提供的Callback接口一览

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."""
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

1. 最常用的Callback：StdOutCallbackHandler

StdOutCallbackHandler将所有事件的日志作为标准输出，打印到终端中。

注意: 当verbose参数设置为true时, StdOutCallbackHandler是被默认启用的，也就是你看到的它将运行过程的日志全部打印到了终端窗口中。

上示例：

from langchain.callbacks import StdOutCallbackHandler
from langchain.chains import LLMChain
from langchain_openai import OpenAI
from langchain.prompts import PromptTemplate

handler = StdOutCallbackHandler()
llm = OpenAI()
prompt = PromptTemplate.from_template("1 + {number} = ")

# Constructor callback: First, let's explicitly set the StdOutCallbackHandler when initializing our chain
chain = LLMChain(llm=llm, prompt=prompt, callbacks=[handler])
chain.invoke({"number":2})

# Use verbose flag: Then, let's use the `verbose` flag to achieve the same result
chain = LLMChain(llm=llm, prompt=prompt, verbose=True)
chain.invoke({"number":2})

# Request callbacks: Finally, let's use the request `callbacks` to achieve the same result
chain = LLMChain(llm=llm, prompt=prompt)
chain.invoke({"number":2}, {"callbacks":[handler]})

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

输出：

对代码和运行结果的解释：

从运行结果可以看出，三次输出的结果相同。再看代码，用三种方式实现了StdOutCallbackHandler的设置。

• 第一种：chain = LLMChain(llm=llm, prompt=prompt, callbacks=[handler])，chain中直接在callbacks中将callback handler传入
• 第二种：使用verbose=True，即使不显式声明callbacks，它也使用StdOutCallbackHandler
• 第三种：chain.invoke({"number":2}, {"callbacks":[handler]})，在invoke时传入callbacks

2. 各种类型的CallBack实践

2.1 通用 callback：BaseCallbackHandler

实现一个自己的Callback handler，继承自BaseCallbackHandler，然后重写自己需要的回调函数即可。

from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import HumanMessage
from langchain_openai import ChatOpenAI


class MyCustomHandler(BaseCallbackHandler):
    def on_llm_new_token(self, token: str, **kwargs) -> None:
        print(f"My custom handler, token: {token}")


# To enable streaming, we pass in `streaming=True` to the ChatModel constructor
# Additionally, we pass in a list with our custom handler
chat = ChatOpenAI(max_tokens=25, streaming=True, callbacks=[MyCustomHandler()])

chat([HumanMessage(content="Tell me a joke")])
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

运行结果：

2.2 异步 CallBack：AsyncCallbackHandler

有时候我们可能在CallBack内做大量的数据处理，可能比较耗时，如果使用通用 CallBack，会阻塞主线程运行，这时候异步 CallBack就比较有用了。

实现一个自己的Callback handler，继承自AsyncCallbackHandler，然后重写自己需要的回调函数即可。

class MyCustomAsyncHandler(AsyncCallbackHandler):
        """Async callback handler that can be used to handle callbacks from langchain."""
        ...... 重写相关回调函数 ......

1
2
3
4

2.3 写日志 / 日志文件： FileCallbackHandler

开发项目过程中，写日志是重要的调试手段之一。正式的项目中，我们不能总是将日志输出到终端中，这样无法传递和保存。

from langchain.callbacks import FileCallbackHandler
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain_openai import OpenAI

logfile = "output.log"

handler = FileCallbackHandler(logfile)

llm = OpenAI()
prompt = PromptTemplate.from_template("1 + {number} = ")

# this chain will both print to stdout (because verbose=True) and write to 'output.log'
# if verbose=False, the FileCallbackHandler will still write to 'output.log'
chain = LLMChain(llm=llm, prompt=prompt, callbacks=[handler], verbose=True)
answer = chain.run(number=2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

运行结果：

题外话：上面的log文件打开后有点乱码，可以用下面方法解析展示出来：

pip install --upgrade ansi2html
pip install ipython
1
2

from ansi2html import Ansi2HTMLConverter
from IPython.display import HTML, display

with open("output.log", "r") as f:
    content = f.read()

conv = Ansi2HTMLConverter()
html = conv.convert(content, full=True)

display(HTML(html))
1
2
3
4
5
6
7
8
9
10

2.4 Token计数：get_openai_callback

Token就是Money，所以知道你的程序运行中使用了多少Token也是非常重要的。通过get_openai_callback来获取token消耗。

from langchain.callbacks import get_openai_callback
from langchain_openai import OpenAI

llm = OpenAI(temperature=0)
with get_openai_callback() as cb:
    llm("What is the square root of 4?")

total_tokens = cb.total_tokens
print("total_tokens: ", total_tokens)

## 输出结果：total_tokens:  20
1
2
3
4
5
6
7
8
9
10
11

3. 总结

本文我们学习了LangChain的Callbacks模块，实践了各种 CallBack 的用法，知道了怎么利用LangChain进行写日志文件、Token计数等。这对于我们debug程序和监控程序的各个阶段非常重要。

如果觉得本文对你有帮助，麻烦点个赞和关注呗 ~~~

---

• 大家好，我是同学小张
• 欢迎 点赞 + 关注 声明：本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：【wpsshop博客】