LangChain 记忆组件深度解析：Chain 组件与 Runnable 深入学习

标签：Runnable Chain chain langchain print query 组件 import

在构建复杂的 AI 应用时，有效管理对话历史和上下文信息至关重要。LangChain 框架提供了多种记忆组件，使得开发者能够轻松实现具有记忆功能的聊天机器人。本文将深入探讨 LangChain 中的记忆组件、Chain 组件以及 Runnable 接口，帮助开发者更好地理解和使用这些强大的工具。

LangChain 缓冲记忆组件的使用与解析

缓冲记忆组件的类型

LangChain 提供了多种缓冲记忆组件，每种组件都有其特定的用途和优势：

ConversationBufferMemory：最简单的缓冲记忆，将所有对话信息全部存储作为记忆。
ConversationBufferWindowMemory：通过设定 k 值，只保留一定数量（2*k）的对话信息作为历史。
ConversationTokenBufferMemory：通过设置最大标记数量（max_token_limits）来决定何时清除交互信息，当对话信息超过 max_token_limits 时，抛弃旧对话信息。
ConversationStringBufferMemory：等同于缓冲记忆，固定返回字符串（这是早期 LangChain 封装的记忆组件）。

缓冲窗口记忆示例

以下是一个使用 ConversationBufferWindowMemory 实现 2 轮对话记忆的示例：

from operator import itemgetter
import dotenv
from langchain.memory import ConversationBufferWindowMemory
from langchain_community.chat_message_histories import FileChatMessageHistory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()

memory = ConversationBufferWindowMemory(
    input_key="query",
    return_messages=True,
    k=2,
)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是OpenAI开发的聊天机器人，请帮助用户解决问题"),
    MessagesPlaceholder("history"),
    ("human", "{query}")
])

llm = ChatOpenAI(model="gpt-3.5-turbo-16k")

chain = RunnablePassthrough.assign(
    history=RunnableLambda(memory.load_memory_variables) | itemgetter("history")
) | prompt | llm | StrOutputParser()

while True:
    query = input("Human: ")
    if query == "q":
        exit(0)
    chain_input = {"query": query}
    print("AI: ", flush=True, end="")
    response = chain.stream(chain_input)
    output = ""
    for chunk in response:
        output += chunk
        print(chunk, flush=True, end="")
    print("\nhistory:", memory.load_memory_variables({}))
    memory.save_context(chain_input, {"output": output})

LangChain 摘要记忆组件的使用与解析

摘要记忆组件的类型

LangChain 提供了两种主要的摘要记忆组件：

ConversationSummaryMemory：将传递的历史对话记录总结成摘要进行保存，使用时填充的记忆为摘要，而非对话数据。
ConversationSummaryBufferMemory：在不超过 max_token_limit 的限制下保存对话历史数据，对于超过的部分进行信息的提取与总结。

摘要缓冲混合记忆示例

以下是一个使用 ConversationSummaryBufferMemory 的示例，限制 max_token_limit 为 300：

from operator import itemgetter
import dotenv
from langchain.memory import ConversationSummaryBufferMemory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个强大的聊天机器人，请根据对应的上下文回复用户问题"),
    MessagesPlaceholder("history"),
    ("human", "{query}"),
])
memory = ConversationSummaryBufferMemory(
    return_messages=True,
    input_key="query",
    llm=ChatOpenAI(model="gpt-3.5-turbo-16k"),
    max_token_limit=300,
)

llm = ChatOpenAI(model="gpt-3.5-turbo-16k")

chain = RunnablePassthrough.assign(
    history=RunnableLambda(memory.load_memory_variables) | itemgetter("history")
) | prompt | llm | StrOutputParser()

while True:
    query = input("Human: ")
    if query == "q":
        exit(0)
    chain_input = {"query": query, "language": "中文"}
    response = chain.stream(chain_input)
    print("AI: ", flush=True, end="")
    output = ""
    for chunk in response:
        output += chunk
        print(chunk, flush=True, end="")
    memory.save_context(chain_input, {"output": output})
    print("")
print("history: ", memory.load_memory_variables({}))

使用摘要记忆组件时，需要注意一些潜在的问题，如多个 System 角色消息的处理和某些聊天模型对消息格式的特殊要求。

LangChain 实体记忆组件的使用与解析

实体记忆组件用于跟踪对话中提到的实体，并记住关于特定实体的既定事实。LangChain 提供了 ConversationEntityMemory 类来实现这一功能。

以下是一个使用 ConversationEntityMemory 的示例：

import dotenv
from langchain.chains.conversation.base import ConversationChain
from langchain.memory import ConversationEntityMemory
from langchain.memory.prompt import ENTITY_MEMORY_CONVERSATION_TEMPLATE
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()
llm = ChatOpenAI(model="gpt-4o", temperature=0)
chain = ConversationChain(
    llm=llm,
    prompt=ENTITY_MEMORY_CONVERSATION_TEMPLATE,
    memory=ConversationEntityMemory(llm=llm),
)
print(chain.invoke({"input": "你好，我最近正在学习LangChain。"}))
print(chain.invoke({"input": "我最喜欢的编程语言是 Python。"}))
print(chain.invoke({"input": "我住在广州"}))
res = chain.memory.entity_store.store
print(res)

记忆组件的持久化与第三方集成

LangChain 的记忆组件本身没有持久化功能，但可以通过 chat_memory 来将对话信息历史持久化。LangChain 集成了近 50 多个第三方对话消息历史存储方案，包括 Postgres、Redis、Kafka、MongoDB、SQLite 等。

内置 Chain 组件的使用与解读

Chain 简介与使用

在 LangChain 中，Chain 用于将多个组件（如 LLM、提示词模板、向量存储、记忆、输出解析器等）串联起来一起使用。LangChain 支持两种类型的链：

使用 LCEL（LangChain Expression Language）构建的链
通过 Chain 类子类构建的链（遗产链）

以下是使用 LLMChain 的基础示例：

import dotenv
from langchain.chains.llm import LLMChain
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()

llm = ChatOpenAI(model="gpt-3.5-turbo-16k")
prompt = ChatPromptTemplate.from_template("请讲一个关于{subject}的冷笑话")

chain = LLMChain(
    llm=llm,
    prompt=prompt,
)

print(chain("程序员"))
print(chain.run("程序员"))
print(chain.apply([{"subject": "程序员"}]))
print(chain.generate([{"subject": "程序员"}]))
print(chain.predict(subject="程序员"))
print(chain.invoke({"subject": "程序员"}))

内置的 Chain

LangChain 提供了多种内置的 Chain，包括 LCEL Chain 和遗产 Chain。例如，create_stuff_documents_chain 函数可以创建一个文档对话链：

import dotenv
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个强大的聊天机器人，能根据对应的上下文信息回复用户问题。\n\n<context>{context}</context>"),
    ("human", "{query}"),
])

llm = ChatOpenAI(model="gpt-3.5-turbo-16k")

chain = create_stuff_documents_chain(llm=llm, prompt=prompt)

documents = [
    Document(page_content="小明喜欢绿色，但不喜欢黄色"),
    Document(page_content="小王喜欢粉色，也有一点喜欢红色"),
    Document(page_content="小泽喜欢蓝色，但更喜欢青色")
]

resp = chain.invoke({"query": "大家都喜欢什么颜色", "context": documents})

print(resp)

RunnableWithMessageHistory 简化代码与使用

RunnableWithMessageHistory 是一个包装器，可以让链自动处理历史消息的填充和存储过程。

以下是一个使用 RunnableWithMessageHistory 的示例：

import dotenv
from langchain_community.chat_message_histories import FileChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()

store = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = FileChatMessageHistory(f"chat_history_{session_id}.txt")
    return store[session_id]

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个强大的聊天机器人，请根据用户的需求回复问题。"),
    MessagesPlaceholder("history"),
    ("human", "{query}"),
])
llm = ChatOpenAI(model="gpt-3.5-turbo-16k")

chain = prompt | llm | StrOutputParser()

with_message_chain = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="query",
    history_messages_key="history",
)

while True:
    query = input("Human: ")
    if query == "q":
        exit(0)
    response = with_message_chain.stream(
        {"query": query},
        config={"configurable": {"session_id": "muxiaoke"}}
    )
    print("AI: ", flush=True, end="")
    for chunk in response:
        print(chunk, flush=True, end="")
print("")

通过使用 RunnableWithMessageHistory，我们可以更方便地管理多个用户的对话历史，并在链中自动处理历史消息的加载和存储。

结论：

LangChain 提供了丰富的记忆组件和 Chain 组件，使得开发者能够轻松构建具有上下文感知能力的 AI 应用。通过合理使用这些组件，我们可以创建更智能、更个性化的对话系统。随着 LangChain 的不断发展，LCEL 表达式和 Runnable 接口的引入进一步简化了应用的构建过程。开发者应该根据具体需求选择合适的记忆组件和 Chain 类型，以实现最佳的应用性能和用户体验。

标签：Runnable,Chain,chain,langchain,print,query,组件,import
From： https://www.cnblogs.com/muzinan110/p/18539692