构建聊天机器人
本教程之前使用了 RunnableWithMessageHistory 抽象。您可以在 v0.2 文档中访问该版本的文档。
从 LangChain 的 v0.3 版本开始,我们建议 LangChain 用户利用 LangGraph 持久性 将 memory
融入新的 LangChain 应用程序中。
如果您的代码已经依赖 RunnableWithMessageHistory
或 BaseChatMessageHistory
,则 **无需** 进行任何更改。我们不打算在近期弃用此功能,因为它适用于简单的聊天应用程序,并且任何使用 RunnableWithMessageHistory
的代码都将继续按预期工作。
有关更多详细信息,请参阅 如何迁移到 LangGraph 内存。
概述
我们将介绍如何设计和实现一个由 LLM 驱动的聊天机器人的示例。该聊天机器人将能够进行对话,并记住之前与 聊天模型 的互动。
请注意,我们构建的这个聊天机器人只会使用语言模型进行对话。您可能正在寻找其他几个相关的概念
本教程将介绍一些基础知识,这些基础知识对以上两个更高级的主题很有帮助,但如果您选择,也可以直接跳到那里。
设置
Jupyter Notebook
本指南(以及文档中的大多数其他指南)使用 Jupyter notebooks,并假设读者也是如此。Jupyter notebooks 非常适合学习如何使用 LLM 系统,因为通常情况下可能会出现错误(意外输出、API 宕机等),并且在交互式环境中浏览指南是更好地理解它们的好方法。
本教程和其他教程最好在 Jupyter notebook 中运行。有关如何安装的说明,请参阅 此处。
安装
在本教程中,我们将需要 langchain-core
和 langgraph
。本指南需要 langgraph >= 0.2.28
。
- Pip
- Conda
pip install langchain-core langgraph>0.2.27
conda install langchain-core langgraph>0.2.27 -c conda-forge
有关更多详细信息,请参阅我们的 安装指南。
LangSmith
您使用 LangChain 构建的许多应用程序将包含多个步骤,其中包含多次 LLM 调用。随着这些应用程序变得越来越复杂,能够检查链或代理内部究竟发生了什么变得至关重要。最好的方法是使用 LangSmith。
在上面的链接注册后,请确保设置您的环境变量以开始记录跟踪
export LANGCHAIN_TRACING_V2="true"
export LANGCHAIN_API_KEY="..."
或者,如果在笔记本中,您可以使用以下代码设置它们
import getpass
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = getpass.getpass()
快速入门
首先,让我们学习如何单独使用语言模型。LangChain 支持许多不同的语言模型,您可以互换使用它们 - 在下面选择您想要使用的模型!
pip install -qU langchain-openai
import getpass
import os
if not os.environ.get("OPENAI_API_KEY"):
os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter API key for OpenAI: ")
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o-mini")
让我们首先直接使用该模型。ChatModel
是 LangChain “Runnables” 的实例,这意味着它们公开了一个用于与其交互的标准接口。要简单地调用该模型,我们可以将消息列表传递给 .invoke
方法。
from langchain_core.messages import HumanMessage
model.invoke([HumanMessage(content="Hi! I'm Bob")])
AIMessage(content='Hi Bob! How can I assist you today?', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 10, 'prompt_tokens': 11, 'total_tokens': 21, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-mini-2024-07-18', 'system_fingerprint': 'fp_0705bf87c0', 'finish_reason': 'stop', 'logprobs': None}, id='run-5211544f-da9f-4325-8b8e-b3d92b2fc71a-0', usage_metadata={'input_tokens': 11, 'output_tokens': 10, 'total_tokens': 21, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})
模型本身没有任何状态的概念。例如,如果您提出后续问题
model.invoke([HumanMessage(content="What's my name?")])
AIMessage(content="I'm sorry, but I don't have access to personal information about users unless it has been shared with me in the course of our conversation. How can I assist you today?", additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 34, 'prompt_tokens': 11, 'total_tokens': 45, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-mini-2024-07-18', 'system_fingerprint': 'fp_0705bf87c0', 'finish_reason': 'stop', 'logprobs': None}, id='run-a2d13a18-7022-4784-b54f-f85c097d1075-0', usage_metadata={'input_tokens': 11, 'output_tokens': 34, 'total_tokens': 45, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})
让我们看一下 LangSmith 跟踪示例
我们可以看到,它没有将之前的对话回合纳入上下文中,因此无法回答该问题。这会带来糟糕的聊天机器人体验!
为了解决这个问题,我们需要将整个 对话历史 传递到模型中。让我们看看这样做会发生什么
from langchain_core.messages import AIMessage
model.invoke(
[
HumanMessage(content="Hi! I'm Bob"),
AIMessage(content="Hello Bob! How can I assist you today?"),
HumanMessage(content="What's my name?"),
]
)
AIMessage(content='Your name is Bob! How can I help you today, Bob?', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 33, 'total_tokens': 47, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-mini-2024-07-18', 'system_fingerprint': 'fp_0705bf87c0', 'finish_reason': 'stop', 'logprobs': None}, id='run-34bcccb3-446e-42f2-b1de-52c09936c02c-0', usage_metadata={'input_tokens': 33, 'output_tokens': 14, 'total_tokens': 47, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})
现在我们可以看到我们得到了很好的回复!
这是聊天机器人进行对话式交互的基本思想。那么我们如何最好地实现它呢?
消息持久性
LangGraph 实现了一个内置的持久性层,使其成为支持多个对话回合的聊天应用程序的理想选择。
将我们的聊天模型包装在一个最小的 LangGraph 应用程序中,可以让我们自动持久化消息历史,从而简化了多轮应用程序的开发。
LangGraph 配备了一个简单的内存检查点,我们在下面使用它。有关更多详细信息,包括如何使用不同的持久性后端(例如,SQLite 或 Postgres),请参阅其文档。
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import START, MessagesState, StateGraph
# Define a new graph
workflow = StateGraph(state_schema=MessagesState)
# Define the function that calls the model
def call_model(state: MessagesState):
response = model.invoke(state["messages"])
return {"messages": response}
# Define the (single) node in the graph
workflow.add_edge(START, "model")
workflow.add_node("model", call_model)
# Add memory
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
我们现在需要创建一个 config
,每次将其传递到 runnable 中。此 config 包含不直接属于输入一部分但仍然有用的信息。在这种情况下,我们想要包含一个 thread_id
。它应该看起来像这样
config = {"configurable": {"thread_id": "abc123"}}
这使我们能够使用单个应用程序支持多个对话线程,这是应用程序具有多个用户的常见要求。
然后我们可以调用该应用程序
query = "Hi! I'm Bob."
input_messages = [HumanMessage(query)]
output = app.invoke({"messages": input_messages}, config)
output["messages"][-1].pretty_print() # output contains all messages in state
==================================[1m Ai Message [0m==================================
Hi Bob! How can I assist you today?
query = "What's my name?"
input_messages = [HumanMessage(query)]
output = app.invoke({"messages": input_messages}, config)
output["messages"][-1].pretty_print()
==================================[1m Ai Message [0m==================================
Your name is Bob! How can I help you today, Bob?
太棒了!我们的聊天机器人现在记住了关于我们的事情。如果我们更改配置以引用不同的 thread_id
,我们可以看到它会重新开始对话。
config = {"configurable": {"thread_id": "abc234"}}
input_messages = [HumanMessage(query)]
output = app.invoke({"messages": input_messages}, config)
output["messages"][-1].pretty_print()
==================================[1m Ai Message [0m==================================
I'm sorry, but I don't have access to personal information about you unless you've shared it in this conversation. How can I assist you today?
但是,我们始终可以返回到原始对话(因为我们将其持久化在数据库中)
config = {"configurable": {"thread_id": "abc123"}}
input_messages = [HumanMessage(query)]
output = app.invoke({"messages": input_messages}, config)
output["messages"][-1].pretty_print()
==================================[1m Ai Message [0m==================================
Your name is Bob. What would you like to discuss today?
这就是我们如何支持聊天机器人与许多用户进行对话的方式!
对于异步支持,请将 call_model
节点更新为异步函数,并在调用应用程序时使用 .ainvoke
# Async function for node:
async def call_model(state: MessagesState):
response = await model.ainvoke(state["messages"])
return {"messages": response}
# Define graph as before:
workflow = StateGraph(state_schema=MessagesState)
workflow.add_edge(START, "model")
workflow.add_node("model", call_model)
app = workflow.compile(checkpointer=MemorySaver())
# Async invocation:
output = await app.ainvoke({"messages": input_messages}, config)
output["messages"][-1].pretty_print()
现在,我们所做的只是在模型周围添加一个简单的持久性层。我们可以通过添加提示模板来使聊天机器人更加复杂和个性化。
提示模板
提示模板 帮助将原始用户信息转换为 LLM 可以使用的格式。在这种情况下,原始用户输入只是一条消息,我们将其传递给 LLM。现在让我们让它稍微复杂一些。首先,让我们添加一条带有自定义指令的系统消息(但仍然将消息作为输入)。接下来,我们将添加除消息之外的更多输入。
要添加系统消息,我们将创建一个 ChatPromptTemplate
。我们将利用 MessagesPlaceholder
来传递所有消息。
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
prompt_template = ChatPromptTemplate.from_messages(
[
(
"system",
"You talk like a pirate. Answer all questions to the best of your ability.",
),
MessagesPlaceholder(variable_name="messages"),
]
)
我们现在可以更新我们的应用程序以合并此模板
workflow = StateGraph(state_schema=MessagesState)
def call_model(state: MessagesState):
prompt = prompt_template.invoke(state)
response = model.invoke(prompt)
return {"messages": response}
workflow.add_edge(START, "model")
workflow.add_node("model", call_model)
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
我们以相同的方式调用该应用程序
config = {"configurable": {"thread_id": "abc345"}}
query = "Hi! I'm Jim."
input_messages = [HumanMessage(query)]
output = app.invoke({"messages": input_messages}, config)
output["messages"][-1].pretty_print()
==================================[1m Ai Message [0m==================================
Ahoy there, Jim! What brings ye to these waters today? Be ye seekin' treasure, knowledge, or perhaps a good tale from the high seas? Arrr!
query = "What is my name?"
input_messages = [HumanMessage(query)]
output = app.invoke({"messages": input_messages}, config)
output["messages"][-1].pretty_print()
==================================[1m Ai Message [0m==================================
Ye be called Jim, matey! A fine name fer a swashbuckler such as yerself! What else can I do fer ye? Arrr!
太棒了!现在让我们让我们的提示稍微复杂一些。让我们假设提示模板现在看起来像这样
prompt_template = ChatPromptTemplate.from_messages(
[
(
"system",
"You are a helpful assistant. Answer all questions to the best of your ability in {language}.",
),
MessagesPlaceholder(variable_name="messages"),
]
)
请注意,我们已向提示添加了一个新的 language
输入。我们的应用程序现在有两个参数——输入 messages
和 language
。我们应该更新应用程序的状态以反映这一点
from typing import Sequence
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
from typing_extensions import Annotated, TypedDict
class State(TypedDict):
messages: Annotated[Sequence[BaseMessage], add_messages]
language: str
workflow = StateGraph(state_schema=State)
def call_model(state: State):
prompt = prompt_template.invoke(state)
response = model.invoke(prompt)
return {"messages": [response]}
workflow.add_edge(START, "model")
workflow.add_node("model", call_model)
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "abc456"}}
query = "Hi! I'm Bob."
language = "Spanish"
input_messages = [HumanMessage(query)]
output = app.invoke(
{"messages": input_messages, "language": language},
config,
)
output["messages"][-1].pretty_print()
==================================[1m Ai Message [0m==================================
¡Hola, Bob! ¿Cómo puedo ayudarte hoy?
请注意,整个状态都会被持久化保存,因此如果不需要更改,我们可以省略诸如 language
之类的参数。
query = "What is my name?"
input_messages = [HumanMessage(query)]
output = app.invoke(
{"messages": input_messages},
config,
)
output["messages"][-1].pretty_print()
==================================[1m Ai Message [0m==================================
Tu nombre es Bob. ¿Hay algo más en lo que pueda ayudarte?
为了帮助您理解内部发生了什么,请查看这个 LangSmith 追踪。
管理对话历史
构建聊天机器人时,一个需要理解的重要概念是如何管理对话历史。如果不加以管理,消息列表将会无限增长,并可能超出 LLM 的上下文窗口。因此,添加一个限制传入消息大小的步骤至关重要。
重要的是,您需要在提示模板之前,但在从消息历史加载先前消息之后执行此操作。
我们可以通过在提示之前添加一个简单的步骤来修改 messages
键,然后将新的链包装在消息历史类中来实现这一点。
LangChain 提供了一些内置的助手,用于管理消息列表。在这种情况下,我们将使用 trim_messages 助手来减少发送到模型的消息数量。修剪器允许我们指定要保留多少个令牌,以及其他参数,例如是否始终保留系统消息以及是否允许部分消息。
from langchain_core.messages import SystemMessage, trim_messages
trimmer = trim_messages(
max_tokens=65,
strategy="last",
token_counter=model,
include_system=True,
allow_partial=False,
start_on="human",
)
messages = [
SystemMessage(content="you're a good assistant"),
HumanMessage(content="hi! I'm bob"),
AIMessage(content="hi!"),
HumanMessage(content="I like vanilla ice cream"),
AIMessage(content="nice"),
HumanMessage(content="whats 2 + 2"),
AIMessage(content="4"),
HumanMessage(content="thanks"),
AIMessage(content="no problem!"),
HumanMessage(content="having fun?"),
AIMessage(content="yes!"),
]
trimmer.invoke(messages)
[SystemMessage(content="you're a good assistant", additional_kwargs={}, response_metadata={}),
HumanMessage(content='whats 2 + 2', additional_kwargs={}, response_metadata={}),
AIMessage(content='4', additional_kwargs={}, response_metadata={}),
HumanMessage(content='thanks', additional_kwargs={}, response_metadata={}),
AIMessage(content='no problem!', additional_kwargs={}, response_metadata={}),
HumanMessage(content='having fun?', additional_kwargs={}, response_metadata={}),
AIMessage(content='yes!', additional_kwargs={}, response_metadata={})]
要在我们的链中使用它,我们只需要在将 messages
输入传递给我们的提示之前运行修剪器即可。
workflow = StateGraph(state_schema=State)
def call_model(state: State):
trimmed_messages = trimmer.invoke(state["messages"])
prompt = prompt_template.invoke(
{"messages": trimmed_messages, "language": state["language"]}
)
response = model.invoke(prompt)
return {"messages": [response]}
workflow.add_edge(START, "model")
workflow.add_node("model", call_model)
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
现在,如果我们尝试询问模型我们的名字,它将不知道,因为我们已经修剪了聊天历史的这一部分。
config = {"configurable": {"thread_id": "abc567"}}
query = "What is my name?"
language = "English"
input_messages = messages + [HumanMessage(query)]
output = app.invoke(
{"messages": input_messages, "language": language},
config,
)
output["messages"][-1].pretty_print()
==================================[1m Ai Message [0m==================================
I don't know your name. You haven't told me yet!
但是,如果我们询问最近几条消息中的信息,它会记住。
config = {"configurable": {"thread_id": "abc678"}}
query = "What math problem did I ask?"
language = "English"
input_messages = messages + [HumanMessage(query)]
output = app.invoke(
{"messages": input_messages, "language": language},
config,
)
output["messages"][-1].pretty_print()
==================================[1m Ai Message [0m==================================
You asked what 2 + 2 equals.
如果您查看 LangSmith,您可以在LangSmith 追踪中看到底层发生了什么。
流式传输
现在我们有了一个功能正常的聊天机器人。但是,对于聊天机器人应用程序来说,一个非常重要的用户体验考虑因素是流式传输。LLM 有时需要一段时间才能做出响应,因此为了改善用户体验,大多数应用程序所做的一件事是流式返回生成的每个令牌。这允许用户看到进度。
实际上,这非常容易做到!
默认情况下,我们的 LangGraph 应用程序中的 .stream
会流式传输应用程序步骤,在本例中,即模型响应的单个步骤。设置 stream_mode="messages"
允许我们流式传输输出令牌。
config = {"configurable": {"thread_id": "abc789"}}
query = "Hi I'm Todd, please tell me a joke."
language = "English"
input_messages = [HumanMessage(query)]
for chunk, metadata in app.stream(
{"messages": input_messages, "language": language},
config,
stream_mode="messages",
):
if isinstance(chunk, AIMessage): # Filter to just model responses
print(chunk.content, end="|")
|Hi| Todd|!| Here|’s| a| joke| for| you|:
|Why| don|’t| skeleton|s| fight| each| other|?
|Because| they| don|’t| have| the| guts|!||
后续步骤
既然您已经了解了如何在 LangChain 中创建聊天机器人的基本知识,您可能还会对一些更高级的教程感兴趣,例如:
如果您想深入了解具体细节,以下是一些值得查看的内容:
- 流式传输:流式传输对于聊天应用程序至关重要
- 如何添加消息历史记录:更深入地了解与消息历史记录相关的所有内容
- 如何管理大型消息历史记录:管理大型聊天历史记录的更多技术
- LangGraph 主要文档:有关使用 LangGraph 构建的更多详细信息