前言

大语言模型一般只会在自己学习的知识中进行搜索，无法根据每一步的输出内容来决定下一步的 action, 这个时候就需要用到 Agent 技术，Agent 其实就是封装了一系列的 tools 工具，当我们给 LLM 一个复杂的任务时，它会将该任务拆解为一步一步的，然后根据对应的上下文内容，选择使用哪一个或多个工具。

其中现在最流行的 Langchain 的核心之一就是封装了很多的 Agent 可以让我们调用，接下来我将演示如何通过 AgentExecutor 构造一个 Langchain Agent 并使用它。

前提条件

• 确保开通了通义千问 API key 和 向量检索服务 API KEY
• 安装依赖：
  pip install langchain, pip install langchain-community, pip install dashVector, pip install dashscope

定义 tools

import os

from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores.dashvector import DashVector
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain.tools.retriever import create_retriever_tool
from langchain_community.embeddings.dashscope import DashScopeEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.messages import HumanMessage

class TestAgent:

    def __init__(self):
        """在阿里云控制台开通 DASHSCOPE 和 DASHVECTOR"""
        os.environ["DASHSCOPE_API_KEY"] = ""
        os.environ["DASHVECTOR_API_KEY"] = ""
        os.environ["DASHVECTOR_ENDPOINT"] = ""

        # 一系列 tool 让 langchain 选择使用
        self.tools = []

    def build_retriever_tool(self):
        """爬取网站内容切分成 chunks 嵌入向量数据库"""
        loader = WebBaseLoader("https://docs.smith.langchain.com/overview")
        docs = loader.load()
        documents = RecursiveCharacterTextSplitter(
            chunk_size=1000, chunk_overlap=200
        ).split_documents(docs)
        vector = DashVector.from_documents(documents, DashScopeEmbeddings(), collection_name="langchain")
        retriever = vector.as_retriever()
        retriever_tool = create_retriever_tool(
            retriever,
            "langsmith_search",
            # 注意这里的 description 很重要，它描述的意思决定了 langchain 是否根据当前上下文选择该工具
            # 因此，description 要尽量描述的和 tool 所做内容吻合
            "Search for information about LangSmith. For any questions about LangSmith, you must use this tool!",
        )
        self.tools.append(retriever_tool)

• 这里我们定义了一个关于 retriever_tool: 关于向 LLM 询问 LangSmith 的问题，将使用到该工具
• 注意 description 很重要
  

使用大模型调用 tools

接下来我们向阿里云的通义千问大模型询问关于 LangSmith 的相关信息，它将会调用我们的 tools。

import os

from langchain.agents import AgentExecutor
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores.dashvector import DashVector
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain.tools.retriever import create_retriever_tool
from langchain_community.embeddings.dashscope import DashScopeEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.messages import HumanMessage

class TestAgent:

    def __init__(self):
        """在阿里云控制台开通 DASHSCOPE 和 DASHVECTOR"""
        os.environ["DASHSCOPE_API_KEY"] = ""
        os.environ["DASHVECTOR_API_KEY"] = ""
        os.environ["DASHVECTOR_ENDPOINT"] = ""

        self.model = ChatTongyi()

        # 一系列 tool 让 langchain 选择使用
        self.tools = []

    def build_retriever_tool(self):
        """爬取网站内容切分成 chunks 嵌入向量数据库"""
        loader = WebBaseLoader("https://docs.smith.langchain.com/overview")
        docs = loader.load()
        documents = RecursiveCharacterTextSplitter(
            chunk_size=1000, chunk_overlap=200
        ).split_documents(docs)
        vector = DashVector.from_documents(documents, DashScopeEmbeddings(), collection_name="langchain")
        retriever = vector.as_retriever()
        retriever_tool = create_retriever_tool(
            retriever,
            "langsmith_search",
            # 注意这里的 description 很重要，它描述的意思决定了 langchain 是否根据当前上下文选择该工具
            # 因此，description 要尽量描述的和 tool 所做内容吻合
            "Search for information about LangSmith. For any questions about LangSmith, you must use this tool!",
        )
        self.tools.append(retriever_tool)

    def invoke_llm_with_tools(self, query=None):
        """
        向 LLM 提出问题，让大模型结合自己所学习的内容和我们自己构造的 tools 来回答
        """
        model_with_tools = self.model.bind_tools(self.tools)

        response = model_with_tools.invoke([HumanMessage(content=query)])
        print(f"ContentString: {response.content}")
        # 可以查看 LLM 在回答我们的问题的时候用到了哪些 tool
        print(f"ToolCalls: {response.tool_calls}")

    def create_agent_executor(self):
        """通过 AgentExecutor 将我们的 tools 封装为 Agent"""
        from langchain import hub
        from langchain.agents import create_tool_calling_agent
        from langchain.agents import AgentExecutor

        prompt = hub.pull("hwchase17/openai-functions-agent")
        agent = create_tool_calling_agent(self.model, self.tools, prompt)
        self.agent_executor = AgentExecutor(agent=agent, tools=self.tools)

if __name__ == '__main__':
    a = TestAgent()
    a.build_retriever_tool()
    a.invoke_llm_with_tools("告诉我关于 LangSmith 的相关信息")

封装 Agent 调用 tools

可以通过 AgentExecutor 将我们的 tools 封装为 Agent, 然后通过 Agent 执行查询。

import os

from langchain.agents import AgentExecutor
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores.dashvector import DashVector
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain.tools.retriever import create_retriever_tool
from langchain_community.embeddings.dashscope import DashScopeEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.messages import HumanMessage

class TestAgent:

    def __init__(self):
        """在阿里云控制台开通 DASHSCOPE 和 DASHVECTOR"""
        os.environ["DASHSCOPE_API_KEY"] = ""
        os.environ["DASHVECTOR_API_KEY"] = ""
        os.environ["DASHVECTOR_ENDPOINT"] = ""

        self.model = ChatTongyi()

        # 一系列 tool 让 langchain 选择使用
        self.tools = []

        self.agent_executor = None

    def build_retriever_tool(self):
        """爬取网站内容切分成 chunks 嵌入向量数据库"""
        loader = WebBaseLoader("https://docs.smith.langchain.com/overview")
        docs = loader.load()
        documents = RecursiveCharacterTextSplitter(
            chunk_size=1000, chunk_overlap=200
        ).split_documents(docs)
        vector = DashVector.from_documents(documents, DashScopeEmbeddings(), collection_name="langchain")
        retriever = vector.as_retriever()
        retriever_tool = create_retriever_tool(
            retriever,
            "langsmith_search",
            # 注意这里的 description 很重要，它描述的意思决定了 langchain 是否根据当前上下文选择该工具
            # 因此，description 要尽量描述的和 tool 所做内容吻合
            "Search for information about LangSmith. For any questions about LangSmith, you must use this tool!",
        )
        self.tools.append(retriever_tool)

    def invoke_llm_with_tools(self, query=None):
        """
        向 LLM 提出问题，让大模型结合自己所学习的内容和我们自己构造的 tools 来回答
        """
        model_with_tools = self.model.bind_tools(self.tools)

        response = model_with_tools.invoke([HumanMessage(content=query)])
        print(f"ContentString: {response.content}")
        # 可以查看 LLM 在回答我们的问题的时候用到了哪些 tool
        print(f"ToolCalls: {response.tool_calls}")

    def create_agent_executor(self):
        """通过 AgentExecutor 将我们的 tools 封装为 Agent"""
        from langchain import hub
        from langchain.agents import create_tool_calling_agent
        from langchain.agents import AgentExecutor

        prompt = hub.pull("hwchase17/openai-functions-agent")
        agent = create_tool_calling_agent(self.model, self.tools, prompt)
        self.agent_executor = AgentExecutor(agent=agent, tools=self.tools)

    def query_with_agent(self, query=None):
        """通过 agent 执行查询，将会调用到我们的 tools"""
        res = self.agent_executor.invoke({"input": query})
        print(f"ContentString: {res['output']}")

if __name__ == '__main__':
    a = TestAgent()
    a.build_retriever_tool()
    # a.invoke_llm_with_tools("告诉我关于 LangSmith 的相关信息")
    a.create_agent_executor()
    a.query_with_agent("告诉我关于 LangSmith 的相关信息")

保存查询历史上下文

上边的查询都是没有历史上下文的，如果说我们需要像试用 ChatGPT 一样可以多次提问，并且带有历史上下文，该如何实现呢？
其实就是一种很直观简单的方式，将历史上下文给保存起来，然后在每次提问的时候，将历史上下文带上。

import os

from langchain.agents import AgentExecutor
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores.dashvector import DashVector
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain.tools.retriever import create_retriever_tool
from langchain_community.embeddings.dashscope import DashScopeEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.messages import HumanMessage, AIMessage


class TestAgent:

    def __init__(self):
        """在阿里云控制台开通 DASHSCOPE 和 DASHVECTOR"""
        os.environ["DASHSCOPE_API_KEY"] = ""
        os.environ["DASHVECTOR_API_KEY"] = ""
        os.environ["DASHVECTOR_ENDPOINT"] = ""

        self.model = ChatTongyi()

        # 一系列 tool 让 langchain 选择使用
        self.tools = []

        self.agent_executor = None

    def build_retriever_tool(self):
        """爬取网站内容切分成 chunks 嵌入向量数据库"""
        loader = WebBaseLoader("https://docs.smith.langchain.com/overview")
        docs = loader.load()
        documents = RecursiveCharacterTextSplitter(
            chunk_size=1000, chunk_overlap=200
        ).split_documents(docs)
        vector = DashVector.from_documents(documents, DashScopeEmbeddings(), collection_name="langchain")
        retriever = vector.as_retriever()
        retriever_tool = create_retriever_tool(
            retriever,
            "langsmith_search",
            # 注意这里的 description 很重要，它描述的意思决定了 langchain 是否根据当前上下文选择该工具
            # 因此，description 要尽量描述的和 tool 所做内容吻合
            "Search for information about LangSmith. For any questions about LangSmith, you must use this tool!",
        )
        self.tools.append(retriever_tool)

    def invoke_llm_with_tools(self, query=None):
        """
        向 LLM 提出问题，让大模型结合自己所学习的内容和我们自己构造的 tools 来回答
        """
        model_with_tools = self.model.bind_tools(self.tools)

        response = model_with_tools.invoke([HumanMessage(content=query)])
        print(f"ContentString: {response.content}")
        # 可以查看 LLM 在回答我们的问题的时候用到了哪些 tool
        print(f"ToolCalls: {response.tool_calls}")

    def create_agent_executor(self):
        """通过 AgentExecutor 将我们的 tools 封装为 Agent"""
        from langchain import hub
        from langchain.agents import create_tool_calling_agent
        from langchain.agents import AgentExecutor

        prompt = hub.pull("hwchase17/openai-functions-agent")
        agent = create_tool_calling_agent(self.model, self.tools, prompt)
        self.agent_executor = AgentExecutor(agent=agent, tools=self.tools)

    def query_with_agent(self, query=None):
        """通过 agent 执行查询，将会调用到我们的 tools"""
        res = self.agent_executor.invoke({"input": query})
        print(f"ContentString: {res['output']}")

    def query_with_history(self):
        """带有历史上下文的查询, 新的提问时，将历史提问和回答信息一起手动带上"""
        self.agent_executor.invoke({"input": "hi! my name is bob", "chat_history": []})
        self.agent_executor.invoke(
            {
                "chat_history": [
                    HumanMessage(content="hi! my name is bob"),
                    AIMessage(content="Hello Bob! How can I assist you today?"),
                ],
                "input": "what's my name?",
            }
        )

    def query_with_memory(self):
        """
        让 langchain 自己将历史信息带上，无需我们手动带上，我们只需要在本地或者数据库存储历史信息即可
        """
        from langchain_community.chat_message_histories import ChatMessageHistory
        from langchain_core.chat_history import BaseChatMessageHistory
        from langchain_core.runnables.history import RunnableWithMessageHistory

        # 这里为了简单直接使用字典保存历史信息，生产环境应该使用数据库保存
        store = {}

        def get_session_history(session_id: str) -> BaseChatMessageHistory:
            if session_id not in store:
                store[session_id] = ChatMessageHistory()
            return store[session_id]

        agent_with_chat_history = RunnableWithMessageHistory(
            self.agent_executor,
            get_session_history,
            input_messages_key="input",
            history_messages_key="chat_history",
        )

        agent_with_chat_history.invoke(
            {"input": "hi! I'm bob"},
            config={"configurable": {"session_id": "<foo>"}},
        )

        res = agent_with_chat_history.invoke(
            {"input": "what's my name?"},
            config={"configurable": {"session_id": "<foo>"}},
        )
        print(res['output'])  # Your name is Bob.

if __name__ == '__main__':
    a = TestAgent()
    a.build_retriever_tool()
    # a.invoke_llm_with_tools("告诉我关于 LangSmith 的相关信息")
    a.create_agent_executor()
    # a.query_with_agent("告诉我关于 LangSmith 的相关信息")
    a.query_with_memory()

• input_messages_key: 当前提问的内容的 key, 在上边的代码中就是 input, 该参数用于将当前提问内容加入历史。
• history_messages_key：历史内容的 key, 用于加载历史消息
• session_id: 用于保存当前对话内容，隔离不同对话
  

总结

本文总结了如何使用 Langchain 的 AgentExecutor 封装 tools, 构造我们自己的 Agent, 然后通过 LLM 调用，或者通过 Agent 调用，最后总结了如何让我们的 Agent 带有记忆功能，可以保存历史对话记录，这里为了简单就没有使用实际的数据库，实际生产中还是要使用数据库去保存这些信息的，感兴趣的朋友可以尝试下。