【LangChain Chains及Memory 03】

博客园 - 尘封~~

尘封~~ · 2026-02-02 · via 博客园 - 尘封~~

一、Chains

1、Chains的基本使用

1.1、chains的基本概念

Chain：链，用于将多个组件（提示词模版、LLM模型、记忆、工具等）链接起来，形成可复用的工作流，完成复杂的工作

Chain的核心思想是通过组合不同的模块化单元，实现比单一组件更强大的功能。比如：

将LLM与Prompt Template结合
将LLM与输出解析器结合
将LLM与外部数据结合，例如用于问答
将LLM与长期记忆结合，例如用于聊天历史记录
通过将第一个LLM的输出作为第二个LLM的输入，将多个LLM按顺序结合在一起

1.2、LCEL及其基本构成

使用LCEL可以构造出结构最简单的Chain

LangChain表达式语音是一种声明式方式，可以轻松地将多个组件链接成AI工作流。它通过Python原生操作符（如：管道符 |）将组件链接成可执行流程，显著简化了AI应用的开发

LCEL的基本构成：提示词Prompt + 模型Model + 输出解释器OutputParser

# 在这个链条中，用户输入被传递给提示模板，然后提示模板的输出被传递给模型，然后模型的输出被传 递给输出解析器。
chain = prompt | model | output_parser
chain.invoke({"input":"What's your name?"})

Prompt：是一个BasePromptTemplate，这意味着它接受了一个模版变量的字典并生成一个PromptValue 。PromptValue可以传递给LLM（它以字符串作为输入）或ChatModel（它以消息序列作为输入）
Model：将PromptValue传递给model。如果我们的model是一个ChatModel，这意味着它将输出一个BaseMessage
OutputParser：将model的输出传递给outputParser，它是一个BaseOutputParser，意味着它可以接受字符串或BaseMessage作为输入
chain：我们可以使用 | 运算符轻松创建这个chain 。| 运算符在LangChain中用于将两个元素组合在一起
invoke：所有LCEL对象都实现了Runable协议，保证一致的调用方式（invoke / batch / stream）

1.3、Runnable

Runnable是LangChain定义的一个抽象接口（Protocol），它强制要求所有LCEL组件实现一组标准方法：

class Runnable(Protocol):
　　def invoke(self, input: Any) -> Any: ... # 单输入单输出
　　def batch(self, inputs: List[Any]) -> List[Any]: ... # 批量处理
　　def stream(self, input: Any) -> Iterator[Any]: ... # 流式输出
　　# 还有其他方法如 ainvoke（异步）等...

目的是每个组件调用方式统一！（无论组件【提示词/模型/工具】多么的复杂，调用方式完全相同）

举例：使用LCEL将不同的组件组合成一个单一链条【这里就用到了管道符 | 】

from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
import os
import dotenv
from langchain_core.output_parsers import StrOutputParser

dotenv.load_dotenv()
base_url = os.getenv("OPENAI_BASE_URLS")
api_key = os.getenv("OPENAI_API_KEY1")

chat = ChatOpenAI(
    model="gpt-4o-mini",
    base_url=base_url,
    api_key=api_key,
)

template = PromptTemplate.from_template(
   template='将一段{name}小品中经典的台词'
)

parser = StrOutputParser()

chain = template|chat|parser
res = chain.invoke({"name":"赵本山和宋丹丹"})
print(res)

注意：使用chain调用时，template|chat|parser这三个组件之间的先后顺序是不能变的，因为每个组件都是需要上一个组件的输出作为输入进行处理

2、基于LCEL构建的Chains的类型

下面是一些常用的函数

create_sql_query_chain
create_stuff_documents_chain
create_openai_fn_runnable
create_structured_output_runnable
load_query_constructor_runnable
create_history_aware_retriever
create_retrieval_chain

2.1、create_sql_query_chain

sql查询链，是创建生成sql查询的链，用于将自然语言转换成数据库的sql语句

from langchain.chains.sql_database.query import create_sql_query_chain
from langchain_community.utilities import SQLDatabase

# 链接mysql数据库
from urllib.parse import quote_plus

db_user = 'root'
db_password = '123456'
db_host = 'localhost'
db_port = 3307
db_name = 'test'

# mysql+pymysql://<username>:<password>@<host>:<port>/<database>
# 如果密码含特殊字符（@ : / ? 等），请用 quote_plus
uri = (
    f"mysql+pymysql://{quote_plus(db_user)}:{quote_plus(db_password)}@{db_host}:{db_port}/{db_name}"
    "?charset=utf8mb4"
)

# 建议加上 pool_pre_ping，网络/空闲连接更稳定
db = SQLDatabase.from_uri(
    uri,
    engine_args={"pool_pre_ping": True, "pool_recycle": 3600},
)

# 快速连通性测试（能返回 1 就说明驱动+连接基本 OK）
try:
    print(db.run("SELECT 1"))
except Exception as e:
    print("DB connection test failed:", repr(e))

dotenv.load_dotenv()
base_url = os.getenv("OPENAI_BASE_URLS")
api_key = os.getenv("OPENAI_API_KEY1")

chat = ChatOpenAI(
    model="gpt-4o-mini",
    base_url=base_url,
    api_key=api_key,
)

# 调用 create_sql_query_chain 创建链条
chain = create_sql_query_chain(chat, db)
res = chain.invoke({"question": "往employees表插入5条数据"})
print(res)

2.2、create_stuff_documents_chain

create_stuff_documents_chain用于将多个文档内容合并成当个长文本的链式工具，并一次性传递给LLM处理（而不是分批次处理）

适合场景：

保持上下文完整，适合需要全局理解所有文档内容的任务（如：总结、回答）
适合处理少量/中等长度文档的场景

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

# 创建大模型对象
dotenv.load_dotenv()
base_url = os.getenv("OPENAI_BASE_URLS")
api_key = os.getenv("OPENAI_API_KEY1")

chat = ChatOpenAI(
    model="gpt-4o-mini",
    base_url=base_url,
    api_key=api_key,
)

# 定义提示词模板
prompt = PromptTemplate.from_template("""
如下文档{docs}中说，香蕉是什么颜色的？
""")

# 创建链条
chain = create_stuff_documents_chain(chat, prompt,document_variable_name="docs")
# 定义输入文档
docs = [
    Document(page_content="苹果，学名Malus pumila Mill.，别称西洋苹果、柰，属于蔷薇科苹果属的植物。苹果是全球最广泛种植和销售的水果之一，具有悠久的栽培历史和广泛的分布范围。苹果的原始种群主要起源于中亚的天山山脉附近，尤其是现代哈萨克斯坦的阿拉木图地区，提供了所有现代苹果品种的基因库。苹果通过早期的贸易路线，如丝绸之路，从中亚向外扩散到全球各地"),
    Document(page_content="香蕉是白色的水果，主要产自热带地区。"),
    Document(page_content="蓝莓是蓝色的浆果，含有抗氧化物质。"),
]

# 执行链条
res = chain.invoke({"docs": docs})
print(res)

二、Memory

1、member的设计理念

输入问题：（{"request":...}）
读取历史信息：从Member中READ历史消息（{“past_message”:[...]}）
构建提示词prompt：读取到的历史消息和当前的问题会被合并，构建一个新的prompt
模型处理：构建好的提示会被传递给语言模型进行处理。语言模型根据提示生成一个输出
解析输出：输出解析器通过正则表达式regex("Answer: (.*)")来解析，返回一个回答({"answer":...})给用户
得到回复并写入Member:新生成的回答会与当前的问题一起写入memory，更新对话历史。memory回存储最新的对话内容，为后续的对话提供上下文支持

2、基础member模块的使用

2.1、Member模块的设计思路

如何设计member模块：

层次1（最直接的方式）：保留一个聊天消息列表
层次2（简单的新思路）：只返回最近交互的k条消息
层次3（稍微复杂一点）：返回过去k条消息的简洁摘要
层次4（更复杂）：从存储的消息中提取实体，并且仅返回有关当前运行中引用实体的信息

LangChain的设计：

针对上述情况，LangChain构建了一些可以直接使用Memory工具，用于存储聊天消息的一些列集成

2.2、基础：ChatMessageHistory

ChatMessageHistory是一个用于存储和管理对话消息的基础类，它直接操作消息对象（如：HumanMessage,AIMessage等），是其他记忆组件的底层存储工具 ===>本质是是一个存储器，把消息对象存储到内存中

在API文档中，ChatMessageHistory还有一个别名类：InMemoryChatMessageHistory；导包时，需使用：

from langchain.memory import ChatMessageHistory

特点：

纯粹是消息对象“存储器”，与记忆策略（如缓冲、窗口、摘要等）无关
不涉及消息的格式化（如转成文本字符串）

from langchain_openai import ChatOpenAI
import os
import dotenv
from langchain.memory import ChatMessageHistory

# 创建大模型对象
dotenv.load_dotenv()
base_url = os.getenv("OPENAI_BASE_URLS")
api_key = os.getenv("OPENAI_API_KEY1")

chat = ChatOpenAI(
    model="gpt-4o-mini",
    base_url=base_url,
    api_key=api_key,
)

# 1、实例化ChatMessageHistory对象
history=ChatMessageHistory()
# 2、添加userMessage和AIMessage
history.add_user_message("你好！")
history.add_ai_message("你好！有什么我可以帮助你的吗？")
history.add_user_message("你能介绍一下自己吗？")
print(type(history))

res = chat.invoke(history.messages)
print(res.content)

2.3、ConversationBufferMemory

针对层次1（保留一个聊天消息列表）-->是一个基础的对话记忆组件，专门按原始顺序存储完整的暴漏出来对话历史，而上面的ChatMessageHistory其实是没有对话历史过程的直接把最终的问题显示出来了

适用场景：对话轮次较少，依赖完整上下文的场景（如：简单的聊天机器）

特点：

完整存储对话历史
简单、无裁剪、无压缩
与Chain/Models无缝集成

支持两种返回格式（通过 return_message参数控制输出格式）

return_messages=True 返回消息对象列表（ListBaseMessage）
teturn_messages=False（默认）返回拼接的纯文本字符串

# 举例1：已字符串的方式返回存储的信息
from langchain_openai import ChatOpenAI
import os
import dotenv
from langchain.memory import ConversationBufferMemory

# 创建大模型对象
dotenv.load_dotenv()
base_url = os.getenv("OPENAI_BASE_URLS")
api_key = os.getenv("OPENAI_API_KEY1")

chat = ChatOpenAI(
    model="gpt-4o-mini",
    base_url=base_url,
    api_key=api_key,
)

#1、实例化ConversationBufferMemory对象
memory = ConversationBufferMemory()
# 2、保存消息到内存中
memory.save_context({"input": "你好！"}, {"output": "你好！有什么我可以帮助你的吗？"})
memory.save_context({"input": "你能介绍一下自己吗？"}, {"output": "我是一个由OpenAI开发的语言模型，旨在帮助用户解答问题、提供信息和进行对话。"})
#3、获取内存中存储的消息
print(memory.load_memory_variables({}))


'''
执行结果：{'history': 'Human: 你好！\nAI: 你好！有什么我可以帮助你的吗？\nHuman: 你能介绍一下自己吗？\nAI: 我是一个由OpenAI开发的语言模型，旨在帮助用户解答问题、提供信息和进行对话。'}
'''

注意：

不管是inputs、outputs的key用什么名字，都认为inputs的key是human，outputs的key是AI
打印的结果的json数据的key，默认是“history”。可以通过ConversationBufferMemory的memory_key属性修改（如：举例2）

# 举例2：已消息列表的方式返存储的信息
from langchain_openai import ChatOpenAI
import os
import dotenv
from langchain.memory import ConversationBufferMemory

# 创建大模型对象
dotenv.load_dotenv()
base_url = os.getenv("OPENAI_BASE_URLS")
api_key = os.getenv("OPENAI_API_KEY1")

chat = ChatOpenAI(
    model="gpt-4o-mini",
    base_url=base_url,
    api_key=api_key,
)

#1、实例化ConversationBufferMemory对象
memory = ConversationBufferMemory(return_messages=True)
# 2、存储相关的信息
memory.save_context({"input": "你好！"}, {"output": "你好！有什么我可以帮助你的吗？"})
memory.save_context({"input": "你能介绍一下自己吗？"}, {"output": "我是一个由OpenAI开发的语言模型，旨在帮助用户解答问题、提供信息和进行对话。"})
#3、获取存储的消息(返回消息)
print(memory.load_memory_variables({}))
print("\n")
# 4、读取内存中的消息（访问原始消息列表）
print(memory.chat_memory.messages)

'''
执行结果：
{'history': [HumanMessage(content='你好！', additional_kwargs={}, response_metadata={}), AIMessage(content='你好！有什么我可以帮助你的吗？', additional_kwargs={}, response_metadata={}), HumanMessage(content='你能介绍一下自己吗？', additional_kwargs={}, response_metadata={}), AIMessage(content='我是一个由OpenAI开发的语言模型，旨在帮助用户解答问题、提供信息和进行对话。', additional_kwargs={}, response_metadata={})]}

[HumanMessage(content='你好！', additional_kwargs={}, response_metadata={}), AIMessage(content='你好！有什么我可以帮助你的吗？', additional_kwargs={}, response_metadata={}), HumanMessage(content='你能介绍一下自己吗？', additional_kwargs={}, response_metadata={}), AIMessage(content='我是一个由OpenAI开发的语言模型，旨在帮助用户解答问题、提供信息和进行对话。', additional_kwargs={}, response_metadata={})]
'''

举例3：结合大模型、提示词模版、Memory使用

# 举例3：结合大模型、提示词、memory

from langchain_openai import ChatOpenAI
import os
import dotenv
from langchain.memory import ConversationBufferMemory
from langchain_core.prompts.prompt import PromptTemplate

# 1、创建大模型对象
dotenv.load_dotenv()
base_url = os.getenv("OPENAI_BASE_URLS")
api_key = os.getenv("OPENAI_API_KEY1")

chat = ChatOpenAI(
    model="gpt-4o-mini",
    base_url=base_url,
    api_key=api_key,
)
# 2、创建提示词模版
template = """
    你可以与人类对话。
    当前对话: {history}
    人类问题: {input}
    回复：
"""
prompt=PromptTemplate.from_template(template)
# 3、创建内存对象
memory = ConversationBufferMemory() # ConversationBufferMemory仅仅是一个存储数据的传统类，它没有实现 Runnable 接口，所有后续就无法使用LCEL(它有一个硬性要求：使用 | 串联的每一个节点，都必须属于 Runnable（可运行）对象)

#4、初始化chain
from langchain.chains import ConversationChain
    #ConversationChain 是 LangChain 专门为“带记忆对话”设计的上层封装（即传统的基于 Class 的链）
    #你调用 chain.invoke() 时，它自动从 memory 读取历史记录并填充给 prompt。
    #然后它调用 chat 模型。
    # 最后拿到回复后，它会自动调用 memory.save_context() 帮你把这轮对话存起来。
chain = ConversationChain(llm=chat, memory=memory, prompt=prompt)

# 4.1 人类提问1
response = chain.invoke({"input":"你好，我的名字叫小明"})
print(response)
# 4.2 人类提问2
res = chain.invoke({"input":"你能介绍一下自己吗？"})
print(res)


'''
{'input': '你好，我的名字叫小明', 'history': '', 'response': '你好，小明！很高兴认识你。有什么我可以帮助你的吗？'}
{'input': '你能介绍一下自己吗？', 'history': 'Human: 你好，我的名字叫小明\nAI: 你好，小明！很高兴认识你。有什么我可以帮助你的吗？', 'response': 'AI: 当然可以！我是一个人工智能助手，旨在回答问题、提供信息和帮助解决各种问题。我可以与您聊天，分享知识，或帮助您找到所需的信息。请问您对什么感兴趣？'}
'''

注意点：

1、提示词中的 “当前对话: {history}” 刚好是ConversationBufferMemory执行后的history

2、第一次提问时返回的history是空的，说明提问前是没有存储记忆的，因为没有聊天记录；第二次提问时返回的history刚好是第一次提问的记录

此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。原文来自 — 版权归原作者所有。

推荐订阅源

博客园 - 尘封~~

一、Chains

1、Chains的基本使用

1.1、chains的基本概念

1.2、LCEL及其基本构成

1.3、Runnable

2、基于LCEL构建的Chains的类型

2.1、create_sql_query_chain

2.2、create_stuff_documents_chain

二、Memory

1、member的设计理念

2、基础member模块的使用

2.1、Member模块的设计思路

2.2、基础：ChatMessageHistory

2.3、ConversationBufferMemory