LangChain建立代理_langgrap 工具代理

语言模型本身无法采取行动——它们只是输出文本。 LangChain 的一个重要用例是创建代理。代理是使用 LLM 作为推理引擎来确定要采取哪些操作以及这些操作的输入应该是什么的系统。然后，这些操作的结果可以反馈给代理，并确定是否需要更多操作，或者是否可以完成。

在本教程中，我们将构建一个可以与多种不同工具交互的代理：一个是本地数据库，另一个是搜索引擎。您将能够向该代理询问问题、观看它调用工具并与其进行对话。

概念​

我们将涵盖的概念是：

• 使用语言模型，特别是它们的工具调用能力
• 创建检索器以向我们的代理公开特定信息
• 使用搜索工具在线查找内容
• 使用 LangGraph 代理，它使用 LLM 来思考要做什么，然后执行
• 使用 LangSmith 调试和跟踪您的应用程序

设置​

Jupyter 笔记本​

本指南（以及文档中的大多数其他指南）使用 Jupyter 笔记本，并假设读者也使用 Jupyter 笔记本。 Jupyter 笔记本非常适合学习如何使用 LLM 系统，因为经常会出现问题（意外输出、API 关闭等），在交互式环境中阅读指南是更好地理解它们的好方法。

安装​

要安装 LangChain 运行：

pip install langchain
1

LangSmith
您使用 LangChain 构建的许多应用程序将包含多个步骤，并多次调用 LLM 调用。随着这些应用程序变得越来越复杂，能够检查链或代理内部到底发生了什么变得至关重要。做到这一点的最佳方法是与 LangSmith 合作。

在上面的链接注册后，请确保设置环境变量以开始记录跟踪：

export LANGCHAIN_TRACING_V2="true"
export LANGCHAIN_API_KEY="..."
1
2

或者，如果在笔记本中，您可以使用以下命令设置它们：

import getpass
import os

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = getpass.getpass()
1
2
3
4
5

定义工具​

我们首先需要创建我们想要使用的工具。我们将使用两个工具：Tavily（用于在线搜索），然后是我们将创建的本地索引的检索器

Tavily

我们在LangChain中有一个内置的工具，可以方便地使用Tavily搜索引擎作为工具。请注意，这需要一个 API 密钥 - 他们有一个免费套餐，但如果您没有或不想创建一个，则可以随时忽略此步骤。

创建 API 密钥后，您需要将其导出为：

export TAVILY_API_KEY="..."
1

如果你是在记事本则要这样子

os.environ["TAVILY_API_KEY"] = getpass.getpass()
1

from langchain_community.tools.tavily_search import TavilySearchResults
1

search = TavilySearchResults(max_results=2)
1

search.invoke("what is the weather in SF")
1

Retriever

我们还将根据我们自己的一些数据创建一个检索器。有关此处每个步骤的更深入说明，请参阅本教程。

from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

loader = WebBaseLoader("https://docs.smith.langchain.com/overview")
docs = loader.load()
documents = RecursiveCharacterTextSplitter(
    chunk_size=1000, chunk_overlap=200
).split_documents(docs)
vector = FAISS.from_documents(documents, OpenAIEmbeddings())
retriever = vector.as_retriever()
1
2
3
4
5
6
7
8
9
10
11
12

retriever.invoke("how to upload a dataset")[0]
1

现在我们已经填充了我们将进行检索的索引，我们可以轻松地将其变成一个工具（代理正确使用它所需的格式）

from langchain.tools.retriever import create_retriever_tool
1

retriever_tool = create_retriever_tool(
    retriever,
    "langsmith_search",
    "Search for information about LangSmith. For any questions about LangSmith, you must use this tool!",
)
1
2
3
4
5

工具​

现在我们已经创建了两者，我们可以创建将在下游使用的工具列表。

tools = [search, retriever_tool]
1

使用语言模型​

接下来我们通过调用工具来学习如何使用语言模型。 LangChain支持多种不同的语言模型，您可以互换使用 - 在下面选择您想要使用的一种！

pip install -qU langchain-openai
1

import getpass
import os

os.environ["OPENAI_API_KEY"] = getpass.getpass()

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-3.5-turbo")
1
2
3
4
5
6
7
8

您可以通过传入消息列表来调用语言模型。默认情况下，响应是 content 字符串。

from langchain_core.messages import HumanMessage

response = model.invoke([HumanMessage(content="hi!")])
response.content
1
2
3
4

现在我们可以看到让这个模型进行工具调用是什么样子的。为了使我们能够使用 .bind_tools 来提供这些工具的语言模型知识

model_with_tools = model.bind_tools(tools)
1

我们现在可以调用该模型。我们首先用一条普通的消息来调用它，看看它如何响应。我们可以查看 content 字段以及 tool_calls 字段。

response = model_with_tools.invoke([HumanMessage(content="Hi!")])

print(f"ContentString: {response.content}")
print(f"ToolCalls: {response.tool_calls}")
1
2
3
4

现在，让我们尝试使用一些期望调用工具的输入来调用它。

response = model_with_tools.invoke([HumanMessage(content="What's the weather in SF?")])

print(f"ContentString: {response.content}")
print(f"ToolCalls: {response.tool_calls}")
1
2
3
4

我们可以看到现在没有内容，但是有一个工具调用！它希望我们调用 Tavily Search 工具。

这还没有调用该工具——它只是告诉我们这样做。为了实际调用它，我们需要创建我们的代理。

创建代理​

现在我们已经定义了工具和 LLM，我们可以创建代理了。我们将使用 LangGraph 来构建代理。目前我们使用高级接口来构建代理，但 LangGraph 的好处在于，这个高级接口由低级、高度可控的 API 支持，以防您想要修改代理逻辑。

现在，我们可以使用 LLM 和工具来初始化代理。

请注意，我们传递的是 model ，而不是 model_with_tools 。这是因为 create_tool_calling_executor 会在后台为我们调用 .bind_tools 。

from langgraph.prebuilt import chat_agent_executor

agent_executor = chat_agent_executor.create_tool_calling_executor(model, tools)
1
2
3

运行代理​

我们现在可以针对一些查询运行代理！请注意，目前这些都是无状态查询（它不会记住以前的交互）。请注意，代理将在交互结束时返回最终状态（其中包括任何输入，我们稍后将看到如何仅获取输出）。

首先，让我们看看当不需要调用工具时它是如何响应的：

response = agent_executor.invoke({"messages": [HumanMessage(content="hi!")]})

response["messages"]
1
2
3

为了准确了解幕后发生的情况（并确保它没有调用工具），我们可以查看 LangSmith 跟踪

现在让我们尝试一下应该调用检索器的示例

response = agent_executor.invoke(
    {"messages": [HumanMessage(content="how can langsmith help with testing?")]}
)
response["messages"]
1
2
3
4

让我们看一下 LangSmith 跟踪，看看幕后发生了什么。

请注意，我们最后返回的状态还包含工具调用和工具响应消息。

现在让我们尝试一下需要调用搜索工具的地方：

response = agent_executor.invoke(
    {"messages": [HumanMessage(content="whats the weather in sf?")]}
)
response["messages"]
1
2
3
4

我们可以检查 LangSmith 跟踪以确保它有效地调用搜索工具。

流媒体消息​

我们已经了解了如何使用 .invoke 调用代理来获取最终响应。如果代理正在执行多个步骤，则可能需要一段时间。为了显示中间进度，我们可以在消息发生时流回消息。

for chunk in agent_executor.stream(
    {"messages": [HumanMessage(content="whats the weather in sf?")]}
):
    print(chunk)
    print("----")
1
2
3
4
5

流媒体代币​

除了流回消息之外，流回令牌也很有用。我们可以使用 .astream_events 方法来做到这一点。
此 .astream_events 方法仅适用于 Python 3.11 或更高版本。

async for event in agent_executor.astream_events(
    {"messages": [HumanMessage(content="whats the weather in sf?")]}, version="v1"
):
    kind = event["event"]
    if kind == "on_chain_start":
        if (
            event["name"] == "Agent"
        ):  # Was assigned when creating the agent with `.with_config({"run_name": "Agent"})`
            print(
                f"Starting agent: {event['name']} with input: {event['data'].get('input')}"
            )
    elif kind == "on_chain_end":
        if (
            event["name"] == "Agent"
        ):  # Was assigned when creating the agent with `.with_config({"run_name": "Agent"})`
            print()
            print("--")
            print(
                f"Done agent: {event['name']} with output: {event['data'].get('output')['output']}"
            )
    if kind == "on_chat_model_stream":
        content = event["data"]["chunk"].content
        if content:
            # Empty content in the context of OpenAI means
            # that the model is asking for a tool to be invoked.
            # So we only print non-empty content
            print(content, end="|")
    elif kind == "on_tool_start":
        print("--")
        print(
            f"Starting tool: {event['name']} with inputs: {event['data'].get('input')}"
        )
    elif kind == "on_tool_end":
        print(f"Done tool: {event['name']}")
        print(f"Tool output was: {event['data'].get('output')}")
        print("--")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

添加到内存中​

如前所述，该代理是无状态的。这意味着它不记得以前的交互。为了给它内存，我们需要传入一个检查指针。当传入检查指针时，我们还必须在调用代理时传入 thread_id （以便它知道要从哪个线程/会话恢复）。

from langgraph.checkpoint.sqlite import SqliteSaver

memory = SqliteSaver.from_conn_string(":memory:")
1
2
3

agent_executor = chat_agent_executor.create_tool_calling_executor(
    model, tools, checkpointer=memory
)

config = {"configurable": {"thread_id": "abc123"}}
1
2
3
4
5

for chunk in agent_executor.stream(
    {"messages": [HumanMessage(content="hi im bob!")]}, config
):
    print(chunk)
    print("----")
1
2
3
4
5

for chunk in agent_executor.stream(
    {"messages": [HumanMessage(content="whats my name?")]}, config
):
    print(chunk)
    print("----")
1
2
3
4
5

LangSmith 迹线示例

结论​

这是一个包装！在本快速入门中，我们介绍了如何创建一个简单的代理。然后，我们展示了如何流回响应 - 不仅是中间步骤，还有令牌！我们还添加了内存，以便您可以与他们进行对话。代理是一个复杂的话题，有很多东西需要学习！