—1—

Agent 技术剖析

我们来看一个例子，对于传统的程序，我们可以想象这样一个场景：一个王子需要经历3个关卡，才可以救到公主，那么王子就必须按部就班走一条确定的路线，一步步去完成这三关，才可以救到公主，他不可以跳过或者修改关卡本身。

但对于 Agent 来说，我们可以将其想象成一个刚出生的原始人类，随着大脑的日渐成熟和身体的不断发育，该人类将会逐步拥有决策能力和记忆能力，这时想象该人类处于一种饥饿状态，那么他就需要吃饭。此时，他刚好走到小河边，通过“记忆”模块，认知到河里的“鱼”是可以作为食物的，那么他此时就会巧妙利用自己身边的工具--鱼钩，进行钓鱼，然后再利用火，将鱼烤熟。第二天，他又饿了，这时他在丛林里散步，遇到了一头野猪，通过“记忆”模块，认知到“野猪”也是可以作为食物的，由于野猪的体型较大，于是他选取了更具杀伤力的长矛进行狩猎。从他这两次狩猎的经历，我们可以发现，他并不是按照预先设定好的流程，使用固定的工具去捕固定的猎物，而是根据环境的变化选择合适的猎物，又根据猎物的种类，去决策使用的狩猎工具。这一过程完美利用了自己的决策、记忆系统，并辅助利用工具，从而做出一系列反应去解决问题。

以一个数学公式来表示：Agent = LLM（决策）+ Memory（记忆）+ Tools（执行）。

通过上述的例子，相信你已经清楚认识到到 Agent 与传统程序比起来，其更加灵活，通过不同的搭配，往往会达到令人意想不到的效果，现在就用代码来实操感受一下 Agent 的实际应用方式，下面的示例代码主要实现的功能是：给予 Agent 一个题目，让 Agent 生成一篇论文。

 —2—

Agent 应用案例实战

# 初始化 agent
agent = initialize_agent(
    tools,  # 配置工具集
    llm,  # 配置大语言模型 负责决策
    agent=AgentType.OPENAI_FUNCTIONS,  # 设置 agent 类型 
    agent_kwargs=agent_kwargs,  # 设定 agent 角色
    verbose=True,
    memory=memory, # 配置记忆模式 )

第一、Tools 相关的配置介绍

首先是配置工具集 Tools，代码如下：可以看到这是一个二元数组，也就意味着本示例中的 Agent 依赖两个工具。

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")

name：工具的名称，这里是 "scrape_website"；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_chain

第二、LLM 的配置介绍

# 初始化大语言模型，负责决策
llm = ChatOpenAI(temperature=0, model="gpt-3.5-turbo-16k-0613")

这段代码初始化了一个名为 llm 的大语言模型对象，它是 ChatOpenAI 类的实例。ChatOpenAI 类用于与大语言模型（如GPT-3）进行交互，以生成决策和回答。在初始化 ChatOpenAI 对象时，提供了以下参数：

temperature：一个浮点数，表示生成文本时的温度。温度值越高，生成的文本将越随机和多样；温度值越低，生成的文本将越确定和一致。在这里设置为 0，因为本 Demo 的目的为生成一个论文，所以我们并不希望大模型有较多的可变性，而是希望生成非常确定和一致的回答。

model：一个字符串，表示要使用的大语言模型的名称。在这里，我们设置为 "gpt-3.5-turbo-16k-0613"，表示使用 GPT-3.5 Turbo 模型。

第三、Agent 类型及角色相关的配置介绍

首先来看一下 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)

在设置 agent_kwargs 时："extra_prompt_messages"：这个键对应的值是一个包含 MessagesPlaceholder 对象的列表。这个对象的 variable_name 属性设置为 "memory"，表示我们希望在构建 Agent 的提示词时，将 memory 变量的内容插入到提示词中。"system_message"：这个键对应的值是一个 SystemMessage 对象，它包含了 Agent 的角色描述和任务要求。

第四、Memory 的配置介绍

# 初始化记忆类型
memory = ConversationSummaryBufferMemory(
    memory_key="memory", return_messages=True, llm=llm, max_token_limit=300)

在设置memory 的记忆类型对象时：利用了 ConversationSummaryBufferMemory 类的实例。该类用于在与 AI 助手的对话中缓存和管理信息。在初始化这个对象时，提供了以下参数：memory_key：一个字符串，表示这个记忆对象的键。在这里设置为 "memory"；return_messages：一个布尔值，表示是否在返回的消息中包含记忆内容。在这里设置为 True，表示希望在返回的消息中包含记忆内容；llm：对应的大语言模型对象，这里是之前初始化的 llm 对象。这个参数用于指定在处理记忆内容时使用的大语言模型；max_token_limit：一个整数，表示记忆缓存的最大令牌限制。在这里设置为 300，表示希望缓存的记忆内容最多包含 300 个 token。

第五、依赖的环境包倒入以及启动主函数

这里导入所需库：这段代码导入了一系列所需的库，包括 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 的使用示例代码就描述完毕了，我们可以看到，其实 Agent 的功能就是其会自主的去选择并利用最合适的工具，从而解决问题，我们提供的 Tools 丰富，则其功能越强大。