使用最强大的AI开发工具 LangChain 构建多人聊天机器人

概述

本章节我们通过一个示例讲解如何设计和实现一个基于大语言模型（LLM）的聊天机器人。这个聊天机器人能够进行对话，并且记住之前与聊天模型的交互内容。

准备

Jupyter Notebook

本教程（以及LangChain系列中的大多数其他章节教程）使用的是 Jupyter Notebook。Jupyter Notebook 非常适合学习如何与 LLM 系统协作，因为很多时候可能会出现问题（如输出异常、API 故障等），而在交互式环境中学习和操作可以更好理解这些内容。

python 库安装

本教程需要使用 langchain-core 和 langgraph。要求 langgraph 的版本为 0.2.28 或更高。

pip install langchain-core langgraph>0.2.27

LangSmith

使用 LangChain 构建的许多应用程序通常包含多个步骤，并多次调用 LLM。随着应用的复杂度不断增加，能够深入了解链（chain）或代理（agent）内部发生了什么帮助我们更好的理解和定位问题。最好的方法就是使用LangSmith.。

在你通过上面的链接注册之后，需要先设置环境变量，才能使用记录跟踪信息：

export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="..."

如果实在Notebook里面，可以像下面这样设置

import getpass
import os

os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = getpass.getpass()

快速开始

首先，让我们学习如何单独使用语言模型。LangChain支持多种语言模型，你可以根据需要互换使用——在下面选择你想要使用的模型即可！

pip install -qU "langchain[openai]"

import getpass
import os

if not os.environ.get("OPENAI_API_KEY"):
  os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter API key for OpenAI: ")

from langchain.chat_models import init_chat_model
model = init_chat_model("gpt-4o-mini", model_provider="openai")

我们先直接使用模型。ChatModels 是 LangChain 中 “Runnable” 的实例，它们提供了一个标准接口用于模型交互。
如果只是简单地调用模型，可以将一组消息列表传递给 .invoke 方法。

from langchain_core.messages import HumanMessage


model.invoke([HumanMessage(content="Hi! I'm Bob")])

AIMessage(content='Hi Bob! How can I assist you today?', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 10, 'prompt_tokens': 11, 'total_tokens': 21, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-mini-2024-07-18', 'system_fingerprint': 'fp_0705bf87c0', 'finish_reason': 'stop', 'logprobs': None}, id='run-5211544f-da9f-4325-8b8e-b3d92b2fc71a-0', usage_metadata={'input_tokens': 11, 'output_tokens': 10, 'total_tokens': 21, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})

模型本身没有任何“状态”概念。例如，如果你接着再问问题：

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

AIMessage(content="I'm sorry, but I don't have access to personal information about users unless it has been shared with me in the course of our conversation. How can I assist you today?", additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 34, 'prompt_tokens': 11, 'total_tokens': 45, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-mini-2024-07-18', 'system_fingerprint': 'fp_0705bf87c0', 'finish_reason': 'stop', 'logprobs': None}, id='run-a2d13a18-7022-4784-b54f-f85c097d1075-0', usage_metadata={'input_tokens': 11, 'output_tokens': 34, 'total_tokens': 45, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})

我们可以看到，模型没有将之前的对话轮次纳入上下文，因此无法回答这个问题。这会导致非常糟糕的聊天体验！

为了解决这个问题，我们需要将整个对话历史传递给模型。
让我们看看这样做会发生什么：

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

AIMessage(content='Your name is Bob! How can I help you today, Bob?', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 33, 'total_tokens': 47, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-mini-2024-07-18', 'system_fingerprint': 'fp_0705bf87c0', 'finish_reason': 'stop', 'logprobs': None}, id='run-34bcccb3-446e-42f2-b1de-52c09936c02c-0', usage_metadata={'input_tokens': 33, 'output_tokens': 14, 'total_tokens': 47, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})

现在我们可以看到，模型给出了一个更好的回应！

这就是支持聊天机器人进行对话交互的基本原理。那么，我们该如何更好地实现这一点呢？

消息持久化（Message persistence）

LangGraph 实现了内置的持久化层，非常适合支持多轮对话的聊天应用。

将我们的聊天模型封装在一个最小化的 LangGraph 应用中，可以自动持久化消息历史，从而简化多轮对话应用的开发。

LangGraph自带一个简单的内存检查点工具（in-memory checkpointer），向下面这样使用。

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)

现在我们需要创建一个配置（config），在每次调用 runnable 时传入。这个配置包含一些虽然不是输入信息但有用的信息。

在本例中，我们希望包含一个 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

==================================[1m Ai Message [0m==================================

Hi Bob! How can I assist you today?

query = "What's my name?"

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

==================================[1m Ai Message [0m==================================

Your name is Bob! How can I help you today, Bob?

我们的聊天机器人现在能记住关于我们的信息了。如果我们把配置里的 thread_id 改成另一个不同的值，就会看到它从新的对话开始，不带任何之前的上下文。

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

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

==================================[1m Ai Message [0m==================================

I'm sorry, but I don't have access to personal information about you unless you've shared it in this conversation. How can I assist you today?

不过，由于我们将对话持久化到了数据库中，随时都可以回到最初的那场对话继续交流。

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

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

==================================[1m Ai Message [0m==================================

Your name is Bob. What would you like to discuss today?

这就是我们如何支持聊天机器人与多个用户进行多场对话的方式！

对于异步支持，需要将 call_model 节点更新为异步函数，并在调用应用程序时使用 .ainvoke：

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

# Define graph as before:
workflow = StateGraph(state_schema=MessagesState)
workflow.add_edge(START, "model")
workflow.add_node("model", call_model)
app = workflow.compile(checkpointer=MemorySaver())

# Async invocation:
output = await app.ainvoke({"messages": input_messages}, config)
output["messages"][-1].pretty_print()

目前，我们所做的只是给模型加了一个简单的持久化层。接下来，我们可以通过加入提示模板，让聊天机器人变得更复杂、更个性化。

提示模板（Prompt Templates）

提示模板将原始用户信息转换成 LLM 可以处理的格式。上面，原始用户输入只是一个消息，我们直接传给了LLM。现在让我们把它做得更复杂一些。
首先，加入一个带有自定义指令的系统消息（但仍以消息作为输入）。然后，除了消息之外，我们还会加入更多输入内容。

要添加系统消息，我们将创建一个ChatPromptTemplate，并利用 MessagesPlaceholder 来传入所有消息。

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

==================================[1m Ai Message [0m==================================

Ahoy there, Jim! What brings ye to these waters today? Be ye seekin' treasure, knowledge, or perhaps a good tale from the high seas? Arrr!

query = "What is my name?"

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

==================================[1m Ai Message [0m==================================

Ye be called Jim, matey! A fine name fer a swashbuckler such as yerself! What else can I do fer ye? Arrr!

很好，现在让我们把提示模板弄得更复杂一点。

假设提示模板现在看起来像这样：

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

需要注意的事，我们已经在提示中添加了一个新的输入参数 language。我们的应用程序现在有两个参数——输入消息和language。我们修改下程序

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

==================================[1m Ai Message [0m==================================

!Hola, Bob! ?Cómo puedo ayudarte hoy?

请注意，整个状态都是持久化的，因此如果不需要更改，我们可以省略像 language 这样的参数：

query = "What is my name?"

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

==================================[1m Ai Message [0m==================================

Tu nombre es Bob. ?Hay algo más en lo que pueda ayudarte?

通过LangSmith，可以帮助你了解内部的过程。

未完待续，下一节，我们继续深入探索该例子。

我是一名有十年以上经验的Java老码农，曾经沉迷于代码的世界，也曾在传统业务系统中摸爬滚打。但时代在变，AI 正在重塑技术格局。我不想被浪潮甩在身后，所以选择重新出发，走上 AI 学习与转型的旅程。

这个公众号，记录的不是鸡汤，也不是“割韭菜”的教程，而是我一个程序员真实的思考、学习、实战经验，以及从困惑到突破的全过程。

如果你也是在技术瓶颈中思考转型、想了解 AI 如何与传统开发结合、又或仅仅想看一个普通工程师的进化之路，欢迎关注，一起探索，一起成长。

关注我 和我一起，紧跟着AI的步伐，不被时代抛弃。