LangChain 构建代理：工具调用与内存管理

API Reference: WebBaseLoader | FAISS | OpenAIEmbeddings | RecursiveCharacterTextSplitter

现在我们已经填充了我们将进行检索的索引，我们可以很容易地将其转换为工具（代理正确使用它所需的格式）

from langchain.tools.retriever import create_retriever_tool

retriever_tool = create_retriever_tool(
    retriever,
    "搜索灵积模型",
    "搜索灵积模型的信息，你可以使用这个工具!",
)

API Reference: create_retriever_tool

工具列表

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

# 创建工具列表
toolList = [search, retriever_tool]

使用语言模型

接下来，让我们通过调用工具来学习如何使用语言模型。LangChain 支持许多不同的语言模型，您可以互换使用。

LangChain 支持内置的 Tavily 工具、支持 DashScopeEmbeddings 工具、支持自定义函数工具。

from langchain_community.tools.tavily_search import TavilySearchResults
search = TavilySearchResults(max_results=1)

from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
loader = WebBaseLoader("https://help.aliyun.com/zh/dashscope/product-overview/concepts-and-glossary")
docs = loader.load()
documents = RecursiveCharacterTextSplitter(
    chunk_size=1000, chunk_overlap=200
).split_documents(docs)
vector = FAISS.from_documents(documents, DashScopeEmbeddings())
retriever = vector.as_retriever()

from langchain.tools.retriever import create_retriever_tool
retriever_tool = create_retriever_tool(
    retriever,
    "search_dashscope_knowledge",
    "当需要搜索灵积模型相关知识的时候必须使用该工具",
)

from langchain_core.tools import tool
@tool
def multiply(first_int: int, second_int: int) -> int:
    """两个数的乘积."""
    return first_int * second_int

# 创建工具列表
toolList = [search, retriever_tool, multiply]

from langchain_community.chat_models.tongyi import ChatTongyi
llm = ChatTongyi(model="qwen-max")
llm_with_tools = llm.bind_tools([search, retriever_tool, multiply])

# 未使用工具
msg = llm.invoke("你好")
print(msg)

# 建议使用工具：search_dashscope_knowledge
msg = llm_with_tools.invoke("灵积模型是什么")
print(msg)

# 建议使用工具：multiply
msg = llm_with_tools.invoke("计算 10 的 5 倍")
print(msg)

这还没有调用那个工具——大模型只是告诉我们建议使用哪个工具。为了真正调用它，下面我们需要创建我们的代理。

创建代理

我们现在可以在几个查询上运行代理！请注意，目前，这些都是无状态查询（也就是它不会记住以前的交互）。请注意，代理将返回最终交互结束时的状态。

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

# 创建一个代理
agent_executor = create_react_agent(llm, toolList)

msg = agent_executor.invoke(
    {"messages": [HumanMessage(content="你好")]}
)
print(msg)

为了准确地看到到底发生了什么（并确保它没有调用工具），我们可以查看 LangSmith 跟踪。

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

# 建议使用工具：search_dashscope_knowledge
msg = agent_executor.invoke(
    {"messages": [HumanMessage(content="灵积模型是什么")]}
)
print(msg)

现在让我们尝试一个需要调用自定义工具的示例：

# 建议使用工具：multiply
msg = llm_with_tools.invoke("计算 10 的 5 倍等于的结果")
print(msg)

流式消息

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

for chunk in agent_executor.stream(
    {"messages": [HumanMessage(content="你好")]}
): print(chunk)
print("----")

# 建议使用工具：search_dashscope_knowledge
for chunk in agent_executor.stream(
    {"messages": [HumanMessage(content="灵积模型是什么")]}
): print(chunk)
print("----")

# 建议使用工具：multiply
for chunk in llm_with_tools.stream("计算 10 的 5 倍的结果"): print(chunk)
print("----")

Streaming tokens

除了流回消息之外，流回令牌也很有用。我们可以使用 .astream_events 方法。

此.astream_events 方法仅适用于 Python 3.11 或更高版本。

async def main():
     async for event in agent_executor.astream_events(
        {"messages": [HumanMessage(content="你好")]}, version="v1"
    ):
        kind = event["event"]
        if kind == "on_chain_start":
            if (
                event["name"] == "Agent"
            ):   
                print(
                    f"Starting agent: {event['name']} with input: {event['data'].get('input')}"
                )
        elif kind == "on_chain_end":
            if (
                event["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:
                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("--")

import asyncio
if __name__ == "__main__":
    asyncio.run(main())

内存缓存

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

# 创建缓存
from langgraph.checkpoint.sqlite import SqliteSaver
memory = SqliteSaver.from_conn_string(":memory:")

# 创建代理
from langgraph.prebuilt import create_react_agent
from langchain_core.messages import HumanMessage
agent_executor = create_react_agent(llm, toolList, checkpointer=memory)
config = {"configurable": {"thread_id": "abc123"}}

for chunk in agent_executor.stream(
    {"messages": [HumanMessage(content="你好，我是小明")]}, config=config
): print(chunk)

for chunk in agent_executor.stream(
    {"messages": [HumanMessage(content="我的名字是什么？")]}, config=config
): print(chunk)

结论

就这样结束了！在这个快速的开始中，我们介绍了如何创建一个简单的代理。然后我们展示了如何流回响应——不仅是中间步骤，还有令牌！我们还添加了内存，这样你就可以和他们对话了。代理是一个复杂的话题，有很多东西要学！

有关代理的更多信息，请查看 LangGraph 文档。这有它自己的一组概念、教程和操作指南。