Langchain组件（续）

本文基于黑马2026的课程AI大模型RAG与智能体开发编写，课程内容地址：黑马程序员大模型RAG与Agent智能体项目实战教程，基于主流的LangChain技术从大模型提示词到实战项目_哔哩哔哩_bilibili

Langchain组件

【扩展】运算符的重载

前文代码中： chain = chat_prompt_template | model

在语法上使用了 | 运算符的重载

在 Python 中，运算符（如 +、|）的行为由类的魔法方法决定。例如：

  • a + b 本质调用的是 a.__add__(b) 双下划线
  • a | b 本质调用的是 a.__or__(b)

只需要自行实现类的方法，即可对|符号的功能进行重写。

示例：

  • 让 a|b|c 的代码得到一个自定义的类对象(类似列表即[a, b, c])
  • 调用run方法依次输出a、b、c
  • 我们需要重写 | 即__or__方法

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87



class Test(object):
    """测试类，代表一个带名字的对象"""

    def __init__(self, name):
        """初始化方法，设置对象的名称属性"""
        self.name = name

    def __or__(self, other):
        """
        重载或运算符(|)

        当执行 a | b 时，Python会调用这个方法
        参数：
            self: 左边的操作数 (a)
            other: 右边的操作数 (b)
        返回：
            一个新的 MySequence 对象，包含 self 和 other
        """
        return MySequence(self, other)

    def __str__(self):
        """
        重载字符串表示方法
        当使用 print() 或 str() 时会调用此方法
        """
        return self.name


class MySequence(object):
    """序列类，用于存储多个对象的链式结构"""

    def __init__(self, *args):
        """
        初始化方法

        参数：
            *args: 可变参数，可以接收任意数量的参数
                   例如 MySequence(a, b, c) 中 args = (a, b, c)
        """
        self.sequence = []  # 创建一个空列表用于存储序列元素
        for arg in args:    # 遍历所有传入的参数
            self.sequence.append(arg)  # 将每个参数添加到序列中

    def __or__(self, other):
        """
        重载或运算符(|)

        当 MySequence 对象与其他对象进行 | 运算时调用
        参数：
            self: 已有的 MySequence 对象
            other: 要添加到序列的新对象
        返回：
            self 自身，支持链式调用
        """
        self.sequence.append(other)  # 将新元素添加到序列末尾
        return self  # 返回自身，实现链式调用

    def run(self):
        """运行方法，打印序列中的所有元素"""
        for i in self.sequence:  # 遍历序列中的每个元素
            print(i)             # 打印元素（会调用元素的 __str__ 方法）


# 主程序入口
if __name__ == '__main__':
    # 创建多个 Test 对象
    a = Test('a')
    b = Test('b')
    c = Test('c')
    e = Test('e')
    f = Test('f')
    g = Test('g')

    # 使用 | 运算符进行链式调用
    # 执行过程：
    # 1. a | b  →  a.__or__(b)  →  MySequence(a, b)
    # 2. MySequence(a, b) | c  →  MySequence.__or__(c)  →  MySequence(a, b, c)
    # 3. 继续添加 e, f, g，最终得到 MySequence(a, b, c, e, f, g)
    d = a | b | c | e | f | g

    # 运行序列，打印所有元素
    d.run()

    # 打印 d 的类型，可以看到它是 MySequence 类
    print(type(d))

Runnable接口 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

 # 通过多次按住ctrl+鼠标左键点击PromptTemplate可以进入到源码中进行查看
from langchain_core.prompts import PromptTemplate
from langchain_community.llms.tongyi import Tongyi


prompt = PromptTemplate.from_template("你是一个AI助手")
model = Tongyi(model="qwen3-max")

chain = prompt | model | prompt | model
# chain.invoke()
# chain.stream()
print(type(chain))

通过结果我们可以看到不论 | 后面怎么追加，得到的类型都是RunnableSequence，而RunnableSequence又是Runnable接口的一个子类，所以chain不管怎么追加，invoke方法和stream方法都是有的，通过这个形式，在Langchain里面只要想添加链，在 | 后面添加就可以了

StrOutputParser解析器

字符串输出解析器

有如下代码，想要以第一次模型的输出结果，第二次去询问模型：

  • 链的构建完全符合要求（参与的组件）
  • 但是运行报错（ValueError: Invalid input type <class ’langchain_core.messages.ai.AIMessage’>. Must be a PromptValue, str, or list of BaseMessages.）
1

 chain = prompt | model | model

错误的主要原因是：

  • prompt的结果是PromptValue类型，输入给了model

  • model的输出结果是：

    然后将AIMessage输入给下一个model，但是这个输入类型是不符合要求的，所以导致报错

模型（ChatTongyi）源码中关于invoke方法明确指定了input的类型：

StrOutputParser是LangChain内置的简单字符串解析器

  • 可以将AIMessage解析为简单的字符串，符合了模型invoke方法要求（可传入字符串，不接收AIMessage类型）
  • 是Runnable接口的子类（可以加入链）
1
2

parser = StrOutputParser()
chain = prompt | model | parser | model

代码示例

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

from langchain_core.messages import AIMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_community.chat_models.tongyi import ChatTongyi

parser = StrOutputParser()
model = ChatTongyi(model="qwen3-max")
prompt = PromptTemplate.from_template(
    "我邻居姓：{lastname}，刚生了{gender}，请起名，仅告知我名字无需其它内容。"
)

chain = prompt | model | parser | model | parser

# res: AIMessage = chain.invoke({"lastname": "张", "gender": "女儿"})
# print(res:content) # 因为输出结果是AIMessage，所以拿到模型的回复内容需要用到:content

# 如果不想用上述形式，解决方法：在第二个model后面再加一个parse![OP](assets/SOP2.png)
res: str = chain.invoke({"lastname": "张", "gender": "女儿"})
print(res)
print(type(res))

此处返回的类型本应显示为<class ‘str’>，但是返回了TextAccesser类型，显示TextAccesser是 LangChain v1.0+ 的正常行为：

它本质上就是字符串 - 继承自 str，所有字符串操作都支持

向后兼容设计 - 为了让旧代码（使用 .text() 方法）和新代码（使用 .text 属性）都能工作

可以把它当作普通字符串使用

JsonOutputParser&多模型执行链 

1

chain = prompt | model | parser | model | parser

在前面我们完成了这样的需求去构建多模型链，不过这种做法并不标准，因为：

上一个模型的输出，没有被处理就输入下一个模型。

正常情况下我们应该有如下处理逻辑：

invoke｜stream 初始输入 → 提示词模板 → 模型 → 数据处理 → 提示词模板 → 模型 → 解析器 → 结果

即：

  • 上一个模型的输出结果，应该作为提示词模版的输入，构建下一个提示词，用来二次调用模型。

根据输出和输入的要求：

invoke｜stream 初始输入 → 提示词模板 → 模型 → 数据处理 → 提示词模板 → 模型 → 解析器 → 结果

代码示例

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import PromptTemplate

# 创建所需的解析器
str_parser = StrOutputParser()
json_parser = JsonOutputParser()

# 模型创建
model = ChatTongyi(model="qwen3-max")

# 第一个提示词模板
first_prompt = PromptTemplate.from_template(
    "我邻居姓：{lastname}，刚生了{gender}，请帮忙起名字，"
    # 如果没有下面这行模板，输出结果会报错
    "并封装为JSON格式返回给我。要求key是name，value就是你起的名字，请严格遵守格式要求。"
)

# 第二个提示词模板
second_prompt = PromptTemplate.from_template(
    "姓名：{name}，请帮我解析含义。"
)

# 构建链   （AIMessage("{name: 张若曦}")
chain = first_prompt | model | json_parser | second_prompt | model | str_parser

for chunk in chain.stream({"lastname": "张", "gender": "女儿"}):
    print(chunk, end="", flush=True)

RunnableLambda&函数加入链

chain = first_prompt | model | json_parser | second_prompt | model | str_parser

前文我们根据JsonOutputParser完成了多模型执行链条的构建。

  • 除了JsonOutputParser这类固定功能的解析器之外
  • 我们也可以自己编写Lambda匿名函数来完成自定义逻辑的数据转换，想怎么转换就怎么转换，更自由。

想要完成这个功能，可以基于RunnableLambda类实现。

RunnableLambda类是LangChain内置的，将普通函数等转换为Runnable接口实例，方便自定义函数加入chain。

语法：

RunnableLambda(函数对象或lambda匿名函数**)**

函数直接如链

chain = first_prompt | model | (lambda ai_msg: {“name”: ai_msg.content}) | second_prompt | model | str_parser

跳过RunnableLambda类，直接让函数加入链也是可以的。

因为Runnable接口类在实现__or__的时候，支持Callable接口的实例。

  • 函数就是Callable接口的实例

如上代码示例，|符号（底层是调用__or__）组链，是支持函数加入的。其本质是将函数自动转换为RunnableLambda

代码示例

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_community.chat_models.tongyi import ChatTongyi

model = ChatTongyi(model="qwen3-max")
str_parser = StrOutputParser()

first_prompt = PromptTemplate.from_template(
    "我邻居姓：{lastname}，刚生了{gender}，请帮忙起名字，仅生成一个名字，并告知我名字，不要额外信息。"
)

second_prompt = PromptTemplate.from_template(
    "姓名{name}，请帮我解析含义。"
)

# 函数的入参：AIMessage -> dict  ({"name": "xxx"})
# my_func = RunnableLambda(lambda ai_msg: {"name": ai_msg.content})

chain = first_prompt | model | (lambda ai_msg: {"name": ai_msg.content}) | second_prompt | model | str_parser

for chunk in chain.stream({"lastname": "曹", "gender": "女孩"}):
    print(chunk, end="", flush=True)

Memory临时会话记忆

如果想要封装历史记录，除了自行维护历史消息外，也可以借助LangChain内置的历史记录附加功能。

LangChain提供了History功能，帮助模型在有历史记忆的情况下回答。

  • 基于RunnableWithMessageHistory在原有链的基础上创建带有历史记录功能的新链（新Runnable实例）
  • 基于InMemoryChatMessageHistory为历史记录提供内存存储（临时用）

完整代码

代码示例

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import PromptTemplate, ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory

model = ChatTongyi(model="qwen3-max")
# prompt = PromptTemplate.from_template(
#     "你需要根据会话历史回应用户问题。对话历史：{chat_history}，用户提问：{input}，请回答"
# )
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你需要根据会话历史回应用户问题。对话历史："),
        MessagesPlaceholder("chat_history"),
        ("human", "请回答如下问题：{input}")
    ]
)

str_parser = StrOutputParser()

# 创建一个打印prompt的函数
def print_prompt(full_prompt):
    print("【", full_prompt.to_string(), "】")
    return full_prompt


base_chain = prompt | print_prompt | model | str_parser


store = {}      # 创建一个空字典：key就是session，value就是InMemoryChatMessageHistory类对象
# 实现通过会话id获取InMemoryChatMessageHistory类对象
def get_history(session_id):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()

    return store[session_id]

# 创建一个新的链，对原有链增强功能：自动附加历史消息
conversation_chain = RunnableWithMessageHistory(
    base_chain,     # 被增强的原有chain
    get_history,    # 通过会话id获取InMemoryChatMessageHistory类对象
    input_messages_key="input",             # 表示用户输入在模板中的占位符
    history_messages_key="chat_history"     # 表示历史模板中的占位符
)

# 业务
if __name__ == '__main__':
    # 固定格式，添加LangChain的配置，为当前程序配置所属的session_id
    session_config = {
        "configurable": {
            "session_id": "user_001"
        }
    }

   
    res = conversation_chain.invoke({"input": "小明有2个猫"}, session_config)
    print("第1次执行：", res)
    
    res = conversation_chain.invoke({"input": "小刚有1只狗"}, session_config)
    print("第2次执行：", res)

    res = conversation_chain.invoke({"input": "总共有几个宠物"}, session_config)
    print("第3次执行：", res)

当前会话是临时的，如果将第1次和第2次执行注释掉，直接执行第3次则会提示如下信息：

Memory长期会话记忆

使用InMemoryChatMessageHistory仅可以在内存中临时存储会话记忆，一旦程序退出，则记忆丢失。

InMemoryChatMessageHistory 类继承自 BaseChatMessageHistory

在官方注释中给出了相关实现的指南，并给出了基于文件的历史消息存储示例代码。

我们可以自行实现一个基于Json格式和本地文件的会话数据保存。

FileChatMessageHistory类实现，核心思路：

  • 基于文件存储会话记录，以session_id为文件名，不同session_id有不同文件存储消息

继承BaseChatMessageHistory实现如下3个方法：

  • add_messages:同步模式，添加消息
  • messages:同步模式，获取消息
  • clear：同步模式，清除消息

如下方代码，官方在BaseChatMessageHistory类的注释中提供了一个基于文件存储的示例代码。

其余核心代码

代码示例

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

import os, json
from typing import Sequence

from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import message_to_dict, messages_from_dict, BaseMessage
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 import RunnableWithMessageHistory


# message_to_dict：单个消息对象（BaseMessage类实例） -> 字典
# messages_from_dict：[字典、字典...]  -> [消息、消息...]
# AIMessage、HumanMessage、SystemMessage 都是BaseMessage的子类


class FileChatMessageHistory(BaseChatMessageHistory):
    def __init__(self, session_id, storage_path):
        self.session_id = session_id        # 会话id
        self.storage_path = storage_path    # 不同会话id的存储文件，所在的文件夹路径
        # 完整的文件路径
        self.file_path = os.path.join(self.storage_path, self.session_id)

        # 确保文件夹是存在的
        os.makedirs(os.path.dirname(self.file_path), exist_ok=True)

    def add_messages(self, messages: Sequence[BaseMessage]) -> None:
        # Sequence序列 类似list、tuple
        all_messages = list(self.messages)      # 已有的消息列表
        all_messages.extend(messages)           # 新的和已有的融合成一个list

        # 将数据同步写入到本地文件中
        # 类对象写入文件 -> 一堆二进制
        # 为了方便，可以将BaseMessage消息转为字典（借助json模块以json字符串写入文件）
        # 官方message_to_dict：单个消息对象（BaseMessage类实例） -> 字典
        # new_messages = []
        # for message in all_messages:
        #     d = message_to_dict(message)
        #     new_messages.append(d)

        new_messages = [message_to_dict(message) for message in all_messages]
        # 将数据写入文件
        with open(self.file_path, "w", encoding="utf-8") as f:
            json.dump(new_messages, f)

    @property       # @property装饰器将messages方法变成成员属性用
    def messages(self) -> list[BaseMessage]:
        # 当前文件内： list[字典]
        try:
            with open(self.file_path, "r", encoding="utf-8") as f:
                messages_data = json.load(f)    # 返回值就是：list[字典]
                return messages_from_dict(messages_data)
        except FileNotFoundError:
            return []

    def clear(self) -> None:
        with open(self.file_path, "w", encoding="utf-8") as f:
            json.dump([], f)





model = ChatTongyi(model="qwen3-max")
# prompt = PromptTemplate.from_template(
#     "你需要根据会话历史回应用户问题。对话历史：{chat_history}，用户提问：{input}，请回答"
# )
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你需要根据会话历史回应用户问题。对话历史："),
        MessagesPlaceholder("chat_history"),
        ("human", "请回答如下问题：{input}")
    ]
)

str_parser = StrOutputParser()


def print_prompt(full_prompt):
    print("="*20, full_prompt.to_string(), "="*20)
    return full_prompt


base_chain = prompt | print_prompt | model | str_parser

def get_history(session_id):
    return FileChatMessageHistory(session_id, "./chat_history")

# 创建一个新的链，对原有链增强功能：自动附加历史消息
conversation_chain = RunnableWithMessageHistory(
    base_chain,     # 被增强的原有chain
    get_history,    # 通过会话id获取InMemoryChatMessageHistory类对象
    input_messages_key="input",             # 表示用户输入在模板中的占位符
    history_messages_key="chat_history"     # 表示用户输入在模板中的占位符
)


if __name__ == '__main__':
    # 固定格式，添加LangChain的配置，为当前程序配置所属的session_id
    session_config = {
        "configurable": {
            "session_id": "user_001"
        }
    }

    res = conversation_chain.invoke({"input": "小明有2个猫"}, session_config)
    print("第1次执行：", res)
    
    res = conversation_chain.invoke({"input": "小刚有1只狗"}, session_config)
    print("第2次执行：", res)

    # res = conversation_chain.invoke({"input": "总共有几个宠物"}, session_config)
    # print("第3次执行：", res)

先执行第一次和第二次，然后将第一次和第二次注销掉，再执行第三次，可以发现还是能够正常执行

Document loaders：文档加载器

文档加载器提供了一套标准接口，用于将不同来源（如 CSV、PDF 或 JSON等）的数据读取为 LangChain 的文档格式。这确保了无论数据来源如何，都能对其进行一致性处理。

文档加载器（内置或自行实现）需实现BaseLoader接口。

Class Document，是LangChain内文档的统一载体，所有文档加载器最终返回此类的实例。一个基础的Document类实例，基于如下代码创建：

可以看到，Document类其核心记录了：

  • page_content：文档内容
  • metadata：文档元数据（字典）

不同的文档加载器可能定义了不同的参数，但是其都实现了统一的接口（方法）。

  • load()：一次性加载全部文档
  • lazy_load()：延迟流式传输文档，对大型数据集很有用，避免内存溢出。

一个简单的CSVLoader的使用示例如下：

LangChain内置了许多文档加载器，详细参见官方文档：https://docs.langchain.com/oss/python/integrations/document_loaders

我们简单的学习如下几个常用的文档加载器：

  • CSVLoader
  • JSONLoader
  • PDFLoader

CSVLoader

自定义CSV文件的解析和加载

代码示例

data.csv

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

name,age,gender,hobby
王梓涵,25,男,"吃饭,rap"
刘若曦,22,女,"睡觉,rap"
陈俊宇,20,男,"吃饭,rap"
赵思瑶,28,女,"睡觉,rap"
黄浩然,15,男,"吃饭,rap"
林雨桐,20,女,"唱跳,rap"
周博文,20,男,"吃饭,rap"
吴诗琪,24,女,"吃饭,rap"
马子轩,22,男,"睡觉,rap"
孙悦然,27,女,"吃饭,rap"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

from langchain_community.document_loaders import CSVLoader


loader = CSVLoader(
    file_path="./data/stu.csv",
    csv_args={
        "delimiter": ",",       # 指定分隔符（需要与csv文件中的分隔符一致，否则会导致整个文件解析失败，或者所有数据都被错误地读成一行（或一列））
        "quotechar": '"',       # 指定带有分隔符文本的引号包围是单引号还是双引号
        # 如果数据原本有表头，就不要下面的代码，如果没有可以使用
        # "fieldnames": ['name', 'age', 'gender', '爱好']
    },
    encoding="utf-8"            # 指定编码为UTF-8
)

# 批量加载 .load()   ->  [Document, Document, ...]
# documents = loader.load()
#
# for document in documents:
#     print(type(document), document)

# 懒加载  .lazy_load()  迭代器[Document]
for document in loader.lazy_load():
    print(document)

以下为懒加载的结果，如果内存较大可以选择批量加载，否则选择懒加载即可

JSONLoader

JSONLoader用于将JSON数据加载为Document类型对象。

使用JSONLoader需要额外安装： pip install jq

jq是一个跨平台的json解析工具，LangChain底层对JSON的解析就是基于jq工具实现的。

将JSON数据的信息抽取出来，封装为Document对象，抽取的时候依赖jq_schema语法。

了解jq的基本抽取规则后，即可使用JSONLoader加载JSON文件了。

代码示例

stu.json

1
2
3
4
5
6
7
8
9

{
    "name": "周杰轮",
    "age": 11,
    "hobby": ["唱", "跳", "RAP"],
    "other": {
        "addr": "深圳",
        "tel": "12332112321"
    }
}

stu_json_lines.json

1
2
3

{"name": "周杰轮", "age": 11, "gender": "男"}
{"name": "蔡依临", "age": 12, "gender": "女"}
{"name": "王力鸿", "age": 11, "gender": "男"}

stus.json

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

name,age,gender,hobby
王梓涵,25,男,"吃饭,rap"
刘若曦,22,女,"睡觉,rap"
陈俊宇,20,男,"吃饭,rap"
赵思瑶,28,女,"睡觉,rap"
黄浩然,15,男,"吃饭,rap"
林雨桐,20,女,"唱跳,rap"
周博文,20,男,"吃饭,rap"
吴诗琪,24,女,"吃饭,rap"
马子轩,22,男,"睡觉,rap"
孙悦然,27,女,"吃饭,rap"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

from langchain_community.document_loaders import JSONLoader

loader = JSONLoader(
    file_path="./data/stu.json",
    # 单独抽取
    # jq_schema=".name",
    # jq_schema=".other.addr",	#抽取深圳
    
    # 整体抽取
    jq_schema=".",
    # text_content=False,     # 告知JSONLoader 我抽取的内容不是字符串
)

# 由于文件比较小，此处使用全加载
document = loader.load()
print(document)


 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

from langchain_community.document_loaders import JSONLoader

loader = JSONLoader(
    file_path="./data/stus.json",
    # 告诉JSONLoader，我需要抽取的JSON对象是全部的name字段
    jq_schema=".[].name",
    text_content=False,     # 告知JSONLoader 我抽取的内容不是字符串
)

# 由于文件比较小，此处使用全加载
document = loader.load()
print(document)


 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from langchain_community.document_loaders import JSONLoader

loader = JSONLoader(
    file_path="./data/stu_json_lines.json",
    jq_schema=".name",
    text_content=False,     # 告知JSONLoader 我抽取的内容不是字符串
    json_lines=True         # 告知JSONLoader 这是一个JSONLines文件（每一行都是一个独立的标准JSON）
)

document = loader.load()
print(document)

运行结果同上

总结

TextLoader

除了前文学习的三个Loader以外，还有一个基本的加载器：TextLoader

作用：读取文本文件（如.txt），将全部内容放入一个Document对象中。

RecursiveCharacterTextSplitter，递归字符文本分割器，主要用于按自然段落分割大文档。是LangChain官方推荐的默认字符分割器。

它在保持上下文完整性和控制片段大小之间实现了良好平衡，开箱即用效果佳。

代码示例

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

# pip install langchain_text_splitters

loader = TextLoader("./data/Python基础语法.txt", encoding="utf-8")

docs = loader.load()        # [Document]

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,         # 分段的最大字符数
    chunk_overlap=50,       # 分段之间允许重叠字符数
    # 文本自然段落分隔的依据符号
    separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""],
    length_function=len,    # 统计字符的依据函数
)

split_docs = splitter.split_documents(docs)
print(len(split_docs))
for doc in split_docs:
    print("="*20)
    print(doc)
    print("="*20)

PyPDFLoader

LangChain内支持许多PDF的加载器，我们选择其中的PyPDFLoader使用。

PyPDFLoader加载器，依赖PyPDF库，所以，需要安装它：

pip install pypdf

PyPDFLoader使用还是比较简单的，如下代码即可快速加载PDF中的文字内容了：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader(
    file_path="./data/pdf2.pdf",
    mode="single",        # 默认是page模式，每个页面形成一个Document文档对象，
                          # single模式，不管有多少页，只返回1个Document对象
    password="itheima"	  # 用于读取含有密码的pdf文件
)

i = 0
for doc in loader.lazy_load():
    i += 1
    print(doc)
    print("="*20, i)

Vector stories向量存储

基于LangChain的向量存储，存储嵌入数据，并执行相似性搜索。

代码示例

Info.csv

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

source,info
黑马程序员,Python是世界上最好的编程语言
传智教育,股票代码003032
黑马程序员,LangChain极大地方便了大模型开发
黑马程序员,AI和Python是下一个十年的风口
传智教育,Python学起来很简单的
黑马程序员,学习Python键盘敲烂月薪过万
黑马程序员,努力带来成就，Python助力辉煌
黑马程序员,学习Python的时候也要记得好好休息打打篮球
黑马程序员,明天晚上吃啥子呀
黑马程序员,如何快速减肥呢

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.document_loaders import CSVLoader

vector_store = InMemoryVectorStore(
    embedding=DashScopeEmbeddings()
)


loader = CSVLoader(
    file_path="./data/info.csv",
    encoding="utf-8",
    source_column="source",     # 指定本条数据的来源是哪里
)

documents = loader.load()
# id1 id2 id3 id4 ...
# 向量存储的 新增、删除、检索
vector_store.add_documents(
    documents=documents,        # 被添加的文档，类型：list[Document]
    ids=["id"+str(i) for i in range(1, len(documents)+1)] # 给添加的文档提供id（字符串）  list[str]
)

# 删除  传入[id, id...]
vector_store.delete(["id1", "id2"])

# 检索 返回类型list[Document]
result = vector_store.similarity_search(
    "瑞达法",
    3       # 检索的结果要几个
)

print(result)

由于这是内存存储，只要程序停止数据就丢失了，所以下面我们基于外部数据库的存储：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

from langchain_chroma import Chroma
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.document_loaders import CSVLoader

# Chroma 向量数据库（轻量级的）
# 确保 langchain-chroma chromadb 这两个库安装了的，没有的话请pip install

vector_store = Chroma(
    collection_name="test",     # 当前向量存储起个名字，类似数据库的表名称
    embedding_function=DashScopeEmbeddings(),       # 嵌入模型
    persist_directory="./chroma_db"     # 指定数据存放的文件夹
)


loader = CSVLoader(
    file_path="./data/info.csv",
    encoding="utf-8",
    source_column="source",     # 指定本条数据的来源是哪里
)

documents = loader.load()
# id1 id2 id3 id4 ...
# 向量存储的 新增、删除、检索
vector_store.add_documents(
    documents=documents,        # 被添加的文档，类型：list[Document]
    ids=["id"+str(i) for i in range(1, len(documents)+1)] # 给添加的文档提供id（字符串）  list[str]
)

# 删除  传入[id, id...]
vector_store.delete(["id1", "id2"])

# 检索 返回类型list[Document]
result = vector_store.similarity_search(
    "Python是不是简单易学呀",
    3,        # 检索的结果要几个
    filter={"source": "黑马程序员"}  # 过滤的值和values
)

print(result)

检索向量并构建提示词

向量存储的实例，通过**add_texts(list[str])**方法可以快速添加到向量存储中。

流程：

1.先通过向量存储检索匹配信息

2.将用户提问和匹配信息一同封装到提示词模板中提问模型

代码示例

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

"""
提示词：用户的提问 + 向量库中检索到的参考资料
"""
from langchain_community.chat_models import ChatTongyi
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser


def print_prompt(prompt):
    print(prompt.to_string())
    print("=" * 20)
    return prompt


model = ChatTongyi(model="qwen3-max")
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "以我提供的已知参考资料为主，简洁和专业的回答用户问题。参考资料:{context}。"),
        ("user", "用户提问：{input}")
    ]
)

vector_store = InMemoryVectorStore(embedding=DashScopeEmbeddings(model="text-embedding-v4"))

# 准备一下资料（向量库的数据）
# add_texts 传入一个 list[str]
vector_store.add_texts(
    ["减肥就是要少吃多练", "在减脂期间吃东西很重要,清淡少油控制卡路里摄入并运动起来", "跑步是很好的运动哦"])

input_text = "怎么减肥？"

# 检索向量库
result = vector_store.similarity_search(input_text, 2)
reference_text = "["
for doc in result:
    reference_text += doc.page_content
reference_text += "]"

chain = prompt | print_prompt | model | StrOutputParser()

res = chain.invoke({"input": input_text, "context": reference_text})
print(res)

RunnablePassthrough的使用

代码示例

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143

"""
核心思路：构建一个 RAG（检索增强生成）链
流程：用户提问 → 向量库检索相关资料 → 将提问+资料拼成提示词 → 发送给大模型 → 输出回答

RunnablePassthrough 的作用：
  当链的输入是一个简单的字符串（而非 dict）时，
  RunnablePassthrough 负责将原始输入原封不动地传递到下游。
  这样用户调用 chain.invoke("怎么减肥？") 时，
  "怎么减肥？" 既会送给 retriever 做检索，也会作为 "input" 填入提示词模板。
"""
from langchain_community.chat_models import ChatTongyi
from langchain_core.documents import Document
from langchain_core.runnables import RunnablePassthrough
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser


def print_prompt(prompt):
    """调试用：打印最终拼装好的提示词，方便观察模板变量是否被正确替换"""
    print(prompt.to_string())
    print("=" * 20)
    return prompt  # 注意：必须把 prompt 返回回去，否则链就断了


# ======================== 1. 初始化大模型 ========================
model = ChatTongyi(model="qwen3-max")

# ======================== 2. 定义提示词模板 ========================
# 模板中有两个变量：
#   {context} —— 来自向量检索的结果（参考资料）
#   {input}   —— 来自用户的原始提问
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "以我提供的已知参考资料为主，简洁和专业的回答用户问题。参考资料:{context}。"),
        ("user", "用户提问：{input}")
    ]
)

# ======================== 3. 创建内存向量库 ========================
# InMemoryVectorStore：数据只存在内存中，程序结束就没了（适合学习演示）
# DashScopeEmbeddings：调用通义千问的 embedding 模型，把文本转成向量
vector_store = InMemoryVectorStore(embedding=DashScopeEmbeddings(model="text-embedding-v4"))

# 准备一下资料（向量库的数据）
# add_texts 传入一个 list[str]，内部会自动调用 embedding 模型将文本转为向量并存储
vector_store.add_texts(
    ["减肥就是要少吃多练", "在减脂期间吃东西很重要,清淡少油控制卡路里摄入并运动起来", "跑步是很好的运动哦"])

input_text = "怎么减肥？"

# ======================== 4. 创建检索器（Retriever） ========================
# as_retriever() 将向量库包装成一个 Retriever 对象
# search_kwargs={"k": 2} 表示每次检索返回最相似的 2 条文档
# 重要：retriever 实现了 Runnable 接口，所以可以用 | 管道符与其他组件串联
#   - 输入：str（用户的提问）
#   - 输出：list[Document]（检索到的文档列表）
retriever = vector_store.as_retriever(search_kwargs={"k": 2})


def format_func(docs: list[Document]):
    """
    将检索到的 Document 列表转成一个拼接好的字符串，
    方便嵌入到提示词的 {context} 中。
    例如输出："[减肥就是要少吃多练跑步是很好的运动哦]"
    """
    if not docs:
        return "无相关参考资料"

    formatted_str = "["
    for doc in docs:
        formatted_str += doc.page_content
    formatted_str += "]"

    return formatted_str

# ======================== 5. 构建链（Chain） ========================
# 这是最核心的部分，拆解来看：
#
# 第一层（dict 构造）：{"input": ..., "context": ...}
#   这一层负责生成一个 dict，作为 prompt 模板的输入参数：
#
#   "input": RunnablePassthrough()
#       → RunnablePassthrough() 会把链的原始输入（即 input_text）原样透传
#       → 等价于：lambda x: x  （把 "怎么减肥？" 直接传过去）
#
#   "context": retriever | format_func
#       → 先用 retriever 对输入做向量检索，得到 list[Document]
#       → 再用 format_func 将 list[Document] 格式化为字符串
#       → 等价于：lambda x: format_func(retriever.invoke(x))
#
# 第二层：| prompt
#   → 将 {"input": "...", "context": "..."} 填入提示词模板，得到完整 prompt
#
# 第三层：| print_prompt
#   → 调试用，打印出拼装好的提示词（并透传给下一步）
#
# 第四层：| model
#   → 将提示词发送给大模型，得到 AI 的回复
#
# 第五层：| StrOutputParser()
#   → 将大模型返回的 Message 对象提取为纯字符串
#
chain = (
    {"input": RunnablePassthrough(), "context": retriever | format_func} | prompt | print_prompt | model | StrOutputParser()
)

# ======================== 6. 执行链 ========================
# invoke() 触发整条链的执行：
#   "怎么减肥？" → 向量检索 → 格式化 → 填充模板 → 大模型 → 纯文本回答
res = chain.invoke(input_text)
print(res)

"""
完整数据流总结：

chain.invoke("怎么减肥？")
    │
    ▼
{"input": RunnablePassthrough(), "context": retriever | format_func}
    │                          │
    │                          ▼
    │                    retriever("怎么减肥？")
    │                          │
    │                          ▼
    │                    list[Document] （检索到的 2 条文档）
    │                          │
    │                          ▼
    │                    format_func(docs) → "[减肥就是要少吃多练跑步是很好的运动哦]"
    │
    ▼
{"input": "怎么减肥？", "context": "[减肥就是要少吃多练跑步是很好的运动哦]"}
    │
    ▼
prompt（填充模板） → PromptValue（完整的提示词）
    │
    ▼
model（调用大模型） → AI 回复
    │
    ▼
StrOutputParser() → 纯字符串结果
"""

© 2025 - 2026 凉介
本站于2026年3月31日建立
使用 Hugo 构建
主题 Stack 由 Jimmy 设计