PythonAI算法
基于 LangChain 构建具备记忆功能的聊天机器人
如何使用 LangChain 框架开发一个支持多轮对话记忆的聊天机器人。内容涵盖环境搭建、基础模型调用、消息历史管理、提示词模板设计、上下文窗口优化以及流式输出实现。通过引入 RunnableWithMessageHistory 和自定义过滤函数，解决了模型无状态问题并有效控制了 Token 消耗，同时提供了持久化存储和错误处理的扩展建议，帮助开发者快速构建生产级对话应用。
如何使用 LangChain 框架开发一个支持多轮对话记忆的聊天机器人。内容涵盖环境搭建、基础模型调用、消息历史管理、提示词模板设计、上下文窗口优化以及流式输出实现。通过引入 RunnableWithMessageHistory 和自定义过滤函数，解决了模型无状态问题并有效控制了 Token 消耗，同时提供了持久化存储和错误处理的扩展建议，帮助开发者快速构建生产级对话应用。
本文将详细介绍如何设计和实现一个由大语言模型（LLM）驱动的聊天机器人示例。这个聊天机器人将能够进行多轮对话，并记住先前的交互内容，从而提供更自然、连贯的用户体验。
在开始编码之前，我们需要理解几个关键的高级组件：
我们将介绍如何将上述组件结合在一起，创建一个功能强大且可扩展的对话式聊天机器人。
langchain 库版本建议为 0.2.x。langchain-openai 库版本建议为 0.1.x。首先，安装核心依赖包：
pip install langchain langchain-openai langchain-community
对于生产环境，建议使用虚拟环境（如 venv 或 conda）来隔离依赖，避免版本冲突。
为了使用 OpenAI 模型，需要设置 API Key。同时，为了后续可能的调试需求，建议配置 LangSmith 环境变量。
import getpass
import os
# 设置 OpenAI API Key
os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API Key: ")
# 可选：启用 LangSmith 追踪
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = getpass.getpass("Enter your LangChain API Key: ")
首先，尝试单独使用语言模型。LangChain 支持许多不同的语言模型，您可以随意选择要使用的语言模型！这里我们以 OpenAI 为例。
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
from langchain_core.messages import HumanMessage
response = model.invoke([HumanMessage(content="Hi! I'm Bob")])
print(response.content)
输出结果类似于：
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"
)
创建一个 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
请注意，这里略微改变了输入类型 - 我们现在不再通过消息列表进行传递，而是通过一个包含消息列表的 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
)
创建一个超过 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.
在实际生产中，除了简单的截断，还可以考虑以下策略：
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)
使用 LangSmith 等工具监控 Token 消耗、延迟和错误率。通过分析日志，可以发现哪些 Prompt 模板效率低下，或者哪些会话导致了异常高的 Token 使用量。
本文详细介绍了如何在 LangChain 中创建聊天机器人的基础知识。通过调用 Chat Models 大语言模型，使用 Prompt Templates 提示词模板和 Chat History 管理对话历史，一步一步地实现了一个可以记忆历史对话，支持流式输出的聊天机器人。
我们探讨了从基础的状态无模型到引入消息历史记录，再到提示词工程优化和上下文窗口管理的完整流程。此外，还补充了关于生产环境部署的最佳实践，包括错误处理、安全性和性能监控。掌握这些技能，开发者可以快速构建出既智能又稳健的对话应用。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 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