回调处理器

    在编程领域中，回调是一个非常重要的概念。简而言之，回调是一种特殊的函数或方法，它可以被传递给另一个函数作为参数，并在适当的时候被调用。随着技术的发展，无论是为用户实时显示数据，还是为开发者提供即时的系统日志，LangChain都希望系统能够在关键时刻为开发者提供即时信息。这正是回调处理器（Callbacks)的用途。回调处理器允许开发者在特定事件发生时执行自定义操作，这在许多场景中都非常有用，例如日志记录、性能监控、流式处理等。

什么是回调处理器

    回调处理器就是一种允许开发者在特定事件发生时执行自定义操作的机制。在LangChain框架中，回调处理器是一种特殊的包装机制，其允许开发者定义一系列的方法来响应不同的生命周期事件。每当特定事件被触发时，相应的回调处理器方法就会被执行。设计回调处理器的目的是提供一个统一、模块化和可重用的机制，使开发者能够更轻松地为链组件和Aget组件添加各种回调功能。下面是使用回调处理器的几个好处。

1. 模块化与可重用性：通过定义回调处理器，可以创建一组可重用的操作，并且可以轻松地在不同的LangChain实例或应用中使用这些操作。例如，如果你有多个应用都需要流式输出到WebSocket,那么使用一个统一的WebSocketStreamingHandler可以避免重复的代码。
2. 灵活性：回调处理器提供了一种结构化的方式来响应各种事件，而不仅仅是流式输出。这意味着你可以为各种事件（如链开始、链结束、错误发生等)定义特定的逻辑。
3. 与LangChain框架中的其他组件紧密集成：回调处理器是为LangChain框架特别设计的，确保与其内部机制的兼容性和高效性。
4. 代码清晰且具有维护性：通过使用专门的回调处理器，你的代码结构会更清晰。当其他开发者查看或维护你的代码时，他们可以轻松地找到和理解回调逻辑。然而，是否使用回调处理器取决于业务需求。如果你发现直接在应用逻辑中实现特定功能更适合你的需求，那么完全可以这样做。LangChain框架中的回调处理器只是提供了一个方便、统一的工具，旨在简化开发者的工作。

回调处理器的工作流程

    LangChain的回调机制的核心在于两个参数：callbacks=[]和run_manager。run_manager参数的主要职责是管理和触发回调事件。callbacks=[]参数则是为run _manager提供具体的回调处理器列表。这意味着，在利用链组件（无论是LangChain的内置组件还是开发者自定义的组件)时，通过提供callbacks=-列表，开发者实际上是在为un_manager定义规则：“当特定事件发生时，希望这些特定的回调处理器被触发。”整个回调处理器的工作流程都发生在执行链组件的内部。为了让读者更直观地理解这个过程，请读者参见下图。整个流程包含开始执行链组件、链组件执行中、触发回调和完成4个阶段。

• 开始执行链组件阶段：此时，系统开始执行链组件，并为处理输入数据做好准备。然后检查callbacks=-参数，若传递了callbacks-=[),则系统进人下一步初始化un manager。此时，系统使用传递的callbacks=[列表来初始化run_manager,确保它包含了所有提供的回调处理器。
• 链组件执行中阶段：系统根据输人数据执行链组件。在各个关键点，如数据处理、模型调用等，系统会检查run_manager是否需要触发任何回调。
• 触发回调阶段：如果un_manager在某个执行阶段检测到需要触发的回调事件，它会按照callbacks-=[0中定义的顺序触发回调处理器。
• 完成阶段：当所有任务都完成，并且所有必要的回调都被触发后，链组件的执行就此结束。

回调处理器的使用

    开发者常用的回调处理器的使用方法是：在实例化链组件和Agent组件的时候，通过使用callbacks=[]参数传入回调处理器，整个回调处理器的工作流程都发生在执行链组件的内部。什么时候链组件被执行了，回调处理器就被调用了。链组件的调用方式有两种，一种是构造函数回调，在创建链组件或Agent组件时，通过callbacks=[]参数将回调处理器传入构造函数。这种类型的回调会在整个组件的生命周期中起作用，只要这个组件被调用，相关的回调函数就会被触发；另一种是请求回调，这是在调用组件的run()或apply()方法的callbacks=[]参数传入回调处理器。

    当运行一个链组件或者Agent组件时，回调处理器会负责在其内部定义事件触发时，调用回调处理器中内部定义的方法。例如，如果一个CallbackHandler有一个名为on_task_start的方法，那么每当链组件开始一个新任务时，CallbackManager就会自动调用这个on_task_start方法。

    如下图所示，当一个回调处理器被传递给Agent组件时，这个处理器会自动响应Agent组件在运行过程中触发的各种事件。当LLM启动时（图中①步骤），如果你实现了on_llm_start方法，则该方法会在LLM开始运行时被触发。当工具(Tool)启动时（图中②步骤)，如果你实现了on_tool_start方法，则该方法会在工具（例如数学工具或谷歌搜索工具)开始运行时被触发。当Agnt组件执行某个动作时（图中③步骤)：如果你实现了on_agent_action方法，那么每当Agent组件执行一个动作时，这个方法都会被触发。

    本质上，Agnt组件是一个特殊的复合链组件，可以简化回调处理器的使用范围为链组件。凡是链组件，皆可使用callbacks参数传递具体的回调处理器列表。（此方式仅为简化方便之用，在实际开发应用中、最多的使用场景是链组件实例化的时候。实际上LangChain的各种链组件、LLM、Chat Models、Agents和Tools这些类都可以使用回调处理器。以下是一个具体的链组件，其使用callbacks参数传递具体的回调处理器列表。自定义的回调处理器

class MyCustomHandler (BaseCallbackHandler):def on_llm_start (self,*args,kwargs):print("LLNf始运行")
my_handler= MyCustomHandler ()
# 运行链组件，传递ca11 backs参数的列表
result = my_chain.run ("some input",callbacks=[my handler])

    在上面的例子中，当链开始执行并触发on_Ilm_start事件时，MyCustomHandler中的相应方法将被调用，从而打印出“"LLM开始运行””。

自定义链组件中的回调

    链组件或Agent组件通常都有一些核心的执行方法，如_call、_generate、__run等。这些方法现在都被设计为接收一个名为run_manager的参数。这是因为run_manager允许这些组件在执行过程中与回调系统进行交互，在执行这些组件时，如果传递了callbacks=[]回调处理器列表，那么run_manager就会包含这些处理器，并负责在适当的时机触发它们。

    run_manager主要用于管理和触发回调事件。而callbacks=[]则是为这个管理器提供具体的回调处理器列表。当你在使用链组件（无论是内置的还是自定义的)时，为其提供callbacks=[]列表，实际上是在告诉run_manager：“当特定事件发生时，我希望这些回调处理器被触发。”

    run_manager被绑定到特定的执行或运行，并提供日志方法，使得在执行代码过程中的任何时刻，都可以触发回调事件，如生成新的提示词、完成执行等。因此，可以说任何链组件都可以使用callbacks关键字参数，这是因为这些组件的设计已经考虑到了与回调系统的交互，只要它们的核心方法（如call、_generate、_run等)被调用，run_manager就可以被传递进去，从而使得回调系统可以在执行过程中被触发。分析以下自定义链：

class MyCustomChain (Chain):#···[代码简化]def call(self,inputs:Dict(str,Any],run_manager:Optional[CallbackManagerForChainRun]None,)->Dict[str,str]:#...[代码简化]#当调用语言模型或其他链时，应传递一个回调管理器。response =self.llm.generate_prompt([prompt_value],callbacks=run_manager.get_child() if run_manager else None#如果需要记录此运行的信息，则可以通过调用'run manager'中的方法来做到。if run_manager:run_manager.on_text("记录此次运行的相关信息")return (self.output_key:response.generations[0][0].text)

    在上述代码中，run_manager被作为一个参数传递给call方法，允许开发者在链组件的执行中获取实时反馈，并进行日志记录。这个yCustomChain在被执行时，可以提供callbacks-=[]列表，这样就完成了整个回调工作流程。

handler= MyCallbackHandler ()
chain= MyCustomChain(llm=llm,prompt=prompt,callbacks=[handler])
chain.run(…)

内置回调处理器

    为了简化开发过程，LangChain提供了一系列内置的回调处理器，比如运行一个Agent组件，它的底层都使用到了StdOutCallbackHandler。.例如在下面代码中，设置verbose=True,在运行Agent组件时，也就是事件发生时，会将Agent组件的相关信息打印到标准输出（通常是控制台或命令行界面中)。在程序开发过程中，即时的反馈对于开发者理解程序的运行状态和识别潜在问题是至关重要的。LangChain通过其StdOutCallbackHandler为开发者提供了这一功能。

import os
os.environ["OPENAI API KEY"]="填人你的密钥"
from langchain.agents import load tools
from langchain.agents import initialize agent
from langchain.agents import AgentType
from langchain.llms import OpenAI
llm =OpenAI()
tools =load_tools (["llm-math"],llm=llm)
agent=initialize_agent(tools,llm,agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,verbose=True)
agent.run ("9+7")

    在命令行界面中就可以看到以Entering new AgentExecutor chain…为开始，以Finished chain为结尾的标准输出。这是因为当运行Agent组件并启用verbose=True时，StdOutCallbackHandler将被自动激活，其将Agent组件的活动实时打印到标准输出。这为开发者提供了即时的反馈，帮助他们了解Agent组件的工作情况。

Entering new AgentExecutor chain...
I need to add two numbers together
Action:Calculator
Action Input:9+7
Observation:Answer:16
Thought:I now know the final answer
Final Answer:9+7= 16
Finished chain.
"9+7=16"

    这一系列的输出不仅告诉Agent组件已经开始运行，并完成了其任务，还详细地展示了Agent组件在执行过程中的所有操作和决策。例如，从输出中可以清楚地看到Agt组件收到的任务是“将两个数字加在一起”，接下来它决定采取的行动是使用“计算器”，并为此提供了具体的输人“9+7”。之后，Agnt组件给出了观察结果“答案：16”，并在思考后给出了最终答案“9+7=16”。这样的即时反馈对开发者来说意义重大。首先，它帮助开发者即时了解Agent组件的决策过程和操作顺序。当Agt组件的输出与预期不符时，开发者可以通过这些详细的反馈迅速定位问题所在，而无须深入底层代码中进行调试。其次，这也为开发者提供了一个观察和测试Aget组件在不同情境下的行为的机会，从而优化和完善其功能。

    然而，如果你更倾向于只获取Aget组件的最终执行结果，而不关心其内部的执行过程，那么你可以设置verbose=-False。这样，命令行界面中只会显示大语言模型的最终答案，如’9+7=16’，让输出更为简洁。这对于那些希望集成LangChain到他们的应用中，并希望只展示关键信息的开发者来说是非常有用的。

9+7=16

    还可以给链组件和Aget组件添加内置或者自定义的回调处理器。比如给一个基础的链组件LLMChain添加一个内置回调处理器：StdOutCallbackHandler。.可以先创建一个链组件。在初始化LLMChain的时候，这个链组件没有设置内置回调处理器，也不能设置verbose=True。

from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain
from langchain.llms import OpenAI
llm_chain=LLMChain(llm=OpenAI (),prompt=PromptTemplate.from_template ("(input)"))
llm= chain.run('上海的旅游景点有哪些？)

    运行LLMChain后，大语言模型回答的是：

上海的旅游景点有：\n\n1,上海迪士尼乐园\n2.东方明珠塔1n3.南京路步行街1n4.上海外滩\n5.上海野生动物园\n6,外白渡桥1n7.南京路商业街\n8.上海科技馆\n9.上海老城隍庙\n10.上海博物馆1n11.上海浦江夜游1n12.上海水上乐园\n13.上海徐汇森林公园\n14.上海金茂大厦An...

    如果想要监控这个链组件，则可以添加一些回调逻辑，比如想要命令行输出这个链组件运行的相关信息，则可以给这个链组件增加一个回调处理器。这里导入内置的StdOutCallbackHandler,.并且创建它的实例handler_1。

from langchain.callbacks import StdoutCallbackHandler
handler_1 =StdoutCallbackHandler (）

    回调处理器的使用很简单，将handler_1作为回调处理器通过callbacks=-[handler_1]参数添加到LLMChain中，以实现对链组件运行状态的监控和命令行输出。这意味着可以添加多个回调处理器，完成不同的任务逻辑。

llm_chain =LLMChain(llm=OpenAI (),callbacks=[handler_1],prompt=PromptTemplate.from template("(input )")
llm_chain.run('上海的旅游景点有哪些？)

    这样就给一个LLMChain添加了一个回调处理器，内置的StdOutCallbackHandler完成的是标准的链组件的打印输出。当继续运行这个链组件时，在命令行界面中就可以看到以Entering new AgentExecutor chain…为开始，以Finished chain为结尾的标准输出。

>Entering new LLMChain chain...
Prompt after formatting:
上海的旅游景点有哪些？
Finished chain.
\n\n上海的旅游景点有：1n\n1,东方明珠广播电视塔n2.豫园n3.外滩\n4.南京路步行街
n5.上海野...\n\n

自定义回调处理器

    在程序开发过程中，常常会遇到一些特定的需求，例如为每个用户请求单独创建日志文件，或在发生某一个关键事件时及时发送通知。这些需求超出了内置回调处理器的能力范围，而自定义回调处理器在此时发挥了重要的作用。为了真正充分利用回调处理器，需要设计并实现自己的处理器。每一个链，从它被创建到最终被销毁，都会经历多个关键阶段。在链的生命周期中，每一个阶段都可能会触发某些事件。为了确保在恰当的时机介人并执行相应的操作，需要深入了解每一个阶段并对其进行精确的控制。这样，当链处于某个特定的阶段时，就可以在相应的回调处理器方法中执行预定的代码。

    在某些情况下，可能只希望在特定的请求中执行特定的代码，而不是在链的整个生命周期中。这种需求可以通过请求回调来实现。请求回调为实现这种特定请求提供了灵活性。例如，在一个场景中，你只希望在某个特定的请求中记录日志，而不是在所有的请求中都这样做。通过使用请求回调，你可以轻松地达到这个目的。不过，无论如何设计和使用回调处理器，最关键的始终是理解自己的业务逻辑需求。只有清晰地知道自己想要达到的目的，才能在正确的回调处理器方法中插入合适的代码，确保在合适的时机执行正确的操作。自定义回调处理器提供了一个强大的框架，但如何充分利用这个框架，最终取决于对业务需求的理解和技术的运用。

    在编写自定义回调处理器之前，了解其背后的基础类是非常重要的。BaseCallbackHandler正是这样一个核心类，其提供了一个强大而灵活的框架，可以轻松地响应和处理各种事件。这个类定义了一系列的方法，每一个都与LangChain中的一个特定事件相对应。只有深入理解了这些事件和方法，才能有效地为应用编写自定义的回调处理器。

1. LLM事件on_Ilm_start:当LLM启动并开始处理请求时，这个方法会被调用。它提供了一个机会，例如，初始化某些资源或记录开始时间。on_llm_new_token:当LLM生成一个新的令牌时，这个方法就会被执行。它在流式处理中特别有用，其允许实时捕获和处理每一个生成的令牌。on_Ilm_end:当LLM完成任务并生成了完整的输出时，这个方法就会被调用。它提供了一个机会进行清理操作或记录任务的完成时间。on_llm_error:如果在LLM处理过程中发生任何错误，则这个方法将会被执行。可以在这个方法中添加错误日志或执行其他的错误处理操作。
2. 聊天模型事件on_chat_model_start:这个方法在Chat Model类模型包装器开始工作时被调用。它提供了一个机会进行初始化操作或其他准备工作。
3. 链事件on _hain_start:：当链开始执行时，这个方法会被调用。可以在这个方法中进行一些初始化操作。on_chain_end:在链完成所有任务后，这个方法就会被执行。它提供了一个机会进行清理操作或收集结果。on_chain_error:如果链在执行过程中遇到错误，则这个方法将会被调用。它许进行错误处理或日志记录。
4. 工具事件on_tool_start:当工具开始执行任务时，这个方法就会被调用。on_tool_end:在工具成功完成任务后，这个方法就会被执行。on_tool _error:如果工具在执行过程中发生错误，则这个方法将会被执行。
5. 其他事件on_text:当需要处理任意文本时，这个方法会被调用。它提供了一个机会对文本进行处理或分析。on_agent_action:当代理执行某个特定的操作时，这个方法就会被执行。on_agent_finish:在代理完成所有操作后，这个方法就会被调用。

    BaseCallbackHandler是一个强大的基础类，使开发者可以轻松地定义和处理各种事件。深人理解这个基础类是编写自定义回调处理器的关键。如果你要自定义自己的回调处理器，则可以继承BaseCallbackHandler并重写你需要的方法。例如，如果你想在大语言模型开始输出时打印一条消息，则可以这样做：

class MyCallbackHandler(BaseCallbackHandler):def on_llm_start (self,serialized,prompts,kwargs):print ("LLM has started!")

    一旦你自定义了自己的回调处理器，就可以将其传递给LLMChain,以便在相应的事件发生时执行你的方法。例如：

handler =MyCallbackHandler()
chain =LLMChain(llm=llm,prompt=prompt,callbacks=(handler])