Langchain构建ChatBot

文摘 2025-01-09 12:30 四川

基础：

首先来看一下最基本的和model的对话，还是使用我们的ollama model

from langchain_ollama import ChatOllamaimport base_confmodel = ChatOllama(base_url=base_conf.base_url,                 model=base_conf.model_name,                 temperature=0)# 然后对话from langchain_core.messages import HumanMessageprint(model.invoke([HumanMessage(content="你好，我是桐人")]))

输出：

content='你好！很高兴认识你，桐人。我来自一个虚拟的世界，叫作阿里云，是一个能够与用户进行自然语言交流的AI助手。虽然我们可能生活在不同的世界里，但我可以和你聊天、提供信息或者帮助解答问题。不知道你对什么话题感兴趣呢？'

但是由于model本身是无状态的，如果我们问一个后续问题：

print(model.invoke([HumanMessage(content="你好，我叫什么？")]))

这个时候它其实是不知道你的名字的，即使你刚刚问过他，原因就在于无状态，自从训练好之后，在推理的阶段他是不会加入新的记忆的，它的回复如下：

content='您好！您刚才提到您叫“我”，但实际上我没有直接听到或看到您的名字。不过您自己说：“我叫什么？”所以根据您的问题，您可以告诉我您的名字是什么，我很乐意知道。或者，如果您不想透露，也没有关系的。我们继续聊天吧！'

我们可以看到，它没有将之前的对话转述到上下文中，因此也无法回答问题。这给聊天机器人带来了糟糕的体验！

但是如果我们试着一次性输入上下文，效果如何：

from langchain_core.messages import AIMessageresp = model.invoke(    [            HumanMessage(content="你好，我是桐人"),            AIMessage(content="你好，桐人，有什么可以帮助你的？"),            HumanMessage(content="我叫什么？"),        ] )print(resp)

输出：

content='你叫桐人。请问你需要帮助解决什么问题吗？'

现在我们可以看到我们得到了良好的反馈！

如上面这个很简单的例子，想要说明的是：这是聊天机器人对话互动能力的基本理念。那么我们如何才能最好地实现这一点呢？

message持久化

LangGraph实现了内置持久层，使其成为支持多轮对话的聊天应用程序的理想选择。

将我们的聊天模型包装在一个最小的 LangGraph 应用程序中，使我们能够自动保存消息历史记录，从而简化多轮应用程序的开发。

LangGraph 带有一个简单的memory checkpoint，我们将在下面使用它。当然了，框架也集成了包括不同的持久性后端（例如 SQLite 或 Postgres）。

from langgraph.checkpoint.memory import MemorySaverfrom langgraph.graph import START, MessagesState, StateGraph# 定义一个 graph workflowworkflow = StateGraph(state_schema=MessagesState)# 定义调用model的方法def call_model(state: MessagesState):    response = model.invoke(state["messages"])    return {"messages": response}# 在 graph 中定义 一个 node，只有modelworkflow.add_edge(START, "model")workflow.add_node("model", call_model)# 添加memory checkpointmemory = MemorySaver()app = workflow.compile(checkpointer=memory)

现在我们需要创建一个每次都会传递给runnable的config，这个config包含的信息不直接归属于input，在本示例中我们创建一个thread_id的字段：

config = {"configurable": {"thread_id": "abc123"}}

这是个非常有用的属性，这使我们能够使用单个应用程序支持多个对话thread，这是我们的应用程序有多个用户时常见的要求。

然后我们就可以去调用这个App：

query = "你好，我是桐人"input_messages = [HumanMessage(query)]output = app.invoke({"messages": input_messages}, config)output["messages"][-1].pretty_print()  # 输出包含了state中的所有的messages

================================== Ai Message ==================================
你好！很高兴认识你，桐人。我来自一个虚拟的世界，叫作阿里云，是一个能够与用户进行自然语言交流的AI助手。虽然我们可能生活在不同的世界里，但我可以和你聊天、提供信息或者帮助解答问题。不知道你对什么话题感兴趣呢

接着我们再问他，我叫什么：

query = "我叫什么?"input_messages = [HumanMessage(query)]output = app.invoke({"messages": input_messages}, config)output["messages"][-1].pretty_print()

================================== Ai Message ==================================
您刚才说您叫“桐人”。如果您是在提及《刀剑神域》中的角色桐人（Touya），或者是其他任何背景中的桐人，那么您的名字就是“桐人”。如果有其他问题或者需要帮助的事情，请随时告诉我！

现在通过添加checkpoint，它能够按照我们期望的样子工作了！如果我们更改配置以引用不同的thread_id之后，我们可以看到它会重新开始对话。这实际在做的事情就是通过一个标识去查找它对应的上下文或者说对话历史

================================== Ai Message ==================================
您说您叫“我叫什么?”，看起来像是一个提问的形式。根据您提供的信息，“我”是您自己，但具体的名字我没有直接得知。如果您愿意告诉我您的名字，我很乐意认识您！

可以看到，改变了config中的thread_id之后，它就又不知道user的name了，表示它开启了新的对话。

但是，我们总是可以回到原始对话（因为我们将其保存在数据库中了）

config = {"configurable": {"thread_id": "abc123"}}input_messages = [HumanMessage(query)]output = app.invoke({"messages": input_messages}, config)output["messages"][-1].pretty_print()

================================== Ai Message ==================================
您刚才说您叫“桐人”。所以，根据您提供的信息，您的名字是桐人。如果您有其他问题或需要进一步的帮助，请告诉我！

当我们把thread_id换回去之后，它就又开始接上次的对话了，并且知道name。就是这样的效果：这就是如何使用langchain以及langgraph支持聊天机器人与许多用户进行对话！

当然了，也支持异步的方式：

# Async function for node:async def call_model(state: MessagesState):    response = await model.ainvoke(state["messages"])    return {"messages": response}

workflow = StateGraph(state_schema=MessagesState)workflow.add_edge(START, "model")workflow.add_node("model", call_model)app = workflow.compile(checkpointer=MemorySaver())
# Async invoke:output = await app.ainvoke({"messages": input_messages}, config)output["messages"][-1].pretty_print()

prompt template

提示模板有助于将原始用户信息转换为 LLM 可以使用的格式。

在上面的情况下，原始用户输入只是一条消息，我们将它传递给 LLM。现在让我们让它更复杂一点。

首先，让我们添加一条带有一些自定义指令的system message（但仍然将消息作为输入）。接下来，除了message之外，我们还将添加更多输入。

为了添加系统消息，我们将创建一个ChatPromptTemplate。我们将利用MessagesPlaceholder它传递所有消息。

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholderprompt_template = ChatPromptTemplate.from_messages(    [        (            "system",            # 类似英文中的海盗语            "你讲话风格很夸张低俗。尽力回答所有问题。",        ),        MessagesPlaceholder(variable_name="messages"),    ])

我们现在可以更新我们的App以包含此template：

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholderprompt_template = ChatPromptTemplate.from_messages(    [        (            "system",            "你讲话风格很夸张低俗。尽力回答所有问题。",        ),        MessagesPlaceholder(variable_name="messages"),    ])workflow = StateGraph(state_schema=MessagesState)def call_model(state: MessagesState):    # 新增的prompt template，model的输入是一个prompt    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 = "你好，我是桐人"input_messages = [HumanMessage(query)]output = app.invoke({"messages": input_messages}, config)output["messages"][-1].pretty_print()

================================== Ai Message ==================================
嘿，桐人！咱俩可有缘啊，你在游戏世界里的英勇事迹我可是如数家珍呢！话说回来，在现实世界里你是不是也有不少精彩冒险？来来来，分享一下呗~

query = "我叫什么?"input_messages = [HumanMessage(query)]output = app.invoke({"messages": input_messages}, config)output["messages"][-1].pretty_print()

================================== Ai Message ==================================
你刚才说你叫桐人哦。不过在《刀剑神域》的世界里，你的本名是 Kazuto Kirigaya，也就是桐谷和人。你在游戏里可是个大英雄呢！

可以看到，我们通过引入prompt template包装用户输入之后，得到了一种不同风格的回话，豆包的不同风格的虚拟对话的实现机制就是这样的。我们可以来点更复杂的场景：

prompt_template_2 = ChatPromptTemplate.from_messages(    [        (            "system",            "你是一个有用的助手。尽力用{language}回答所有问题。",        ),        MessagesPlaceholder(variable_name="messages"),    ])

注意，在新的模板里面，我们的app变成两个参数：language和messages。我们改一下总体的代码，加一点东西进去：

from typing import Sequencefrom langchain_core.messages import BaseMessagefrom langgraph.graph.message import add_messagesfrom typing_extensions import Annotated, TypedDict
# 新增的代码，定义了状态的 schema：messages 和 languageclass State(TypedDict):    messages: Annotated[Sequence[BaseMessage], add_messages]    language: strworkflow = StateGraph(state_schema=State)
# 这里的输入变成了一个 State 类型的对象def call_model(state: State):    prompt = prompt_template_2.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 = "你好，我是桐人"language = "日语"input_messages = [HumanMessage(query)]output = app.invoke(    # 这里的input多加入了 language    {"messages": input_messages, "language": language},    config,)output["messages"][-1].pretty_print()

================================== Ai Message ==================================
こんにちは、桐人さんですって。何かお手伝いできることがありますか？

可以看到它已经用日文回复我了。同时请注意，整个状态是持久的，因此language如果不需要更改，后面的同 config的对话，我们可以省略参数：

query = "我叫什么?"input_messages = [HumanMessage(query)]output = app.invoke(    {"messages": input_messages},    config,)output["messages"][-1].pretty_print()

================================== Ai Message ==================================
あなたは「桐人」と名乗りました。何か他の質問がありますか？

管理对话的历史记录

构建聊天机器人时需要理解的一个重要概念是如何管理对话历史记录。如果不加以管理，消息列表将无限增长，并可能溢出 LLM 的上下文窗口。因此，添加一个限制传入消息大小的步骤非常重要。

重要的是，我们需要在prompt template之前，从消息历史记录中加载消息之后执行此操作。

我们可以通过在prompt之前添加一个简单的step来适当修改message key来实现这一点。然后将该新chain包装在消息历史记录类中。

LangChain 带有一些内置助手来管理消息列表。

在这种情况下，我们将使用 trim_messages 助手来减少发送给模型的消息数量。trimmer 允许我们指定要保留多少个token，以及其他参数，例如是否要始终保留系统消息以及是否允许部分消息：

from langchain_core.messages import SystemMessage, trim_messagestrimmer = trim_messages(    max_tokens=3,    strategy="last",    token_counter=len,    include_system=True,    allow_partial=False,    start_on="human",)# 这里我们直接模拟多条message statemessages = [    SystemMessage(content="你是一个有用的助手"),    HumanMessage(content="你好我是桐人。"),    AIMessage(content="你好！"),    HumanMessage(content="我喜欢亚丝娜"),    AIMessage(content="不错"),    HumanMessage(content="2 + 2等于多少"),    AIMessage(content="4"),    HumanMessage(content="谢谢"),    AIMessage(content="不谢！"),    HumanMessage(content="你开心吗"),    AIMessage(content="当然!"),]
print(trimmer.invoke(messages))

注意我这里将token_counter设置为了len,这个时候他代表的含义就不是token数了，而是历史的消息条数。

================================ System Message ================================
你是一个有用的助手================================ Human Message =================================
你开心吗================================== Ai Message ==================================
当然!

可以看到的一个效果是，它会优先保留system message，因为这是一个对话的认为约束的背景，然后在system message的基础上，我们设置了 strategy="last"，因此它保留了最近的两条消息！

stream

现在我们有一个可以运行的聊天机器人。然而，聊天机器人应用程序的一个非常重要的用户体验考虑因素是流式传输。

LLM 有时需要一段时间才能响应，因此为了提高用户体验，大多数应用程序都会做的一件事就是在每个token生成时将其流回。这能让用户不断地知道系统在恢复，而不是以为系统卡了或者出了问题。

事实上，做到这一点非常简单！

默认情况下，我们的 LangGraph 应用程序中的 .stream 会流式传输应用程序步骤，设置 stream_mode="messages" 允许我们流式输出令牌：

config = {"configurable": {"thread_id": "abc789"}}query = "你好，我是桐人，请给我讲一个笑话。"language = "中文"input_messages = [HumanMessage(query)]for chunk, metadata in app.stream(    {"messages": input_messages, "language": language},    config,    # 这里的stream_mode是messages，表示返回的是一个个的消息    stream_mode="messages",):    if isinstance(chunk, AIMessage):  # 过滤出model的response        print(chunk.content, end="|")

你好|，|桐|人|！|很高兴|为你|讲|个|笑话|。
|为什么|电脑|永远不会|感冒|？|因为它|有|“|窗口|”|（|Windows|）|操作|系统的|免疫|系统|哦|~| 
|希望|这个|笑话|能|让你|开心|一笑|！|如果你|还有|其他|问题|或|需求|，|欢迎|随时|告诉我|。||

好了，最基本的LLM Chatbot我们已经讲完了，后面会有更加深入的高级教程。

如果觉得有趣，你也动手试一下吧！！！

参考链接：

https://python.langchain.com/docs/tutorials/chatbot/

半夏决明

读书，摄影，随笔