如何使用 LangChain 构建智能代理。通过结合语言模型的工具调用能力，代理可以执行搜索、检索本地知识库及自定义函数等操作。教程涵盖了环境安装、工具定义（如 Tavily 搜索和 FAISS 向量检索）、LLM 绑定、代理创建、流式响应处理以及基于检查点的内存持久化。最终实现了一个能够自主决策并记忆上下文的多工具交互系统。
单独来看，语言模型无法采取行动 - 它们只能输出文本。LangChain 的一个重要用例是创建代理。代理是使用 LLM 作为推理引擎的系统，用于确定要采取的行动以及这些行动的输入应该是什么。然后，这些行动的结果可以反馈给代理，并确定是否需要更多的行动，或者是否可以结束。
在本教程中，我们将构建一个可以与多个不同工具进行交互的代理：一个是本地数据库，另一个是搜索引擎。您将能够向该代理提问，观察它调用工具，并与它进行对话。
我们将涵盖的概念是：
LangChain 支持 Python 和 JavaScript 两个开发版本，本教程使用 python3.11 版本进行讲解。
安装 langchain（版本 v0.2）：
pip install langchain
集成存在于 langchain-community 包中。我们还需要安装 tavily-python 包本身。
pip install -U langchain-community tavily-python langgraph
我们首先需要创建我们想要使用的工具。我们将使用两个工具：Tavily（在线搜索），然后是我们将创建的本地索引上的检索器。
我们在 LangChain 中有一个内置工具，可以轻松地使用 Tavily 搜索引擎作为工具。请注意，这需要一个 API 密钥。但是如果你没有或者不想创建一个，你可以忽略这一步。
Tavily：https://tavily.com/。创建 API 密钥后，您需要将其导出为：
export TAVILY_API_KEY="..."
from langchain_community.tools.tavily_search import TavilySearchResults
search = TavilySearchResults(max_results=1)
result = search.invoke("今天重庆的天气预报")
print(result)
API Reference: TavilySearchResults
我们还将在我们自己的一些数据上创建一个检索器。有关此处每个步骤的更深入解释，请参阅官方教程。
安装依赖：
pip install faiss-cpu
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()
result = retriever.invoke("灵积模型是什么？")
print(result)
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("----")
除了流回消息之外，流回令牌也很有用。我们可以使用 .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 文档。这有它自己的一组概念、教程和操作指南。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online