LangChain简明教程(5)_langchain中agent 怎么控制结束-CSDN博客

本文链接：https://blog.csdn.net/qiwsir/article/details/147408116

《LangChain简明教程》系列文章目录

模块 III：Agent(2)

Agent

AgentExecutor 是 agent 的运行时环境。它负责调用 agent，执行它选择的操作，将操作输出返回给 agent，并重复该过程，直到 agent 完成任务。在伪代码中，AgentExecutor 可能看起来像这样：

next_action = agent.get_action(...)
while next_action != AgentFinish:
    observation = run(next_action)
    next_action = agent.get_action(..., next_action, observation)
return next_action

AgentExecutor 处理各种复杂性，例如：当 agent 选择了一个不存在的工具时，AgentExecutor 负责处理此时出现的情况，另外还处理工具错误、管理 agent 生成的输出，以及在所有级别提供日志记录和可观测性内容等。

可以说， AgentExecutor 类是 LangChain 中主要的用于执行 agent 的机制，此外，还存在一些其他实验性的机制，如：

计划与执行 Agent (Plan-and-execute Agent)
Baby AGI
Auto GPT

为了更好地理解 agent 框架，让我们从头开始构建一个基本的 agent，然后继续探索预构建的 agents。

在深入构建 agent 之前，有必要回顾一些关键术语和模式：

AgentAction：一个数据类，表示 agent 应该采取的操作。它包含一个 tool 属性（要调用的工具名称）和一个 tool_input 属性（该工具的输入）。
AgentFinish：一个数据类，表示 agent 已完成任务，并应向用户返回响应。它通常包含一个返回值字典，通常有一个键 "output"，其中包含响应文本。
Intermediate Steps（中间步骤）：这些是 agent 之前操作及其相应输出的记录。它们对于为未来的迭代传递上下文至关重要。

在以下示例中，将使用 OpenAI 的函数调用功能来创建 agent。这种方法在构建 agent 时非常可靠。首先创建一个简单的工具，用于计算单词的长度。这个工具很有用，因为语言模型有时会由于分词问题而在计算单词长度时出错。

首先，加载将用于控制 agent 的语言模型：

from langchain.chat_models import ChatOpenAI

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

测试一下，用这个模型统计一个单词的长度。

llm.invoke("how many letters in the word educa?")

上述的返回内容中应指示单词 “educa” 中的字母数量。

接下来，定义一个简单的 Python 函数来计算单词的长度：

from langchain.agents import tool

@tool
def get_word_length(word: str) -> int:
    """Returns the length of a word."""
    return len(word)

tools = [get_word_length]

函数 get_word_length 以一个单词作为输入并返回其长度。

现在，为 agent 创建提示词（prompt）。提示的作用是指导 agent 如何进行推理以及如何格式化输出。在示例中，使用 OpenAI 函数调用功能。下面定义一个包含用户输入和 agent_scratchpad 占位符的提示词：

from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are a very powerful assistant but not great at calculating word lengths.",
        ),
        ("user", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ]
)

那么，agent 如何知道它可以使用哪些工具呢？这要依赖于 `format_tool_to_openai_function。为了将工具提供给 agent，需要将它们格式化为 OpenAI 函数调用：

from langchain.tools.render import format_tool_to_openai_function

llm_with_tools = llm.bind(functions=[format_tool_to_openai_function(t) for t in tools])

现在，通过定义输入映射并连接各个组件来创建 agent：这是 LCEL 语言。我们将在后面详细讨论它。

from langchain.agents.format_scratchpad import format_to_openai_function_messages
from langchain.agents.output_parsers import OpenAIFunctionsAgentOutputParser

agent = (
    {
        "input": lambda x: x["input"],
        "agent_scratchpad": lambda x: format_to_openai

_function_messages(
            x["intermediate_steps"]
        ),
    }
    | prompt
    | llm_with_tools
    | OpenAIFunctionsAgentOutputParser()
)

经过上述操作，就已经创建一个 agent，它能够理解用户输入、使用可用工具并格式化输出。下面演示与其交互过程：

agent.invoke({"input": "how many letters in the word educa?", "intermediate_steps": []})

agent 应该会返回一个 AgentAction，指示下一步要采取的操作。

虽然已经创建了 agent，但现在要为它编写一个运行机制（runtime，也常称为“运行时”），最简单就是不断地调用 agent，执行操作，并重复此过程直到 agent 完成任务。以下是一个示例：

from langchain.schema.agent import AgentFinish

user_input = "how many letters in the word educa?"
intermediate_steps = []

while True:
    output = agent.invoke(
        {
            "input": user_input,
            "intermediate_steps": intermediate_steps,
        }
    )
    if isinstance(output, AgentFinish):
        final_result = output.return_values["output"]
        break
    else:
        print(f"TOOL NAME: {output.tool}")
        print(f"TOOL INPUT: {output.tool_input}")
        tool = {"get_word_length": get_word_length}[output.tool]
        observation = tool.run(output.tool_input)
        intermediate_steps.append((output, observation))

print(final_result)

在上述代码的循环中，反复调用 agent，执行相关操作并更新中间步骤，直到 agent 完成。

在这里插入图片描述

为了简化这一过程，LangChain 提供了 AgentExecutor 类，它封装了 agent 的执行过程，并提供了错误处理、提前停止、跟踪记录以及其他改进功能。下面使用 AgentExecutor 来与 agent 进行交互：

from langchain.agents import AgentExecutor

agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

agent_executor.invoke({"input": "how many letters in the word educa?"})

AgentExecutor 简化了执行过程，并提供了一种与 agent 交互的便捷方式。

到目前为止，我们创建的 agent 是无状态的，这意味着它不会记住之前的交互。为了支持后续问题和对话，需要为 agent 添加记忆功能。这包括两个步骤：

在提示（prompt）中添加一个记忆变量，用于存储聊天历史记录。
在交互过程中跟踪聊天历史记录。

首先在提示中添加一个记忆占位符：

from langchain.prompts import MessagesPlaceholder

MEMORY_KEY = "chat_history"
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are a very powerful assistant but not great at calculating word lengths.",
        ),
        MessagesPlaceholder(variable_name=MEMORY_KEY),
        ("user", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ]
)

现在，创建一个列表来跟踪聊天历史记录：

from langchain.schema.messages import HumanMessage, AIMessage

chat_history = []

在以下的 agent 中创建执行步骤，其中包含记忆内容：

agent = (
    {
        "input": lambda x: x["input"],
        "agent_scratchpad": lambda x: format_to_openai_function_messages(
            x["intermediate_steps"]
        ),
        "chat_history": lambda x: x["chat_history"],
    }
    | prompt
    | llm_with_tools
    | OpenAIFunctionsAgentOutputParser()
)

执行 agent，确认更新聊天历史：

input1 = "how many letters in the word educa?"
result = agent_executor.invoke({"input": input1, "chat_history": chat_history})
chat_history.extend([
    HumanMessage(content=input1),
    AIMessage(content=result["output"]),
])
agent_executor.invoke({"input": "is that a real word?", "chat_history": chat_history})