Langchain学习笔记（二）- SequentialChain & Memory-New Boy

SequentialChain 简介

SequentialChain 是 LangChain 中用于按顺序执行多个 Chains 的工具，适用于需要分步处理输入的场景。

SequentialChain 的特点

顺序执行：SequentialChain 会按照 Chains 列表中的顺序依次执行每个 Chain。
输入输出传递：前一个 Chain 的输出可以作为后一个 Chain 的输入，实现数据在 Chains 之间的传递。
灵活性：可以组合任意数量的 Chains，适用于复杂的多步任务。

实际案例

我们这里设计一个简单连续问答的场景，这里第二个问题用到了第一个问题的答案：

Q1：请问地球到月球的平均距离大概是多少？
Q2: 前面的问题的答案，相当于饶地球赤道几圈？

案例代码

"""
lesson_2.py 使用 SequentialChain 按顺序执行 Chains

本示例演示如何使用 LangChain 的 RunnablePassthrough 和 Chain 来实现顺序执行多个任务。
具体场景：先获取地球到月球的距离，再计算这个距离相当于绕地球赤道几圈。
"""

import getpass
import os
import dotenv
from util.llm_factory import LLMFactory
from langchain.chains import LLMChain, SequentialChain
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

# 加载环境变量
dotenv.load_dotenv()

# 初始化 LLM 模型
llm = LLMFactory.create_llm(llm_type="deepseek")

# 创建第一个提示模板 - 用于获取地球到月球的距离
distance_prompt = ChatPromptTemplate.from_messages(
    [
        # 系统角色提示，定义模型的身份和任务
        ("system", "你是一个专业的天文学家，请准确回答用户关于天体距离的问题。"),
        # 用户输入的问题
        ("human", "{input}"),
    ]
)

# 创建第二个提示模板 - 用于计算距离相当于绕地球赤道几圈
earth_circles_prompt = ChatPromptTemplate.from_messages(
    [
        # 系统角色提示，定义模型的身份和任务
        ("system", "你是一个专业的天文学家，请准确回答用户关于天体距离的问题。"),
        # 用户输入的问题，包含前一个问题的答案作为输入
        (
            "human",
            "地球到月球的距离是：{distance}。这个距离相当于饶地球赤道几圈？",
        ),
    ]
)

# 创建第一个 Chain - 用于获取地球到月球的距离
distance_chain = distance_prompt | llm

# 创建第二个 Chain - 用于计算绕地球赤道的圈数
earth_circles_chain = earth_circles_prompt | llm

# 使用 RunnablePassthrough 创建顺序执行链
# 1. 首先执行 distance_chain 获取距离
# 2. 然后将结果传递给 earth_circles_chain 计算圈数
seq_chain = RunnablePassthrough.assign(distance=distance_chain).assign(
    earth_circles=earth_circles_chain
)

# 调用顺序执行链，传入初始问题
result = seq_chain.invoke({"input": "地球到月球的距离是多少公里？"})

# 打印结果
print("Answer 1 - 地球到月球的距离:", result["distance"].content)
print("Answer 2 - 相当于绕地球赤道的圈数:", result["earth_circles"].content)

整个代码逻辑相对简单，主要流程如下：
定义prompt模板 =》定义chain任务节点 =》将任务节点通过RunnablePassthrough.assign串联起来 =》调用invoke驱动chain

执行结果：

python code/part-1/lesson_2py                                                                                                          
Answer 1 - 地球到月球的距离: 地球到月球的平均距离约为384,400公里。这个距离是地球和月球之间的平均距离，由于月球绕地球的轨道是椭圆形的，实际距离会在这个平均值上下有所波动。
Answer 2 - 相当于绕地球赤道的圈数: 地球的赤道周长约为40,075公里。地球到月球的平均距离约为384,400公里。因此，地球到月球的距离相当于绕地球赤道的圈数可以通过以下计算得出：

\[ \text{圈数} = \frac{\text{地球到月球的距离}}{\text{地球赤道周长}} = \frac{384,400}{40,075} \approx 9.6 \]

所以，地球到月球的距离大约相当于绕地球赤道9.

3. Memory，让Langchain会话具有记忆能力

这个案例，来演示如下功能：

使用Langchain的Memory机制，让会话具有记忆能力
加入session机制，不同session之间的信息是互相隔离的

"""
lesson_3.py 使用 Memory 和 Chain进行对话，并使用 Memory 记录对话历史
"""

import getpass
import os
import dotenv
import sys
from langchain_community.chat_message_histories import ChatMessageHistory

from util.llm_factory import LLMFactory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory

dotenv.load_dotenv()

# 初始化模型
llm = LLMFactory.create_llm(llm_type="deepseek")

# 创建提示模板
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are a helpful assistant. Answer the user's questions based on the conversation history. If the user asks for more details, provide a detailed explanation based on the previous context.",
        ),
        # 将每轮对话历史存储在 history 变量中
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),
    ]
)


# 创建runnable chain
chain = prompt | llm
# 初始化MessageHistory
message_history = ChatMessageHistory()

# 创建session_histories，存储不同会话的message_history
session_histories = {}


def get_message_history(session_id: str) -> ChatMessageHistory:
    if session_id not in session_histories:
        session_histories[session_id] = ChatMessageHistory()
    return session_histories[session_id]


runnable_history = RunnableWithMessageHistory(
    chain,
    get_message_history,
    input_messages_key="input",
    history_messages_key="history",
)


config_1 = {"configurable": {"session_id": "test_session_1"}}
config_2 = {"configurable": {"session_id": "test_session_2"}}

print("开始测试")
Q1 = "你好，我告诉你一个秘密，小明喜欢小李"
print("Q1:", Q1)
response1 = runnable_history.invoke({"input": Q1}, config=config_1)
print("Response 1:", response1.content)

Q2 = "我告诉你的秘密是什么？"
print("Q2:", Q2)
response2 = runnable_history.invoke({"input": Q2}, config=config_2)
print("Response 2:", response2.content)

Q3 = "我告诉你的秘密是什么？"
print("Q3:", Q3)
response3 = runnable_history.invoke({"input": Q3}, config=config_1)
print("Response 3:", response3.content)

执行结果：

开始测试
Q1: 你好，我告诉你一个秘密，小明喜欢小李
Response 1: 你好！谢谢你的分享。看来小明对小李有好感呢！如果你有更多关于他们之间的事情想分享，或者需要建议，随时告诉我哦！😊
Q2: 我告诉你的秘密是什么？
Response 2: 很抱歉，我无法访问之前的对话记录或记住用户告诉我的秘密。我的设计是为了保护用户隐私，不会存储或记忆任何个人信息或对话内容。如果你有任何问题或需要帮助，请随时告诉我！
Q3: 我告诉你的秘密是什么？
Response 3: 你告诉我的秘密是：**小明喜欢小李**。这是一个关于感情的小秘密哦！如果你有更多想分享的，或者需要讨论的，我随时在这里！😊

从测试结果可以看出，Q1和Q2是2个不同的session，所以Q2是不知道答案的，Q3和Q1属于同一个session，所以Q3可以得到答案。

学习笔记

模块化Chain设计 & 链式执行控制

通过拆分"距离查询"和"周长换算"两个独立Chain，体现任务解耦思想。使用ChatPromptTemplate每个Chain包含专属提示模板，采用角色定义（system）和参数插值（{input}）实现标准化输入输出，提高组件复用性。
使用RunnablePassthrough.assign()构建顺序工作流：
- 首链处理原始输入，生成distance结果
- 次链自动接收distance作为输入参数
- 最终结果集聚合两个Chain的输出
数据流管理
- 采用字典结构传递上下文，通过键名匹配保证参数传递准确性。输出结果访问.content属性提取LLM响应内容，兼容不同消息类型的标准化输出。

记忆组件

使用ChatMessageHistory类记录对话历史，以键值对形式存储对话轮次
通过session_histories字典实现多会话隔离，每个session_id对应独立的历史记录
MessagesPlaceholder作为模板变量动态插入历史对话内容

链式结构

构建prompt | llm的简单链式结构（同上面的例子）
RunnableWithMessageHistory包装器将记忆系统与执行链绑定
通过input_messages_key和history_messages_key指定输入/历史变量名

对话上下文管理

使用get_message_history工厂函数按session_id获取对应历史记录
配置字典config_1/config_2演示不同会话的上下文隔离
三次调用验证：config_1保持历史记忆，config_2新建独立上下文

流程图

langchain

实现功能

通过session_id机制轻松实现多用户对话系统
lang历史对话的自动注入使LLM具备上下文理解能力
消息存储格式标准化便于后续分析或持久化存储

目录CONTENT

Langchain学习笔记（二）- SequentialChain & Memory