基于 LangChain 构建具备记忆功能的聊天机器人

输出结果类似于：

Hello Bob! How can you assist me today?

注意：模型本身并没有任何状态的概念。例如，如果你问一个后续问题：

response = model.invoke([HumanMessage(content="What's my name?")])
print(response.content)

输出结果通常是：

I'm sorry, as an AI assistant, I do not have the capability to know your name unless you provide it to me.

我们可以看到它并没有将之前的对话内容纳入上下文，并且无法回答问题。这是因为每次调用都是独立的，模型不知道之前的交互。

为了解决这个问题，我们需要将整个对话历史传递给模型：

from langchain_core.messages import AIMessage

response = model.invoke([
    HumanMessage(content="Hi! I'm Bob"),
    AIMessage(content="Hello Bob! How can you assist me today?"),
    HumanMessage(content="What's my name?")
])
print(response.content)

现在得到了一个很好的回应：

Your name is Bob.

但是我们不能每次都把历史对话手动放进去。那么我们如何最好地实现它呢？这就是引入消息历史记录类的原因。

历史消息管理

我们可以使用一个 Message History 类来包装我们的模型，使其具有状态。这样可以跟踪模型的输入和输出，并把它们存储起来。未来的交互将加载这些消息，并将它们作为输入的一部分传递给模型。

实现持久化存储

首先，确保安装了 langchain-community 包，因为我们将使用它来存储消息历史。

pip install langchain-community

这里的关键部分是定义 get_session_history 函数。这个函数应该接受一个 session_id，并返回一个消息记录对象。session_id 用于区分不同的对话，并在调用模型时作为 config 的一部分传入。

为了演示，我们使用一个简单的内存字典作为存储后端（生产环境建议使用 Redis 或数据库）：

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

store = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

with_message_history = RunnableWithMessageHistory(
    model,
    get_session_history,
    input_messages_key="messages",
    output_messages_key="output"
)

API 调用测试

创建一个 config，每次都要将其传递给可运行的对象。这个配置应该包含的信息包括 session_id：

config = {"configurable": {"session_id": "abc2"}}

response = with_message_history.invoke(
    [HumanMessage(content="Hi! I'm Bob")],
    config=config,
)
print(response.content)

再次询问名字：

response = with_message_history.invoke(
    [HumanMessage(content="What's my name?")],
    config=config,
)
print(response.content)

我们的聊天机器人拥有了记忆。如果更改配置使用不同的 session_id，我们会发现它会重新开始对话，也就是说这里的上下文记忆是根据 session_id 进行隔离的。

config = {"configurable": {"session_id": "abc3"}}
response = with_message_history.invoke(
    [HumanMessage(content="What's my name?")],
    config=config,
)
print(response.content)

输出将是：

I'm sorry, I do not have the ability to know your name unless you tell me.

然而，因为我们将历史对话保存在数据库中，我们可以随时回到原始的对话。

config = {"configurable": {"session_id": "abc2"}}
response = with_message_history.invoke(
    [HumanMessage(content="What's my name?")],
    config=config,
)
print(response.content)

输出恢复为：

Your name is Bob.

这样我们的聊天机器人可以与许多用户进行多轮对话了！

提示词模板设计

现在，我们所做的只是在模型交互前后添加了一个简单的持久化层。我们可以通过添加提示模板来使其变得更复杂和个性化。

系统消息注入

提示模板有助于将输入信息转化为 LLM 可以处理的格式。在这种情况下，原始用户输入只是一条消息，我们将其传递给 LLM。如果我们希望添加一条带有一些自定义说明的系统消息，应该怎么做？

首先，让我们添加一个系统消息。为此，我们将创建一个 ChatPromptTemplate。我们将利用 MessagesPlaceholder 来传递所有的消息。

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    (
        "system",
        "You are a helpful assistant. Answer all questions to the best of your ability."
    ),
    MessagesPlaceholder(variable_name="messages")
])
chain = prompt | model

API 调用

请注意，这里略微改变了输入类型 - 我们现在不再通过消息列表进行传递，而是通过一个包含消息列表的 messages 字典进行传递。

response = chain.invoke({"messages": [HumanMessage(content="hi! I'm bob")]})
print(response.content)

现在我们可以将这个 chain 放在之前的消息历史对象中：

with_message_history = RunnableWithMessageHistory(chain, get_session_history)
config = {"configurable": {"session_id": "abc5"}}

response = with_message_history.invoke(
    [HumanMessage(content="Hi! I'm Jim")],
    config=config,
)
print(response.content)

多语言支持

现在只能使用英文对话，如果我们需要大模型返回指定语言中文，我们可以把提示模板稍微复杂化一点。

prompt = ChatPromptTemplate.from_messages([
    (
        "system",
        "You are a helpful assistant. Answer all questions to the best of your ability in {language}."
    ),
    MessagesPlaceholder(variable_name="messages")
])
chain = prompt | model

我们需要在模板中加入一个新的输入 language。现在就可以传入选择到语言去执行链。

response = chain.invoke(
    {"messages": [HumanMessage(content="hi! I'm bob")], "language": "Spanish"}
)
print(response.content)

这将返回西班牙语的回答。现在将这个上面定义的链封装在一个消息历史记录类中。这一次，因为输入中需要传入多个值，我们需要指定 input_messages_key 的名称。

with_message_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="messages",
    input_variables=["language"]
)

管理对话历史与上下文窗口

现在有一个问题，如果我们对历史对话不加以管理，消息列表将不断增长，可能会超出 LLM 的上下文窗口限制，导致成本增加或请求失败。因此，我们需要管理对话历史，限制传递的消息大小。

消息过滤策略

我们可以在加载历史消息之后，但在填充提示模板之前，加入这个限制。可以通过在提示词的前面添加一个简单的限制，然后将这个新的链包装在消息历史类中。

首先，定义一个函数，让它选择最近的 k 条消息：

from langchain_core.runnables import RunnablePassthrough

def filter_messages(messages, k=10):
    return messages[-k:]

chain = (
    RunnablePassthrough.assign(messages=lambda x: filter_messages(x["messages"]))
    | prompt
    | model
)

API 调用验证

创建一个超过 10 条消息的列表，看看它是否不再记得前面消息中的信息。

messages = [
    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!"),
]

response = chain.invoke(
    {
        "messages": messages + [HumanMessage(content="what's my name?")],
        "language": "English",
    }
)
print(response.content)

由于只保留了最近 10 条，早期的 "hi! I'm bob" 可能被过滤掉，导致模型忘记名字。

但如果我们询问最近十条信息中的内容，它依然记得。

response = chain.invoke(
    {
        "messages": messages + [HumanMessage(content="what's my fav ice cream")],
        "language": "English",
    }
)
print(response.content)

输出：

You mentioned that you like vanilla ice cream.

生产环境优化建议

在实际生产中，除了简单的截断，还可以考虑以下策略：

1. 摘要历史：使用另一个小模型将早期对话总结成一段文本，保留关键信息但减少 Token 消耗。
2. 向量检索：将历史消息向量化，根据当前问题检索最相关的历史片段，而非仅按时间顺序截取。
3. 动态阈值：根据模型的最大上下文窗口动态调整 k 值，确保不超过限制。

流式输出

现在我们有了一个聊天机器人功能。然而，对于聊天机器人应用程序来说，一个非常重要的用户体验考虑因素是流式传输。LLMs 有时需要一段时间才能做出回应，因此为了改进用户体验，大多数应用程序会将生成的每个令牌都流式传输回去。这可以让用户实时看到回复的生成过程。

所有的链都提供了一个 .stream 方法。我们可以简单地使用该方法来获取一个流式响应。

config = {"configurable": {"session_id": "abc15"}}
for r in with_message_history.stream(
    {
        "messages": [HumanMessage(content="hi! I'm todd. tell me a joke")],
        "language": "English",
    },
    config=config,
):
    print(r.content, end="|", flush=True)

输出效果如下：

|Sure|,| Todd|!| Here|'s| a| joke| for| you|:
|Why| don|'t| scientists| trust| atoms|?
|Because| they| make| up| everything|!||

这种流式处理方式显著降低了用户的等待焦虑感，特别是在长文本生成场景下。

最佳实践与扩展

错误处理

在生产环境中，网络波动或 API 超时是常见现象。建议在调用链外层包裹 try-except 块，并实现重试机制。

import time
from langchain_core.exceptions import OutputParserException

try:
    response = chain.invoke(input_data)
except Exception as e:
    print(f"Error occurred: {e}")
    # 实现指数退避重试逻辑
    time.sleep(2 ** attempt)

安全性考量

1. 输入过滤：对用户输入进行敏感词过滤，防止注入攻击。
2. 输出审核：对模型输出进行审核，确保不包含不当内容。
3. 密钥管理：不要将 API Key 硬编码在代码中，务必使用环境变量或密钥管理服务。

性能监控

使用 LangSmith 等工具监控 Token 消耗、延迟和错误率。通过分析日志，可以发现哪些 Prompt 模板效率低下，或者哪些会话导致了异常高的 Token 使用量。

总结

本文详细介绍了如何在 LangChain 中创建聊天机器人的基础知识。通过调用 Chat Models 大语言模型，使用 Prompt Templates 提示词模板和 Chat History 管理对话历史，一步一步地实现了一个可以记忆历史对话，支持流式输出的聊天机器人。

我们探讨了从基础的状态无模型到引入消息历史记录，再到提示词工程优化和上下文窗口管理的完整流程。此外，还补充了关于生产环境部署的最佳实践，包括错误处理、安全性和性能监控。掌握这些技能，开发者可以快速构建出既智能又稳健的对话应用。