[LangChain]Build a Chatbot

0 Useful Docs

• [Build a Chatbot](Build a Chatbot | 🦜️🔗 LangChain)

1 Setup

pip install langchain-core langgraph>0.2.27

2 Quick Start

模型本身并不会保存当前会话的状态，这对于chatbot而言是很糟糕的体验。

因此， 我们需要将整个conversation history传递给模型。

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

3 Message persistence

LangGraph实现了一个内置的持久化层，使其成为支持多次对话交互的聊天应用的理想选择。

通过将聊天模型封装在一个最小化的LangGraph应用中，可以自动保存消息历史记录，简化了多轮对话应用的开发过程。

LangGraph配备了一个简单的内存检查点机制。更多详情，请参阅 documentation ，包括如何使用不同的持久化后端（例如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

4 Prompt templates

Prompt Templates有助于将原始用户信息转化为大型语言模型（LLM）能够处理的格式。

可以通过prompt templates添加一些system message。

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

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()

可以像模板中添加额外的参数

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

现在程序有两个参数 messages 和 language，因此需要更新State

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)

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()

由于整个state都被保存了，因此如果某个参数没有改变，可以直接忽略

query = "What is my name?"

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

5 Managing Conversation History

在构建聊天机器人时，理解如何管理对话历史记录是一个重要的概念。若不加以管理，消息列表将无限制增长，并可能超出LLM的上下文窗口范围。因此，在传递消息之前限制其大小至关重要。

重要的是，这一步骤应在应用提示模板之前，但在从消息历史记录加载先前消息之后执行。我们可以通过在提示前添加一个步骤来适当地修改消息，然后将这个新链封装在消息历史记录类中实现这一目标。

LangChain 提供了一些内置的用于管理消息列表（managing a list of messages）的函数。这里，我们使用 trim_messages 辅助函数来减少发送给模型的消息数量。该修剪器允许指定希望保留的token数量，以及其他参数，如是否总是保留系统消息以及是否允许部分消息存在等。

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参考：trim_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)

6 Streaming

对于聊天机器人应用而言，一个非常重要的用户体验考量是流式传输（streaming）。由于语言模型有时需要一段时间才能生成回复，因此大多数应用程序会将每个生成的token即时流式传输给用户，以展示处理进度，从而改善用户体验。

在默认情况下，LangGraph 应用程序中的 .stream 方法会对应用程序步骤进行流式传输。通过设置 stream_mode=”messages”，我们可以改为流式传输输出 tokens：

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="|")