构建聊天机器人

笔记

本教程之前使用了 RunnableWithMessageHistory 抽象类。您可以在 v0.2 文档中访问该版本的文档。

在 LangChain v0.3 版本发布后，我们建议 LangChain 用户利用 LangGraph 持久化将 memory 集成到新的 LangChain 应用中。

如果您的代码已经依赖于 RunnableWithMessageHistory 或 BaseChatMessageHistory，则无需进行任何更改。由于该功能适用于简单的聊天应用，且使用 RunnableWithMessageHistory 的任何代码将继续按预期工作，因此我们计划在近期不弃用此功能。

请参阅如何迁移到 LangGraph Memory 以获取更多详细信息。

概览

我们将介绍一个如何设计和实现基于大型语言模型的聊天机器人的示例。该聊天机器人能够进行对话，并记住与聊天模型之前的互动。

请注意，我们构建的这个聊天机器人仅使用语言模型来进行对话。您可能还希望了解以下几个相关概念：

对话式RAG: 在外部数据源上实现聊天机器人体验
代理: 构建一个可以采取行动的聊天机器人

本教程将涵盖一些基础知识，对后续两个更高级的主题会有帮助，但您也可以根据需要直接跳转至相关内容。

设置

Jupyter Notebook

本指南（以及文档中的大多数其他指南）使用 Jupyter 笔记本，并假设读者也熟悉该工具。Jupyter 笔记本非常适合学习如何使用大语言模型系统，因为很多时候会出现问题（如输出意外、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 LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="..."

或者，如果在笔记本中，你可以通过以下方式设置它们：

import getpass
import os

os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_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.chat_models import init_chat_model

model = init_chat_model("gpt-4o-mini", model_provider="openai")

让我们首先直接使用该模型。ChatModel 是 LangChain “可运行”对象的实例，这意味着它们提供了一个标准接口以供与之交互。要简单地调用模型，我们可以向 .invoke 方法传递消息列表。

from langchain_core.messages import HumanMessage

model.invoke([HumanMessage(content="Hi! I'm Bob")])

API 参考：HumanMessage

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?"),
    ]
)

API 参考：AIMessage

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)

API 参考：MemorySaver | StateGraph

现在我们需要创建一个 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"),
    ]
)

API 参考：ChatPromptTemplate | MessagesPlaceholder

现在我们可以更新我们的应用程序以整合此模板：

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)

API 参考：BaseMessage | add_messages

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 跟踪。

管理对话历史

在构建聊天机器人时，有一个重要概念需要理解，那就是如何管理对话历史。如果不对对话历史进行管理，消息列表将无限增长，可能会超出大型语言模型的上下文窗口容量。因此，添加一个步骤来限制传入的消息数量非常重要。

重要的是，您应在提示模板之前执行此操作，但在加载消息历史记录中的先前消息之后。

我们可以通过在提示前添加一个简单的步骤来适当修改 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)

API 参考：SystemMessage | trim_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 跟踪中发生的一切。

流式传输

现在我们已经有一个可以正常运行的聊天机器人了。然而，对于聊天机器人应用来说，有一个非常重要的用户体验考虑因素是流式传输。大型语言模型有时需要一段时间才能响应，因此为了改善用户体验，大多数应用程序都会在生成每个标记时立即流式返回。这可以让用户看到进度。

实际上，这样做非常简单！

默认情况下，我们的 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 中创建聊天机器人的基本方法，以下是一些更高级的教程，你可能会感兴趣：

对话式RAG: 在外部数据源上实现聊天机器人体验
代理: 构建一个可以采取行动的聊天机器人

如果你想深入了解具体细节，以下一些内容值得查看：

流式传输: 流式传输对于聊天应用来说是至关重要的
如何添加消息历史记录: 详细了解与消息历史记录相关的所有内容
如何管理大型消息历史记录: 更多关于管理大型聊天历史记录的技术
LangGraph 主文档: 了解使用 LangGraph 构建的更多详细信息

概览​

设置​

Jupyter Notebook​

安装​

LangSmith​

快速入门​

消息持久化​

提示模板​

管理对话历史​

流式传输​

下一步​

概览

设置