接（上）
七、Agents
Agents这一模块在langchain的使用过程中也是十分重要的，官方文档是这样定义它的“The core idea of agents is to use a language model to choose a sequence of actions to take. In chains, a sequence of actions is hardcoded (in code). In agents, a language model is used as a reasoning engine to determine which actions to take and in which order.”也就是说，在使用Agents时，其行为以及行为的顺序是由LLM的推理机制决定的，并不是像传统的程序一样，由核心代码预定义好去运行的。
举一个例子来对比一下，对于传统的程序，我们可以想象这样一个场景：一个王子需要经历3个关卡，才可以救到公主，那么王子就必须按部就班的走一条确定的路线，一步步去完成这三关，才可以救到公主，他不可以跳过或者修改关卡本身。但对于Agents来说，我们可以将其想象成一个刚出生的原始人类，随着大脑的日渐成熟和身体的不断发育，该人类将会逐步拥有决策能力和记忆能力，这时想象该人类处于一种饥饿状态，那么他就需要吃饭。此时，他刚好走到小河边，通过“记忆”模块，认知到河里的“鱼”是可以作为食物的，那么他此时就会巧妙的利用自己身边的工具——鱼钩，进行钓鱼，然后再利用火，将鱼烤熟。第二天，他又饿了，这时他在丛林里散步，遇到了一头野猪，通过“记忆”模块，认知到“野猪”也是可以作为食物的，由于野猪的体型较大，于是他选取了更具杀伤力的长矛进行狩猎。从他这两次狩猎的经历，我们可以发现，他并不是按照预先设定好的流程，使用固定的工具去捕固定的猎物，而是根据环境的变化选择合适的猎物，又根据猎物的种类，去决策使用的狩猎工具。这一过程完美的利用了自己的决策、记忆系统，并辅助利用工具，从而做出一系列反应去解决问题。以一个数学公式来表示，可以说Agents=LLM（决策）+Memory（记忆）+Tools（执行）。
通过上述的例子，相信你已经清楚的认知到Agents与传统程序比起来，其更加灵活，通过不同的搭配，往往会达到令人意想不到的效果，现在就用代码来实操感受一下Agents的实际应用方式，下文的示例代码主要实现的功能是——给予Agent一个题目，让Agent生成一篇论文。
在该示例中，我们肯定是要示例化Agents，示例化一个Agents时需要关注上文中所描述的它的三要素：LLM、Memory和tools，其代码如下：
# 初始化 agent
agent = initialize_agent(
tools, # 配置工具集
llm, # 配置大语言模型 负责决策
agent=AgentType.OPENAI_FUNCTIONS, # 设置 agent 类型
agent_kwargs=agent_kwargs, # 设定 agent 角色
verbose=True,
memory=memory, # 配置记忆模式 )7.1tools相关的配置介绍
首先是配置工具集tools，如下列代码，可以看到这是一个二元数组，也就意味着本示例中的Agents依赖两个工具。
from langchain.agents import initialize_agent, Tool
tools = [
Tool(
name="Search",
func=search,
description="useful for when you need to answer questions about current events, data. You should ask targeted questions"
),
ScrapeWebsiteTool(),
]先看第一个工具：在配置工具时，需要声明工具依赖的函数，由于该示例实现的功能为依赖网络收集相应的信息，然后汇总成一篇论文，所以创建了一个search函数，这个函数用于调用Google搜索。它接受一个查询参数，然后将查询发送给Serper API。API的响应会被打印出来并返回。
# 调用 Google search by Serper
def search(query):
serper_google_url = os.getenv("SERPER_GOOGLE_URL")
payload = json.dumps({
"q": query
})
headers = {
'X-API-KEY': serper_api_key,
'Content-Type': 'application/json'
}
response = requests.request("POST", serper_google_url, headers=headers, data=payload)
print(f'Google 搜索结果: \n {response.text}')
return response.text再来看一下所依赖的第二个工具函数，这里用了另一种声明工具的方式Class声明—— ScrapeWebsiteTool()，它有以下几个属性和方法：
class ScrapeWebsiteTool(BaseTool):
name = "scrape_website"
description = "useful when you need to get data from a website url, passing both url and objective to the function; DO NOT make up any url, the url should only be from the search results"
args_schema: Type[BaseModel] = ScrapeWebsiteInput
def _run(self, target: str, url: str):
return scrape_website(target, url)
def _arun(self, url: str):
raise NotImplementedError("error here")1.name：工具的名称，这里是 "scrape_website"。 2.description：工具的描述。 args_schema：工具的参数模式，这里是 ScrapeWebsiteInput 类，表示这个工具需要的输入参数，声明代码如下，这是一个基于Pydantic的模型类，用于定义 scrape_website 函数的输入参数。它有两个字段：target 和 url，分别表示用户给agent的目标和任务以及需要被爬取的网站的URL。
class ScrapeWebsiteInput(BaseModel):
"""Inputs for scrape_website"""
target: str = Field(
description="The objective & task that users give to the agent")
url: str = Field(description="The url of the website to be scraped")run 方法：这是工具的主要执行函数，它接收一个目标和一个URL作为参数，然后调用 scrape website 函数来爬取网站并返回结果。scrape_website 函数根据给定的目标和URL爬取网页内容。首先，它发送一个HTTP请求来获取网页的内容。如果请求成功，它会使用BeautifulSoup库来解析HTML内容并提取文本。如果文本长度超过5000个字符，它会调用 summary 函数来对内容进行摘要。否则，它将直接返回提取到的文本。其代码如下：
# 根据 url 爬取网页内容，给出最终解答
# target ：分配给 agent 的初始任务
# url ： Agent 在完成以上目标时所需要的URL，完全由Agent自主决定并且选取，其内容或是中间步骤需要，或是最终解答需要
def scrape_website(target: str, url: str):
print(f"开始爬取： {url}...")
headers = {
'Cache-Control': 'no-cache',
'Content-Type': 'application/json',
}
payload = json.dumps({
"url": url
})
post_url = f"https://chrome.browserless.io/content?token={browserless_api_key}"
response = requests.post(post_url, headers=headers, data=payload)
# 如果返回成功
if response.status_code == 200:
soup = BeautifulSoup(response.content, "html.parser")
text = soup.get_text()
print("爬取的具体内容:", text)
# 控制返回内容长度，如果内容太长就需要切片分别总结处理
if len(text) > 5000:
# 总结爬取的返回内容
output = summary(target, text)
return output
else:
return text
else:
print(f"HTTP请求错误，错误码为{response.status_code}")从上述代码中我们可以看到其还依赖一个summary 函数，用此函数解决内容过长的问题，这个函数使用Map-Reduce方法对长文本进行摘要。它首先初始化了一个大语言模型（llm），然后定义了一个大文本切割器（text_splitter）。接下来，它创建了一个摘要链（summary_chain），并使用这个链对输入文档进行摘要。
# 如果需要处理的内容过长，先切片分别处理，再综合总结
# 使用 Map-Reduce 方式
def summary(target, content):
# model list ： https://platform.openai.com/docs/models
# gpt-4-32k gpt-3.5-turbo-16k-0613
llm = ChatOpenAI(temperature=0, model="gpt-3.5-turbo-16k-0613")
# 定义大文本切割器
# chunk_overlap 是一个在使用 OpenAI 的 GPT-3 或 GPT-4 API 时可能会遇到的参数，特别是需要处理长文本时。
# 该参数用于控制文本块（chunks）之间的重叠量。
# 上下文维护：重叠确保模型在处理后续块时有足够的上下文信息。
# 连贯性：它有助于生成更连贯和一致的输出，因为模型可以“记住”前一个块的部分内容。
text_splitter = RecursiveCharacterTextSplitter(
separators=["\n\n", "\n"], chunk_size=5000, chunk_overlap=200)
docs = text_splitter.create_documents([content])
map_prompt = """
Write a summary of the following text for {target}:
"{text}"
SUMMARY:
"""
map_prompt_template = PromptTemplate(
template=map_prompt, input_variables=["text", "target"])
summary_chainarun 方法：这是一个异步版本的 run 方法，这里没有实现，如果调用会抛出一个 NotImplementedError 异常。
7.2LLM的配置介绍
# 初始化大语言模型，负责决策
llm = ChatOpenAI(temperature=0, model="gpt-3.5-turbo-16k-0613")这段代码初始化了一个名为 llm 的大语言模型对象，它是 ChatOpenAI 类的实例。ChatOpenAI 类用于与大语言模型（如GPT-3）进行交互，以生成决策和回答。在初始化 ChatOpenAI 对象时，提供了以下参数：
1.temperature：一个浮点数，表示生成文本时的温度。温度值越高，生成的文本将越随机和多样；温度值越低，生成的文本将越确定和一致。在这里设置为 0，因为本demo的目的为生成一个论文，所以我们并不希望大模型有较多的可变性，而是希望生成非常确定和一致的回答。 2.model：一个字符串，表示要使用的大语言模型的名称。在这里，我们设置为 "gpt-3.5-turbo-16k-0613"，表示使用 GPT-3.5 Turbo 模型。
7.3Agent类型及角色相关的配置介绍
首先来看一下AgentType这个变量的初始化，这里是用来设置agent类型的一个参数，具体可以参考官网：AgentType
可以看到官网里列举了7中agent类型，可以根据自己的需求进行选择，在本示例中选用的是第一种类型OpenAi functions。此外，还要设定agent角色以及记忆模式：
# 初始化agents的详细描述
system_message = SystemMessage(
content="""您是一位世界级的研究员，可以对任何主题进行详细研究并产生基于事实的结果；
您不会凭空捏造事实，您会尽最大努力收集事实和数据来支持研究。
请确保按照以下规则完成上述目标：
1/ 您应该进行足够的研究，尽可能收集关于目标的尽可能多的信息
2/ 如果有相关链接和文章的网址，您将抓取它以收集更多信息
3/ 在抓取和搜索之后，您应该思考“根据我收集到的数据，是否有新的东西需要我搜索和抓取以提高研究质量？”如果答案是肯定的，继续；但不要进行超过5次迭代
4/ 您不应该捏造事实，您只应该编写您收集到的事实和数据
5/ 在最终输出中，您应该包括所有参考数据和链接以支持您的研究；您应该包括所有参考数据和链接以支持您的研究
6/ 在最终输出中，您应该包括所有参考数据和链接以支持您的研究；您应该包括所有参考数据和链接以支持您的研究"""
)
# 初始化 agent 角色模板
agent_kwargs = {
"extra_prompt_messages": [MessagesPlaceholder(variable_name="memory")],
"system_message": system_message,
}
# 初始化记忆类型
memory = ConversationSummaryBufferMemory(
memory_key="memory", return_messages=True, llm=llm, max_token_limit=300)1️⃣在设置agent_kwargs时："extra_prompt_messages"：这个键对应的值是一个包含 MessagesPlaceholder 对象的列表。这个对象的 variable_name 属性设置为 "memory"，表示我们希望在构建 agent 的提示时，将 memory 变量的内容插入到提示中。"system_message"：这个键对应的值是一个 SystemMessage 对象，它包含了 agent 的角色描述和任务要求。
7.4Memory的配置介绍
# 初始化记忆类型
memory = ConversationSummaryBufferMemory(
memory_key="memory", return_messages=True, llm=llm, max_token_limit=300)在设置 memory 的记忆类型对象时：利用了 ConversationSummaryBufferMemory 类的实例。该类用于在与AI助手的对话中缓存和管理信息。在初始化这个对象时，提供了以下参数：1.memory_key：一个字符串，表示这个记忆对象的键。在这里设置为 "memory"。2.return_messages：一个布尔值，表示是否在返回的消息中包含记忆内容。在这里设置为 True，表示希望在返回的消息中包含记忆内容。3.llm：对应的大语言模型对象，这里是之前初始化的 llm 对象。这个参数用于指定在处理记忆内容时使用的大语言模型。4。max_token_limit：一个整数，表示记忆缓存的最大令牌限制。在这里设置为 300，表示希望缓存的记忆内容最多包含 300 个token。
7.5依赖的环境包倒入以及启动主函数
这里导入所需库：这段代码导入了一系列所需的库，包括os、dotenv、langchain相关库、requests、BeautifulSoup、json和streamlit。
import os
from dotenv import load_dotenv
from langchain import PromptTemplate
from langchain.agents import initialize_agent, Tool
from langchain.agents import AgentType
from langchain.chat_models import ChatOpenAI
from langchain.prompts import MessagesPlaceholder
from langchain.memory import ConversationSummaryBufferMemory
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains.summarize import load_summarize_chain
from langchain.tools import BaseTool
from pydantic import BaseModel, Field
from langchain.schema import SystemMessage
from typing import Type
from bs4 import BeautifulSoup
import requests
import json
import streamlit as st
# 加载必要的参数
load_dotenv()
serper_api_key=os.getenv("SERPER_API_KEY")
browserless_api_key=os.getenv("BROWSERLESS_API_KEY")
openai_api_key=os.getenv("OPENAI_API_KEY")main 函数：这是streamlit应用的主函数。它首先设置了页面的标题和图标，然后创建了一些header，并提供一个文本输入框让用户输入查询。当用户输入查询后，它会调用agent来处理这个查询，并将结果显示在页面上。
def main():
st.set_page_config(page_title="AI Assistant Agent", page_icon=":dolphin:")
st.header("LangChain 实例讲解 3 -- Agent", divider='rainbow')
st.header("AI Agent :blue[助理] :dolphin:")
query = st.text_input("请提问题和需求：")
if query:
st.write(f"开始收集和总结资料 【 {query}】 请稍等")
result = agent({"input": query})
st.info(result['output'])至此Agent的使用示例代码就描述完毕了，我们可以看到，其实Agents的功能就是其会自主的去选择并利用最合适的工具，从而解决问题，我们提供的Tools越丰富，则其功能越强大。
八、Callbacks
Callbacks对于程序员们应该都不陌生，就是一个回调函数，这个函数允许我们在LLM的各个阶段使用各种各样的“钩子”，从而达实现日志的记录、监控以及流式传输等功能。在Langchain中，该回掉函数是通过继承 BaseCallbackHandler 来实现的，该接口对于每一个订阅事件都声明了一个回掉函数。它的子类也就可以通过继承它实现事件的处理。如官网所示：
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."""这个类包含了一系列方法，这些方法在 langchain 的不同阶段被调用，以便在处理过程中执行自定义操作。参考源码BaseCallbackHandler：
on_llm_start: 当大语言模型（LLM）开始运行时调用。 on_chat_model_start: 当聊天模型开始运行时调用。 on_llm_new_token: 当有新的LLM令牌时调用。仅在启用流式处理时可用。 on_llm_end: 当LLM运行结束时调用。 on_llm_error: 当LLM出现错误时调用。 on_chain_start: 当链开始运行时调用。 on_chain_end: 当链运行结束时调用。 on_chain_error: 当链出现错误时调用。 on_tool_start: 当工具开始运行时调用。 on_tool_end: 当工具运行结束时调用。 on_tool_error: 当工具出现错误时调用。 on_text: 当处理任意文本时调用。 on_agent_action: 当代理执行操作时调用。 on_agent_finish: 当代理结束时调用。
8.1基础使用方式StdOutCallbackHandler
StdOutCallbackHandler 是 LangChain 支持的最基本的处理器，它继承自 BaseCallbackHandler。这个处理器将所有回调信息打印到标准输出，对于调试非常有用。以下是如何使用 StdOutCallbackHandler 的示例：
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")在这个示例中，我们首先从 langchain.callbacks 模块导入了 StdOutCallbackHandler 类。然后，创建了一个 StdOutCallbackHandler 实例，并将其赋值给变量 handler。接下来，导入了 LLMChain、OpenAI 和 PromptTemplate 类，并创建了相应的实例。在创建 LLMChain 实例时，将 callbacks 参数设置为一个包含 handler 的列表。这样，当链运行时，所有的回调信息都会被打印到标准输出。最后，使用 chain.run() 方法运行链，并传入参数 name="Super Mario"。在链运行过程中，所有的回调信息将被 StdOutCallbackHandler 处理并打印到标准输出。
8.2自定义回调处理器
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)这个示例展示了如何通过继承 BaseCallbackHandler 来实现自定义的回调处理器。在这个例子中，创建了一个名为 TimerHandler 的自定义处理器，它用于跟踪 Chain 或 LLM 交互的起止时间，并统计每次交互的处理耗时。从 langchain.callbacks.base 模块导入 BaseCallbackHandler 类。导入 time 模块，用于处理时间相关操作。
定义 TimerHandler 类，继承自 BaseCallbackHandler。在 TimerHandler 类的 init 方法中，初始化 previous_ms 和 durations 属性。定义 current_ms 方法，用于返回当前时间的毫秒值。重写 on_chain_start、on_chain_end、on_llm_start 和 on_llm_end 方法，在这些方法中记录开始和结束时间，并计算处理耗时。接下来，我们创建了一个 OpenAI 实例、一个 TimerHandler 实例以及一个 PromptTemplate 实例。然后，我们创建了一个使用 timerHandler 作为回调处理器的 LLMChain 实例。最后，我们运行了两次Chain，分别查询蓝色和紫色的十六进制代码。在链运行过程中，TimerHandler 将记录每次交互的处理耗时，并将其添加到 durations 列表中。
输出如下：
8.3callbacks使用场景总结
1️⃣通过构造函数参数 callbacks 设置。这种方式可以在创建对象时就设置好回调处理器。例如，在创建 LLMChain 或 OpenAI 对象时，可以通过 callbacks 参数设置回调处理器。
timerHandler = TimerHandler()
llm = OpenAI(callbacks=[timerHandler])
response = llm.predict("What is the HEX code of color BLACK?") print(response) 在这里构建llm的时候我们就直接指定了构造函数。
2️⃣通过运行时的函数调用。这种方式可以在运行时动态设置回调处理器，如在Langchain的各module如Model，Agent，Tool，以及 Chain的请求执行函数设置回调处理器。例如，在调用 LLMChain 的 run 方法或 OpenAI 的 predict 方法时，可以通过 callbacks 参数设置回调处理器。以OpenAI 的 predict 方法为例：
timerHandler = TimerHandler()
llm = OpenAI()
response = llm.predict("What is the HEX code of color BLACK?", callbacks=[timerHandler])
print(response) 这段代码首先创建一个 TimerHandler 实例并将其赋值给变量 timerHandler。然后创建一个 OpenAI 实例并将其赋值给变量 llm。调用 llm.predict() 方法，传入问题 "What is the HEX code of color BLACK?"，并通过 callbacks 参数设置回调处理器 timerHandler。
两种方法的主要区别在于何时和如何设置回调处理器。
构造函数参数 callbacks 设置：在创建对象（如 OpenAI 或 LLMChain）时，就通过构造函数的 callbacks 参数设置回调处理器。这种方式的优点是你可以在对象创建时就确定回调处理器，后续在使用该对象时，无需再次设置。但如果在后续的使用过程中需要改变回调处理器，可能需要重新创建对象。
通过运行时的函数调用：在调用对象的某个方法（如 OpenAI 的 predict 方法或 LLMChain 的 run 方法）时，通过该方法的 callbacks 参数设置回调处理器。这种方式的优点是你可以在每次调用方法时动态地设置回调处理器，更加灵活。但每次调用方法时都需要设置，如果忘记设置可能会导致回调处理器不生效。
在实际使用中，可以根据需要选择合适的方式。如果回调处理器在对象的整个生命周期中都不会变，可以选择在构造函数中设置；如果回调处理器需要动态变化，可以选择在运行时的函数调用中设置。
九、总结
至此，Langchain的各个模块使用方法就已经介绍完毕啦，相信你已经感受到Langchain的能力了～不难发现，LangChain 是一个功能十分强大的AI语言处理框架，它将Model IO、Retrieval、Memory、Chains、Agents和Callbacks这六个模块组合在一起。Model IO负责处理AI模型的输入和输出，Retrieval模块实现了与向量数据库相关的检索功能，Memory模块则负责在对话过程中存储和重新加载历史对话记录。Chains模块充当了一个连接器的角色，将前面提到的模块连接起来以实现更丰富的功能。Agents模块通过理解用户输入来自主调用相关工具，使得应用更加智能化。而Callbacks模块则提供了回调机制，方便开发者追踪调用链路和记录日志，以便更好地调试LLM模型。总之，LangChain是一个功能丰富、易于使用的AI语言处理框架，它可以帮助开发者快速搭建和优化AI应用。本文只是列举了各模块的核心使用方法和一些示例demo，建议结合本文认真阅读一遍官方文档会更加有所受益～
作者：腾讯程序员
文章来源：腾讯技术工程
推荐阅读
更多腾讯AI相关技术干货，请关注专栏腾讯技术工程 欢迎添加极术小姐姐微信（id:aijishu20)加入技术交流群，请备注研究方向。
