[langchain教程]langchain02——用langchain构建聊天机器人

首先，让我们学习如何单独使用语言模型。
详见[langchain教程]langchain01——用langchain调用大模型

然而，单独使用语言模型存在一个问题：没有将之前的对话轮次作为上下文，因此无法回答涉及上下文的问题。
为了解决这个问题，我们需要将整个对话历史传递给模型

消息历史

在许多问答应用中，用户希望基于之前的对话进行问答，这意味着应用需要记忆过去的问题和答案，并将这些信息纳入之后的思考中。

可以使用消息历史类ChatHistory来包装模型，使其具有状态。 ChatHistory将跟踪底层链的输入和输出，并将它们作为消息附加到消息数据库中。 未来的交互将加载这些消息并将其作为输入的一部分传递给链。

使用消息历史的一个关键部分是 get_session_history 函数，这个函数用于获取或创建指定会话的聊天历史记录。
函数输入是一个 session_id ，session_id用于区分不同的对话，应该作为配置config的一部分在调用新链时传入。
而函数的输出是消息历史对象。

在下面这段代码中，创建了一个空字典 store 用于存储不同会话的聊天历史记录。每个会话将通过一个唯一的 session_id 来标识。

from langchain_core.chat_history import (
    BaseChatMessageHistory,
    InMemoryChatMessageHistory,
)
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.llms import Ollama
from langchain_core.messages import HumanMessage

model = Ollama(model="deepseek-r1:8b")

store = {}


def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
    	# InMemoryChatMessageHistory 用于在内存中存储会话历史记录
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

# RunnableWithMessageHistory 将语言模型与会话历史记录功能结合起来
with_message_history = RunnableWithMessageHistory(model, get_session_history)

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

response = with_message_history.invoke(
    [
    HumanMessage(content="Hi! I'm Bob")
    ],
    config=config,
)
#response.content：'Hi Bob! How can I assist you today?'

response = with_message_history.invoke(
    [
    HumanMessage(content="What's my name?")
    ],
    config=config,
)
#response.content ：'Your name is Bob. How can I help you today, Bob?'

如果更改配置以引用不同的 session_id，可以看到它开始新的对话。

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

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

# response.content："I'm sorry, I cannot determine your name as I am an AI assistant and do not have access to that information."

提示词模板

可以通过添加提示词模板来使消息历史变得更加复杂和个性化。

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


# 之前只有model with_message_history = RunnableWithMessageHistory(model, get_session_history)
chain = prompt | model
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)

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

这里用的是MessagesPlaceholder 消息占位符，用于在运行时动态插入用户和机器人的消息列表，variable_name="messages" 指定了占位符的名称。所以调用invoke函数时，传递的是一个包含 messages 键的字典，其中包含一系列消息，而不是直接传递消息列表。

插入消息列表后的提示词如下所示：

ChatPromptValue(
	messages=[
		SystemMessage(content='You are a helpful assistant. Answer all questions to the best of your ability.'), 			
		HumanMessage(content="hi! I'm bob")
	]
)

最后的输出如下所示：

Hi Jim! Nice to meet you. How can I assist you today?

Hi Jim! It looks like your name is Jim. How can I assist you today?

更复杂的提示词模板示例

将更复杂的链封装在一个消息历史类中时，由于输入中有多个键，需要指定正确的键来保存聊天历史。

以下是一个完整的提示词模板示例：

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.chat_history import (
    BaseChatMessageHistory,
    InMemoryChatMessageHistory,
)
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.llms import Ollama
from langchain_core.messages import HumanMessage

model = Ollama(model="deepseek-r1:8b")

store = {}


def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
    	# InMemoryChatMessageHistory 用于在内存中存储会话历史记录
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]


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

with_message_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="messages",
)

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

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

response1 = with_message_history.invoke(
    {"messages": [HumanMessage(content="whats my name?")], "language": "Spanish"},
    config=config,
)
print(response1)

输出如下所示：

¡Hola! Soy un asistente amable y dispuesto a ayudarte. ¿En qué puedo asistirte hoy?


¡Hola! No sé tu nombre todavía. ¿Puedes decirme tu nombre, por favor? Estoy aquí para ayudarte con cualquier pregunta que tengas.

管理对话历史

构建聊天机器人时，一个重要的概念是如何管理对话历史。如果不加以管理，消息列表将无限增长，溢出大语言模型的上下文窗口。

因此，需要添加一个限制传入消息大小。
需要在提示模板之后但在从消息历史加载之前的消息之后修改 messages 键，然后将该新链封装在消息历史类中来实现。

LangChain 提供了一些内置的函数来管理消息列表，使用 trim_messages 实现：

• 获取最后 n 个tokens，可以设置 max_tokens = 65， strategy="last"
• 获取最前 n 个tokens，可以设置 max_tokens = 65， strategy="first"
• 始终保留初始系统消息，可以指定 include_system=True
• 允许拆分消息的内容，可以指定 allow_partial=True
• 确保第一条消息（不包括系统消息）始终是特定类型，可以指定 start_on = 'human'

使用修剪器需要安装transformers，否则会报错ImportError: Could not import transformers python package. This is needed in order to calculate get_token_ids.

安装transformers：

pip install transformers

使用修剪器：

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

tmp = trimmer.invoke(messages)

tmp的结果如下所示：

[SystemMessage(content="you're a good assistant"),
 HumanMessage(content='whats 2 + 2'),
 AIMessage(content='4'),
 HumanMessage(content='thanks'),
 AIMessage(content='no problem!'),
 HumanMessage(content='having fun?'),
 AIMessage(content='yes!')]

要在链中使用它，只需在将 messages 输入传递给提示之前运行修剪器。

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.chat_history import (
    BaseChatMessageHistory,
    InMemoryChatMessageHistory,
)
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.llms import Ollama
from langchain_core.messages import HumanMessage,AIMessage

model = Ollama(model="deepseek-r1:8b")

store = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
    	# InMemoryChatMessageHistory 用于在内存中存储会话历史记录
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

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

from langchain_core.messages import SystemMessage, trim_messages

trimmer = trim_messages(
    max_tokens=30,
    strategy="last",
    token_counter=model,
    include_system=True,
    allow_partial=False,
    start_on="human",
)

messages = [
    SystemMessage(content="you're a good assistant"),
    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?"),
    HumanMessage(content="I'm bob"),
    AIMessage(content="hi bob!")
]

from operator import itemgetter

from langchain_core.runnables import RunnablePassthrough

chain = (
    RunnablePassthrough.assign(messages=itemgetter("messages") | trimmer)
    | prompt
    | model
)


with_message_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="messages",
)

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

"""
messages=[
SystemMessage(content='You are a helpful assistant. Answer all questions to the best of your ability in Chinese.'), 
HumanMessage(content='thanks'), 
AIMessage(content='no problem!'), 
HumanMessage(content="I'm bob"), 
AIMessage(content='hi bob!'), 
HumanMessage(content='whats my name?')
]

"""
response = with_message_history.invoke(
    {
        "messages": messages + [HumanMessage(content="whats my name?")],
        "language": "English",
    },
    config=config,
)

"""
messages=[
SystemMessage(content='You are a helpful assistant. Answer all questions to the best of your ability in Chinese.'),
HumanMessage(content='thanks'), 
AIMessage(content='no problem!'), 
HumanMessage(content="I'm bob"), 
AIMessage(content='hi bob!'), 
HumanMessage(content='what problem did i ask?')]
"""
response1 = with_message_history.invoke(
    {
        "messages": [HumanMessage(content="what math problem did i ask?")],
        "language": "English",
    },
    config=config,
)

print(response)
print(response1)

输出如下所示：

你叫Bob。

您提出的问题在对话中未明确定义。要获得准确的帮助，请提供更多的上下文或直接澄清您的问题。

流式处理

对于聊天机器人应用程序来说，一个非常重要的用户体验考虑是流式处理。大型语言模型有时可能需要一段时间才能响应。

因此，为了改善用户体验，大多数应用程序所做的一件事是随着每个令牌的生成流回。这样用户就可以看到进度。可以使用.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="|")

输出如下所示：

|Hi| Todd|!| Sure|,| here|'s| a| joke| for| you|:| Why| couldn|'t| the| bicycle| find| its| way| home|?| Because| it| lost| its| bearings|!| 😄||