此文章是我根据黑马程序员的“大模型RAG与Agent智能体项目实战教程，基于主流的LangChain技术从大模型提示词到实战项目”的视频中所做的关于大模型RAG的个人笔记。如果你也在学习视频中遇到困惑或问题，希望这个文档可以帮你找到灵感。

 

学习视频地址：https://www.bilibili.com/video/BV1yjz5BLEoY/?spm_id_from=333.1391.0.0&p=39

 

 

目录

 

前置准备

大模型Prompt工程指南

LangChain

RAG

LangChain调用大模型

chain链

Memory

文档加载器

向量存储与检索

 

 

 

 

前置准备

需要现在python编译器中下载openai，下载完成后，我们需要到大模型服务平台百炼控制台，去获取我们所创建的api-key，接着选个大模型获取他的API代码示例，我先以qwen3-max的模型为例

from openai import OpenAI
import os

client = OpenAI(
    # 如果没有配置环境变量，请用阿里云百炼API Key替换：api_key="sk-xxx"
    api_key="略",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

messages = [{"role": "user", "content": "你是谁"}]
completion = client.chat.completions.create(
    model="qwen3-max",  # 您可以按需更换为其它深度思考模型
    messages=messages,
    extra_body={"enable_thinking": True},
    stream=True
)
is_answering = False  # 是否进入回复阶段
print("\n" + "=" * 20 + "思考过程" + "=" * 20)
for chunk in completion:
    delta = chunk.choices[0].delta
    if hasattr(delta, "reasoning_content") and delta.reasoning_content is not None:
        if not is_answering:
            print(delta.reasoning_content, end="", flush=True)
    if hasattr(delta, "content") and delta.content:
        if not is_answering:
            print("\n" + "=" * 20 + "完整回复" + "=" * 20)
            is_answering = True
        print(delta.content, end="", flush=True)

将APIKEY暴漏在代码中是不合理的，我们可以将API-KEY保存到电脑的环境变量，或者pycharm的临时变量。

我们来分析这段API示例代码

client = OpenAI(
    # 如果没有配置环境变量，请用阿里云百炼API Key替换：api_key="sk-xxx"
    api_key="略",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

api_key为用你的模型所需要的key

base_url是你要使用的模型的访问域名

 

messages = [{"role": "user", "content": "如果我是特朗普，我的下一步应该做什么才能复兴美国"}]

messages为一个list类型，可以设置多条消息

role有三类：

system角色：设定助手的整体行为、角色和规则，为对话提供上下文框架（如指定助手身份、回答风格、核心要求），是全局的背景设定，影响后续所有交互。

assistant角色：代表AI助手的回答，可以在代码中认为设定。

user角色：代表用户，发送问题，指令或需求。

 

completion = client.chat.completions.create(
    model="qwen3-max",  # 您可以按需更换为其它深度思考模型
    messages=messages,
    extra_body={"enable_thinking": True},
    stream=True
)

model：你所使用模型的名称

messages：你向模型所发送的消息

extra_body：是否开启深度思考

stream：实时流式获取

 

我们在调用模型的时候通过在message的list内，阻止历史消息提供给模型。

因为当前的消息是一次性的，如果是生产系统可以将消息保存到文件，数据库等持久化工具内，需要的时候提取使用。

后续学习LangChain库会学习短期记忆和长期记忆的使用方法

messages = [
    {"role": "system","content":"你是一个AI助手，回答应该特别简洁"},
    {"role": "user", "content": "小明有五个宠物"},
    {"role": "assistant","content":"好的"},
    {"role": "user", "content": "小红有八个宠物"},
    {"role": "assistant","content":"好的"},
    {"role": "user", "content": "总共有几只宠物"}
]

//AI回复
我们被问到：“总共有几只宠物”。首先需要回顾之前的对话历史。用户之前说：“小明有五个宠物”，然后“小红有八个宠物”。现在问“总共有几只宠物”。这里的“总共”应该是指小明和小红的宠物加起来的总数。所以需要将5和8相加。5+8=13。需要注意是否有陷阱，比如是否有重复的宠物，或者是否包括其他人的宠物，但根据对话历史，只有小明和小红被提到，且问题明确是“总共有几只”，结合上下文，应该是求和。另外，用户要求回答特别简洁，所以直接给出数字即可。确认计算：5+8=13，没错。不需要额外解释。所以回答“13”即可。
====================完整回复====================
13

 

 

 

Ollama

为了避免未来阿里云免费额度到期或者不提供免费活动导致无法开发，我们可以基于Ollama部署本地模型，供代码调用。

ollama是一款旨在简化大型语言模型本都部署和运行过程的开源软件。

ollama提供了一个轻量级、易于扩展的框架，让开发者能够在本地机器上轻松构建和管理LLMS(大型语言模型)。通过它，我们可以导入和定制自己的模型，无需关注复杂的底层实现细节。网址：https://ollama.com

简单来说可以认为是阿里云百炼平台的本地版，在自己电脑上部署和运行大模型，由自己的电脑硬件提供算力支撑模型运行。

 

不过对自身电脑消耗很大，因为要占用硬件，所以能尽量用云端模型就用云端模型吧。

 

 

 

大模型Prompt工程指南

提示工程是指在不更新模型权重的情况下如何与大模型交互以到其行为以获得所需结果的方法。

提示词工程就是更好的向模型提问的技巧。

 

Zero-shot是指在训练阶段不存在与测试阶段完全相同的类别，但是模型可以使用训练过的知识来推广到测试集中的新类别上。这种能力能称为零样本学习，因为模型在训练时从未见过测试集中的新类别，在模型训练和提示词优化中均有体现。

它基于已训练的能力，不提供任何示例，仅通过语言描述去描述任务的需求目标与约束，让模型直接生成结果。简单来说就是用语言定义任务，解放模型的预训练知识。

更准确的说法：零样本学习，就是我们给模型一个它从未见过的任务，但它能利用自己“肚子里”已有的海量知识，直接理解并用它的通用能力来完成任务，而不需要再进行任何额外的训练。

 

Few-shot思想是指少样本学习，当模型在学习了一定类别的大量数据后，对于新的类别，只需要少量的样本就能快速学习，极端点的有单样本学习。

用来基于给予模型少量示例，引导模型对其示例输出结果。

少量学习样本的代码示例

from openai import OpenAI

# 1. 获取client对象，OpenAI类对象
client = OpenAI(
    # base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
    api_key= "略",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

examples_data = {  # 示例数据
    '新闻报道': '今日，股市经历了一轮震荡，受到宏观经济数据和全球贸易紧张局势的影响。投资者密切关注美联储可能的政策调整，以适应市场的不确定性。',
    '财务报告': '本公司年度财务报告显示，去年公司实现了稳步增长的盈利，同时资产负债表呈现强劲的状况。经济环境的稳定和管理层的有效战略执行为公司的健康发展奠定了基础。',
    '公司公告': '本公司高兴地宣布成功完成最新一轮并购交易，收购了一家在人工智能领域领先的公司。这一战略举措将有助于扩大我们的业务领域，提高市场竞争力',
    '分析师报告': '最新的行业分析报告指出，科技公司的创新将成为未来增长的主要推动力。云计算、人工智能和数字化转型被认为是引领行业发展的关键因素，投资者应关注这些趋势'
}
# 分类列表
examples_types = ['新闻报道', '财务报道', '公司公告', '分析师报告']

# 提问数据
questions = [
    "今日，央行发布公告宣布降低利率，以刺激经济增长。这一降息举措将影响贷款利率，并在未来几个季度内对金融市场产生影响。",
    "ABC公司今日发布公告称，已成功完成对XYZ公司股权的收购交易。本次交易是ABC公司在扩大业务范围、加强市场竞争力方面的重要举措。据悉，此次收购将进一步巩固ABC公司在行业中的地位，并为未来业务发展提供更广阔的发展空间。详情请见公司官方网站公告栏",
    "公司资产负债表显示，公司偿债能力强劲，现金流充足，为未来投资和扩张提供了坚实的财务基础。",
    "最新的分析报告指出，可再生能源行业预计将在未来几年经历持续增长，投资者应该关注这一领域的投资机会",
    "小明喜欢小新哟"
]

"""
[
    {"role": "system",      "content": "你是金融专家，将文本分类为['新闻报道', '财务报道', '公司公告', '分析师报告']，不清楚的分类为'不清楚类别' 下面有示例："},

    {"role": "user",        "content": "今日，央行发布公告宣布降............."},
    {"role": "assistant",   "content": "新闻报道"},
    {"role": "user",        "content": "ABC公司今日发布公告称，已成功完成对XYZ公司股................."},
    {"role": "assistant",   "content": "财务报告},
    {"role": "user",        "content": "公司资产负债表显示，公司偿债能力强劲，现金流充足..................."},
    {"role": "assistant",   "content": "公司公告"},
    {"role": "user",        "content": "最新的分析报告指出，可再生能源............."},
    {"role": "assistant",   "content": "分析师报告"},

    {"role": "user",        "content": "要提问的问题"}
]
"""

messages = [
    {"role": "system",
     "content": "你是金融专家，将文本分类为['新闻报道', '财务报道', '公司公告', '分析师报告']，不清楚的分类为'不清楚类别' 下面有示例："},
]

#将examples_data的键值一起导入到messages的list中。
for key, value in examples_data.items():
    messages.append({"role": "user", "content": value})
    messages.append({"role": "assistant", "content": key})

# 向模型提问
for q in questions:
    response = client.chat.completions.create(
        # model="qwen3-max",
        model="qwen3:4b",
        messages=messages + [{"role": "user", "content": f"按照示例，回答这段文本的分类类别：{q}"}]
    )

    print(response.choices[0].message.content)

 

Python中的json主要完成：

将python字典、列表转换成json字符串

将json字符串转换为python字典或列表。

主要使用python的内置库

json.dumps（字典或列表,ensure_ascii=False）将字典或列表转换为json字符串

ensure_ascii参数保证中文能正常显示

json.loads(json字符串)：将json字符串转换为python字典或列表。

 

接下来我们用提示词优化来实现对金融文本信息的提取（也是通过少量学习实现）

import json

from openai import OpenAI

# 1. 获取client对象，OpenAI类对象
client = OpenAI(
    # base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
    api_key= "略",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

schema = ['日期', '股票名称', '开盘价', '收盘价', '成交量']
examples_data = [  # 示例数据
    {
        "content": "2023-01-10，股市震荡。股票强大科技A股今日开盘价100人民币，一度飙升至105人民币，随后回落至98人民币，最终以102人民币收盘，成交量达到520000。",
        "answers": {
            "日期": "2023-01-10",
            "股票名称": "强大科技A股",
            "开盘价": "100人民币",
            "收盘价": "102人民币",
            "成交量": "520000"
        }
    },
    {
        "content": "2024-05-16，股市利好。股票英伟达美股今日开盘价105美元，一度飙升至109美元，随后回落至100美元，最终以116美元收盘，成交量达到3560000。",
        "answers": {
            "日期": "2024-05-16",
            "股票名称": "英伟达美股",
            "开盘价": "105美元",
            "收盘价": "116美元",
            "成交量": "3560000"
        }
    }
]
questions = [  # 提问问题
    "2025-06-16，股市利好。股票传智教育A股今日开盘价66人民币，一度飙升至70人民币，随后回落至65人民币，最终以68人民币收盘，成交量达到123000。",
    "2025-06-06，股市利好。股票黑马程序员A股今日开盘价200人民币，一度飙升至211人民币，随后回落至201人民币，最终以206人民币收盘。"
]

"""
[
    {"role": "system",      "content": f"你帮我完成信息抽取，我给你句子，你抽取{schema}信息，按JSON字符串输出，如果某些信息不存在，用'原文未提及'表示，请参考如下示例："},

    {"role": "user",        "content": "2023-01-10，股市震荡。股票强大科技A股今日开盘价100人民币，一度飙升至105人民币，随后回落至98人民币，最终以102人民币收盘，成交量达到520000。"},
    {"role": "assistant",   "content": '{"日期":"2023-01-10","股票名称":"强大科技A股","开盘价":"100人民币","收盘价":"102人民币","成交量":"520000"}'},
    {"role": "user",        "content": "2024-05-16，股市利好。股票英伟达美股今日开盘价105美元，一度飙升至109美元，随后回落至100美元，最终以116美元收盘，成交量达到3560000。"},
    {"role": "assistant",   "content": '{"日期":"2024-05-16","股票名称":"英伟达美股","开盘价":"105美元","收盘价":"116美元","成交量":"3560000"}'},

    {"role": "user",        "content": f"按照上述示例，现在抽取这个句子的信息：{要抽取的句子文本}"}]}
]
"""

//设置模型的system角色
messages = [
    {"role": "system",
     "content": f"你帮我完成信息抽取，我给你句子，你抽取{schema}信息，按JSON字符串输出，如果某些信息不存在，用'原文未提及'表示，请参考如下示例："}
]

//将答案示例放在模型中供模型进行少量学习
for example in examples_data:
    messages.append(
        {"role": "user", "content": example["content"]}
    )

    messages.append(
        {"role": "assistant", "content": json.dumps(example["answers"], ensure_ascii=False)}
    )

//把每个问题丢给模型，让他生成金融提取的答案
for q in questions:
    response = client.chat.completions.create(
        model="qwen3-max",
        messages=messages + [{"role": "user", "content": f"按照上述的示例，现在抽取这个句子的信息：{q}"}]
    )

    print(response.choices[0].message.content)

实现效果

 

 

最后使用提示词工程来实现金融文本的匹配

from openai import OpenAI

client = OpenAI(
    # base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
    base_url="http://localhost:11434/v1"
)

examples_data = {
    "是": [
        ("公司ABC发布了季度财报，显示盈利增长。", "财报披露，公司ABC利润上升。"),
        ("公司ITCAST发布了年度财报，显示盈利大幅度增长。", "财报披露，公司ITCAST更赚钱了。")
    ],
    "不是": [
        ("黄金价格下跌，投资者抛售。", "外汇市场交易额创下新高。"),
        ("央行降息，刺激经济增长。", "新能源技术的创新。")
    ]
}

questions = [
    ("利率上升，影响房地产市场。", "高利率对房地产有一定的冲击。"),
    ("油价大幅度下跌，能源公司面临挑战。", "未来智能城市的建设趋势越加明显。"),
    ("股票市场今日大涨，投资者乐观。", "持续上涨的市场让投资者感到满意。")
]

"""
    {"role": "system",      "content": f"你帮我完成文本匹配，我给你2个句子，被[]包围，你判断它们是否匹配，回答是或不是，请参考如下示例："},
     
    {"role": "user",        "content": "句子1：[公司ABC发布了季度财报，显示盈利增长。]句子2：[财报披露，公司ABC利润上升。]"},
    {"role": "assistant",   "content": "是"},
    {"role": "user",        "content": "句子1：[公司ITCAST发布了年度财报，显示盈利大幅度增长。]句子2：[财报披露，公司ITCAST更赚钱了。]"},
    {"role": "assistant",   "content": "是"},
    {"role": "user",        "content": "句子1：[黄金价格下跌，投资者抛售。]句子2：[外汇市场交易额创下新高。]"},
    {"role": "assistant",   "content": "不是"},
    {"role": "user",        "content": "句子1：[央行降息，刺激经济增长。]句子2：[新能源技术的创新。]"},
    {"role": "assistant",   "content": "不是"}, 
    
    {"role": "user",        "content": f"按照上述示例，回答这2个句子的情况。句子1: [...]，句子2: [...]"}
"""

messages = [
    {"role": "system", "content": f"你帮我完成文本匹配，我给你2个句子，被[]包围，你判断它们是否匹配，回答是或不是，请参考如下示例："},
]

for key, value in examples_data.items():
    for t in value:
        messages.append(
            {"role": "user", "content": f"句子1：[{t[0]}]，句子2：[{t[1]}]"}
        )
        messages.append(
            {"role": "assistant", "content": key}
        )

for q in questions:
    response = client.chat.completions.create(
        model="qwen3:4b",
        messages=messages + [{"role": "user", "content": f"句子1：[{q[0]}]，句子2：[{q[1]}]"}]
    )

    print(response.choices[0].message.content)

 

 

LangChain

LangChain是围绕LLMs（大语言模型）建立的一个框架，它自身并不开发LLMs，它的核心理念是为各种LLMs实现通用的接口，把LLMs相关的组件“链接”在一起，简化LLMs应用的开发难度，方便开发者快速地开发复杂地LLMs的应用。

 

Langchain的功能：

prompts：提示词工程

model：调用各类模型

history：管理会话历史记录（记忆）

indexes：管理和分析各类文档

chains：构建功能的执行链条

Agent：构建智能体

 

Langchain的环境部署

可以到pycharm的编辑器里面分别搜索下载

langchain : 核心包

langchain-community：社区支持包，提供了更多的第三方模型调用

langchain-ollama：Ollama支持包，没用Ollama的可以不下

dashscope：阿里云通用千问的python sdk

chromadb：轻量向量数据库

编辑器软件包安装如下

 

 

RAG

LangChain调用大模型

通用的基础大模型存在一些问题：

LLM的知识不是实时的，模型训练好之后不具备自动更新知识的能力，会导致部分消息滞后。

LLM的领域知识是缺乏的，大模型的知识来源于训练数据，这些数据主要来自公开的互联网和开源数据集，无法覆盖特定领域或高度专业化的内部知识。

幻觉问题，LLM有时会回答生成中看似合理但实际上是错误的信息。

数据安全性

 

RAG：检索增强生成技术，利用检索外部文档提升生成结果质量。为大模型提供了从特定数据源检索到的信息，以此来修正和补充生成的答案。RAG=检索技术+LLM提示。

 

RAG 标准流程由索引、检索和生成三个核心阶段组成

索引阶段：通过处理多种来源多种格式的文档提取其中文本，将其切割成标准长度的文本块，并进行嵌入向量化，向量存储在向量数据库中。

检索阶段：用户输入的查询被转化为向量表示，通过相似度匹配从向量数据库中检索出最相关的文本块。

生成阶段：检索到的相关文本与原始查询共同构成提示词，输入大语言模型，生成精准且上下文匹配的回答。

 

RAG就是在向模型提问之前基于已有的知识库或文档内容做检索，确保向模型提问的内容更精准以及包含足够的信息量用以提供给模型。

RAG核心价值：解决知识实效性问题、降低模型幻觉、无需重新训练模型。

 

RAG中，向量数据库是一个重要的节点

离线阶段：知识和信息作为向量存入向量库

在线阶段：用户的提问作为向量去向量库中进行匹配。

 

向量：它把一段文字的语义信息，转换成一串固定长度的数字列表，让计算机“看懂”文字的含义并作相似度计算。

 

文本嵌入模型通过深度学习等技术，从文本中提取语义特征并映射为固定长度的数字序列。向量嵌入的过程中，我们一般选用合适的文本嵌入模型来完成。在向量匹配的过程中，主要通过余弦相似度等算法来完成。

如何更为精准的完成语义匹配，生成向量的纬度是一个很重要的指标。比如1536维表示一段文本固定得到1536长度的数字序列。而且，生成向量的纬度越多，就更好的记录文本的语义特征，做语义匹配会更加精准。更多的向量会在计算存储与匹配中带来更大的压力，所以选择合适的向量维度也很重要。

 

Langchain目前支持三种类型的模型：LLMs（大语言模型）、chat model（聊天模型）、Embedding Models（嵌入模型）

我们可以使用Langchain来调用大模型，如果你在电脑的环境变量设置的api-key无法识别，可以在pycharm中的编辑运行环境中配置环境变量。

 

下面是调用千问3大语言模型的代码示例

//导入langchain中关于通义千问系列的支持
from langchain_community.llms.tongyi import Tongyi

//创建qwen-max的大语言模型
model = Tongyi(model = "qwen-max")

//调用提问
res = model.invoke("牛顿为什么老年信天主教")

//打印回答
print(res);

成功实现

 

如果需要流式输出结果，需要将invoke方法改为stream方法即可，并将结果进行循环监听刷新区来进行输出。

 

有的人可能会疑惑流式输出是怎么做到的，其实是因为当res=model.stream后，其实返回的是一个类似于传送带的东西（迭代器），这行代码做完就直接运行下面的代码去了。所以并不会阻塞程序的运行。

import time
from langchain_community.llms.tongyi import Tongyi

# 假设你已经配置好了 Key
model = Tongyi(model="qwen-max")

print("1. 准备调用 stream...")
start_time = time.time()

# 这一行代码执行得非常快，几乎不耗时
# 因为它只是建立了一个“管道”，并没有等待 AI 写完
res = model.stream("写一首关于春天的五行诗") 

print(f"2. stream() 函数已返回！耗时: {time.time() - start_time:.4f}秒")
print("   (看，这里 AI 还没写完，甚至可能还没开始写，但代码已经往下走了)")
print("-" * 20)

print("3. 开始循环接收数据：")
for chunk in res:
    # 真正的等待时间发生在这里
    print(chunk, end="", flush=True)

print("\n4. 循环结束，AI 回答完毕。")

 

Chat Model（聊天模型）

聊天模型包含下面几种类型，使用时需要按照约定传入合适的值：

AIMessage：就是AI输出的消息，可以是针对问题的回答。（类似Assisant）

HumanMessage：人类消息就是用户消息，由人给出的信息发送给LLMs的提示信息。（user）。

SystemMessage：可以用于指定模型具体所处的环境和背景，在这里可以给出具体的知识(system)。

 

先导入通义支持的聊天模型后，向大语言模型一样进行配置后调用即可（角色不同）

from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

# 得到模型对象, qwen3-max就是聊天模型
model = ChatTongyi(model="qwen3-max")

# 准备消息列表
messages = [
    SystemMessage(content="你是一个边塞诗人。"),
    HumanMessage(content="写一首唐诗"),
    AIMessage(content="锄禾日当午，汗滴禾下土，谁知盘中餐，粒粒皆辛苦。"),
    HumanMessage(content="按照你上一个回复的格式，在写一首唐诗。")
]

# 调用stream流式执行
res = model.stream(input=messages)

# for循环迭代打印输出，通过.content来获取到内容
for chunk in res:
    print(chunk.content, end="", flush=True)

Landchain聊天模型可以进行简写，主要是角色也可以用双引号进行包装。

from langchain_community.chat_models.tongyi import ChatTongyi

# 得到模型对象, qwen3-max就是聊天模型
model = ChatTongyi(model="qwen3-max")

# 准备消息列表
messages = [
    # (角色，内容)  角色：system/human/ai
    ("system", "你是一个边塞诗人。"),
    ("human", "写一首唐诗。"),
    ("ai", "锄禾日当午，汗滴禾下土，谁知盘中餐，粒粒皆辛苦。"),
    ("human", "按照你上一个回复的格式，在写一首唐诗。")
]

# 调用stream流式执行
res = model.stream(input=messages)

# for循环迭代打印输出，通过.content来获取到内容
for chunk in res:
    print(chunk.content, end="", flush=True)

它们两的区别是不简写的方式是静态的，直接得到了Message类的类对象

而简写的方式是动态的，需要在运行的时候由Langchain内部机制转换为Message对象

messages = [

SystemMessage(content="你是一个边塞诗人。"),

HumanMessage(content="写一首唐诗"),

AIMessage(content="锄禾日当午，汗滴禾下土，谁知盘中餐，粒粒皆辛苦。"),

HumanMessage(content="按照你上一个回复的格式，在写一首唐诗。")

]

messages = [

# (角色，内容) 角色：system/human/ai

("system", "你是一个边塞诗人。"),

("human", "写一首唐诗。"),

("ai", "锄禾日当午，汗滴禾下土，谁知盘中餐，粒粒皆辛苦。"),

("human", "按照你上一个回复的格式，在写一首唐诗。")

]

简写的好处是避免导包，写起来更简单。

 

 

LangChain调用嵌入模型

与上面相似，如果不指定模型的话默认模型是text-embeddings-v1

embed_query参数为字符串

embed_document参数为列表

他们的结果都是返回一大串数字（数字序列用来存储语义的）

from langchain_community.embeddings import DashScopeEmbeddings

//默认模型为text-embeddings-v1
model = DashScopeEmbeddings()

print(model.embed_query("我喜欢你"))
print("\n\n\n\n\n\n")
print(model.embed_documents(["我讨厌你","猎奇作品","哼哈二将"]))

 

 

PromptTemplate类

LangChain提供了PromptTemplate类，用来协助优化提示词。PromptTemplate表示提示词模板，可以构建一个自定义的基础提示词模板，支持变量的注入，最终生成所需的提示词。

from langchain_core.prompts import PromptTemplate
from langchain_community.llms.tongyi import Tongyi

prompt_template = PromptTemplate.from_template(
    "我的孩子姓{lastname}，生了个{gender}，你帮我取个名字，简洁回答"
)

#可以调用.format方法注入信息使用
prompt_text = prompt_template.format(lastname = "江",gender = "女儿")

model = Tongyi(model="qwen-max")
res = model.invoke(input=prompt_text)
print(res)

 

但是也可以以链的方式对模型进行注入(链的形式后面再说)

from langchain_core.prompts import PromptTemplate
from langchain_community.llms.tongyi import Tongyi

prompt_template = PromptTemplate.from_template(
    "我的孩子姓{lastname}，生了个{gender}，你帮我取个名字，简洁回答"
)

model = Tongyi(model="qwen-max")
chain = prompt_template | model
res = chain.invoke(input={"lastname":"刘", "gender" : "女孩"})
print(res)

 

FewShort提示词模板

from langchain_core.prompts import PromptTemplate, FewShotPromptTemplate
from langchain_community.llms.tongyi import Tongyi

# 示例的模板
example_template = PromptTemplate.from_template("单词：{word}, 反义词：{antonym}")

# 示例的动态数据注入 要求是list内部套字典
examples_data = [
    {"word": "大", "antonym": "小"},
    {"word": "上", "antonym": "下"},
]

few_shot_template = FewShotPromptTemplate(
    example_prompt=example_template,    # 示例数据的模板
    examples=examples_data,             # 示例的数据（用来注入动态数据的），list内套字典
    prefix="告知我单词的反义词，我提供如下的示例：",                   # 示例之前的提示词
    suffix="基于前面的示例告知我，{input_word}的反义词是？",          # 示例之后的提示词
    input_variables=['input_word']      # 声明在前缀或后缀中所需要注入的变量名
)

prompt_text = few_shot_template.invoke(input={"input_word": "左"}).to_string()
print(prompt_text)

model = Tongyi(model="qwen-max")

print(model.invoke(input=prompt_text))

各个属性说明

example_prompt：给模型示例的模板

examples：用来注入动态数据的，类型为list套字典。

prefix：前缀，把少量示例丢给模型学习前所提供的文字说明

suffix：后缀，模型经过少量示例学习后对所要做的事的说明

input_variables：声明在前缀或后缀中所需注入的变量名

 

 

ChatPromptTemplate

ChatPromptTemplate：支持注入任意数量的历史会话信息

通过from_messages方法，总列表中获取多轮会话作为聊天的基础模板

ChatPromptTemplate.from_message(
    [
        ("system","....."),
        ("ai","....."),
        ....
        ("human","....")
    ]
)

历史会话信息并不是静态的，而是随着对话的进行不停地积攒，即动态的。所以，历史会话需要支持动态注入。而MessagePlaceHolder作为占位符（也相当于类对象），提供了占位的key，它相当于告诉模板，这里需要留个空位当作动态插槽，不管里面有多少条消息（字典，列表），都需要插在这里。然后基于invoke动态注入历史会话记录。必须是invoke，format无法注入因为它只能解析变量名而不能解析占位符和字典、列表。

chat_template = ChatPromptTemplate.from_message(
    [
        ("system","....."),
        ("ai","....."),
        MessagePlaceholder("history"),
        ("human","....")
    ]
)

history_data = [
    ("ai","....."),
    ("human","...."),
    ("ai","....."),
    ("human","....")
]

prompt_data = chat_template.invoke("history": history_data)

 

 

chain链

将组件串联，上一个组件的输出作为下一个组件的输入是LangChain链的核心工作原理，实现数据的自动化流转与组件的协同工作。

chain = prompt_template | model

核心前提：只有Runnable子类的对象才能入链。不过我们目前所学到的组件都是Runnable接口的子类。

继承图如下：

 

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.chat_models.tongyi import ChatTongyi

chat_prompt_template = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个边塞诗人，可以作诗。"),
        MessagesPlaceholder("history"),
        ("human", "请再来一首唐诗"),
    ]
)

history_data = [
    ("human", "你来写一个唐诗"),
    ("ai", "床前明月光，疑是地上霜，举头望明月，低头思故乡"),
    ("human", "好诗再来一个"),
    ("ai", "锄禾日当午，汗滴禾下锄，谁知盘中餐，粒粒皆辛苦"),
]

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

# 组成链，要求每一个组件都是Runnable接口的子类
chain = chat_prompt_template | model

# 通过链去调用invoke或stream
# res = chain.invoke({"history": history_data})
# print(res.content)

# 通过stream流式输出
for chunk in chain.stream({"history": history_data}):
    print(chunk.content, end="", flush=True)

 

chain变量是RunnableSequence类型的，因为Runnable基类内部对__or__方法的改写，导致在后面继续使用"|"添加新的组件时，始终返回的是RunnableSequence类型，这就是链的基础架构。

 

StrOutputParser字符串输出解析器

首先先看一个例子 chain = prompt | model | model

会输出错误的信息，是因为model返回的是AImessage的类，而我们输入给模型的类必须是PromptValue。（参数类型错误）

而StrOutPutParser是LangChain内置的简单字符串解析器，可以将AIMessage解析为简单的字符串，符合了模型invoke的方法要求（可传入字符串，不接收AIMessage）

parser = StrOutPutParser()
chain = prompt | model | parser | model  #不会报错了

 

JsonOutputParser

与字符串输出解析器同理，名为json输出解析器，将AIMessage的内容解析成json格式（前提是你要求模型输出的内容为字典格式）

不过我们之前所做的chain调用执行链并不完整，我们应该把上一个模型的输出，经过数据处理后才输入下一个模型，而不是没有被处理就直接丢给下一个模型。

数据处理：将模型输出的AIMessage转换成字典注入到新的提示词模板中再输入给下一个模型。

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

str_parser = StrOutputParser()
json_parser = JsonOutputParser()

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

first_prompt = PromptTemplate.from_template(
    "我的武器攻击值为{attack}，我的血量为{blood}，根据这两个属性值为我生成战斗力值，"
    "仅仅回复我我的战斗力值为多少，并封装为json格式给我，要求key是power,value就是战斗力值，"
    "请严格遵循格式要求，无需额外回复"
)

second_prompt = PromptTemplate.from_template(
    "BOSS的战斗力为1000，而我的战斗力为{power}，分析我能不能打败BOSS"
)

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

res = chain.invoke({"attack":50,"blood":800})

print(res)
print(type(res))

成功结果

 

在构建链的时候要注意整体的兼容性，注意前后组件的输入和输出要求。

 

自定义函数加入链

除了上面两个关于AIMessage的解析器外，我们也可以自己编写LangChain匿名函数来实现自定义逻辑的数据转换，我们需要基于RunnableLambda类来进行实现。RunnableLambda类是LangChain内置的，将普通函数等转换为Runnable接口实例，方便自定义函数加入Chain。本质是把函数自动转换为RunnableLambda，所以能够加入链。

 

语法：RunnableLambda（Lambda...）

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

str_parser = StrOutputParser()
# json_parser = JsonOutputParser()
my_func = RunnableLambda(lambda ai_msg:{"power":ai_msg.content})

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

first_prompt = PromptTemplate.from_template(
    "我的武器攻击值为{attack}，我的血量为{blood}，根据这两个属性值为我生成战斗力值，"
    "仅仅回复我我的战斗力值为多少"
    "无需额外回复"
)

second_prompt = PromptTemplate.from_template(
    "BOSS的战斗力为1000，而我的战斗力为{power}，分析我能不能打败BOSS"
)

chain = first_prompt | model | my_func | second_prompt |  model | str_parser

res = chain.invoke({"attack":1,"blood":20})

print(res)
print(type(res))

成功结果

 

可以看出相比于必须把内容强转json，自定义函数加入链会更灵活和方便。

 

 

Memory

临时记忆

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

基于两个类，RunnableWithMessageHistory和InMemoryChatMessageHistory。

RunnableWithMessageHistory：在原有链的基础上创建带有历史记录功能的新链（也是Runnable实例）

InMemoryChatMessageHistory：为历史记录提供内存存储。

 

下面为创建临时记忆会话的示例。

from langchain_community.chat_models import ChatTongyi
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableWithMessageHistory


#打印提示词，方便调试
def print_prompt(full_prompt):
    print("="*20, full_prompt.to_string(), "="*20)
    return full_prompt

model = ChatTongyi(model="qwen3-max")
prompt = PromptTemplate.from_template(
    "你需要根据对话历史回复用户问题。对话历史:{chat_history}。"
    "用户当前输入:{input}，请给出简洁回应"
)

base_chain = prompt | print_prompt() | model | StrOutputParser()

#存放多个会话ID所对应的历史会话记录
chat_history_store = {}
def get_history(session_id):
    if(session_id not in chat_history_store):
        #存入新的实例
        chat_history_store[session_id] = InMemoryChatMessageHistory()
    return chat_history_store[session_id]

conversation_chain = RunnableWithMessageHistory(
    base_chain,  #被附加历史消息的Runnable，通常是chain
    get_history,  #获取历史会话的函数
    input_messages_key="input",
    history_messages_key="chat_history"
)

if __name__ == '__main__':
    session_config = {"configurable" : {"session_id":"user_001"}}

    print(conversation_chain.invoke({"input":"小明有一只猫"},session_config))
    print(conversation_chain.invoke({"input": "小刚有五匹马"}, session_config))
    print(conversation_chain.invoke({"input": "总共有几只宠物"}, session_config))

流程：先把input内容注入到base_chain中的用户输入占位符中。conversation_chain里面有个获取历史会话的函数，它会将传进来的字典中寻找名为"configurable"的字典，寻找里面对应的属性作为参数调用获取历史对话的函数，返回了InMemoryChatMessageHistor对象，拿到这个对象后，RunnableWithMessageHistory会自动读取里面的所有历史内容，并将这些内容放到history_messages_key所表示的历史记录占位符中。

 

我一开始会有为什么conversation_chain怎么读取我sessionId的问题，后面查资料得知，session_id 能被自动读取，并不是因为 invoke 函数有什么“固定格式”的魔法，而是因为 RunnableWithMessageHistory 这个类被特意设计成会去 config 字典里的 configurable 子字典中寻找它需要的配置项。

 

成功运行如图

 

 

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

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

 

Memory长期会话记忆

使用FileChatMessageHistory类实现，核心思路是基于文件存储会话记录，以session_id为文件名，不同session_id有不同文件存储消息。它继承了BaseChatMessageHistory实现3种方法，分别是：add_messages：添加消息、messages:获取消息、clear:清除消息。

 

我们进行conversation_chain的调用时，我们的LangChain框架会先自动调用get_history获取历史会话记录，而这个会话记录，我们通过我们自定义的FileChatMessageHistory类（继承BaseChatMessageHistory类）中的messages方法获取。然后模型会根据历史消息和用户的新输入来进行回答，回答完毕后，LangChain框架会将这对消息放进历史消息的队列中通过add_messages方法重新写入文件中。

import json
import os.path
from typing import Sequence

from langchain_community.chat_models import ChatTongyi
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import BaseMessage, message_to_dict, messages_from_dict
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableWithMessageHistory



class FileChatMessageHistory(BaseChatMessageHistory):
    def __init__(self , session_id, storge_path):
        self.session_id = session_id    #会话id
        self.storge_path = storge_path  #不同会话id的存储文件，所在的文件夹路径
        #完整的文件夹路径
        self.file_path = os.path.join(self.storge_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)   #新的和已有的融合

        # 将数据同步写入本地文件中
        #将BaseMessage对象转换成字典，然后借助json模块以json字符串写入文件
        # 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]:
        try:
            with open(self.file_path,"r",encoding="UTF-8") as f:
                messages_data = json.load(f) #返回的是一堆字典
                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)


    #打印提示词，方便调试
def print_prompt(full_prompt):
    print("="*20, full_prompt.to_string(), "="*20)
    return full_prompt

model = ChatTongyi(model="qwen3-max")
prompt = PromptTemplate.from_template(
    "你需要根据对话历史回复用户问题。对话历史:{chat_history}。"
    "用户当前输入:{input}，请给出简洁回应"
)

base_chain = prompt | print_prompt | model | StrOutputParser()

#存放多个会话ID所对应的历史会话记录
def get_history(session_id):
    return FileChatMessageHistory(session_id,"./chat_history")

conversation_chain = RunnableWithMessageHistory(
    base_chain,  #被附加历史消息的Runnable，通常是chain
    get_history,  #获取历史会话的函数，这个函数返回的是InMemoryChatMessageHistor对象
    # RunnableWithMessageHistory 拿到这个对象后，会立即从中读取所有已有的对话内容。
    #然后把这些读取到的历史内容，自动填充到你 PromptTemplate 中定义的 {chat_history} 占位符里。
    input_messages_key="input",     #表示用户输入在模板中的占位符
    history_messages_key="chat_history"     #表示历史消息在模板中的占位符
)

if __name__ == '__main__':
    session_config = {"configurable" : {"session_id":"user_001"}}
    print(conversation_chain.invoke({"input":"小明有一只猫"},session_config))
    print(conversation_chain.invoke({"input": "小刚有五匹马"}, session_config))
    print(conversation_chain.invoke({"input": "总共有几只宠物"}, session_config))

注意：@property是python的一个修饰器，它将一个方法伪装成成员属性。，让我们调用时像访问变量一样，不用加括号。

 

成功结果

 

会创建一个文件将历史会话记录作为json格式保存起来，达到长期记忆的结果。（因为即使程序关闭，这个文件依旧存在）

 

文档加载器

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

文档加载器需实现BaseLoader接口

Class Document是Langchain内文档的统一载体，所有文档加载器最终返回此类的实例。

 

Document类里面的核心属性是

page_content：文档内容

metadata：文档元数据（字典）

 

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

load（）：一次性加载全部文档。

lazy_load（）：延迟流式传输文档，避免内存溢出。

loader = CSVLoader(
    ....初始化参数
)

#一次性加载全部文档
documents = loader.load()

#对于大数据集，分段返回文档
for document in loader.lazy_load():
    print(document)

 

CSVLoader就是在LangChain框架中加载CSV文件，加载成功得到（封装）Document对象。

CSV本质上是纯文本的表格文件，拥有表头和对应文本。

from langchain_community.document_loaders import CSVLoader


loader = CSVLoader(
    file_path="./data/stu.csv",
    csv_args={
        "delimiter": ",",       # 指定分隔符
        "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用于将JSON数据加载为Document对象，需额外安装jq库，jq是一个跨平台的json解析工具，LangChain底层对json的解析就是基于jq工具实现的。

因为全部文档加载器的使用方法都差不多，我就不多赘述了。

from langchain_community.document_loaders import JSONLoader

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

document = loader.load()
print(document)

 

TextLoader，读取文本文件（txt），将全部内容放入一个Document对象中。

RecursiveCharacterTextSplitter是递归字符文本分割器，主要用于将自然段落分割大文档。是LangChain官方推荐的默认字符分割器。它在保持上下文完整性和控制片段大小之间实现了良好平衡，开箱即用效果佳。

下面是将一个大文档加载后对其进行文本分割的代码示例

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就是PDF文件的加载器了，使用前需下载pypdf的依赖库，使用与上面类似，且十分简单。

from langchain_community.document_loaders import PyPDFLoader

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

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

 

 

向量存储与检索

基于LangChain的向量存储，存储嵌入数据，并执行相似性搜索可以基于InMemoryVectorStore完成内部向量存储或者Chroma完成外部数据库向量存储。

 

 

向量存储类均提供三个通用API接口：add_document，delete，similarity_search

 

我们需要先创建一个Chroma对象（操作向量存储的），初始化它的嵌入文本转向量模型，数据存放的文件夹和数据库的表名称，然后以CSV文件为例，先用CSVLoader将这个文档加载成为Document对象，将这个对象传入Chroma中实现向量的存储，最后使用相似匹配的方法返回匹配到的文档（由匹配到的向量得来的）结果。（可以指定过滤器和返回数目）

下面是外部实现向量的持久化存储的代码实现

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": "黑马程序员"}
)

print(result)

 

成功结果

 

 

基于向量检索构建提示词，其实就是将用户的输入与向量库中的向量进行匹配后，将匹配到的文档加入到用户的提示词中一起丢给模型，不多赘述了。

"""
提示词：用户的提问 + 向量库中检索到的参考资料
"""
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类

那我们想将向量检索加入链该怎么实现呢？我们可以使用RunnablePassthrough类来进行实现。

简单来说，RunnablePassthrough类的作用是将输入数据原封不动地传递下去，不做任何修改。而且优势是不用我们进行手动的向量文档匹配，它会拿着用户输入自行取匹配。

代码实现如下：

"""
提示词：用户的提问 + 向量库中检索到的参考资料
"""
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


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 = "怎么减肥？"

# langchain中向量存储对象，有一个方法：as_retriever，可以返回一个Runnable接口的子类实例对象
retriever = vector_store.as_retriever(search_kwargs={"k": 2})


def format_func(docs: list[Document]):
    if not docs:
        return "无相关参考资料"

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

    return formatted_str

# chain
chain = (
    {"input": RunnablePassthrough(), "context": retriever | format_func} | prompt | print_prompt | model | StrOutputParser()
)

res = chain.invoke(input_text)
print(res)
"""
retriever:
    - 输入：用户的提问       str
    - 输出：向量库的检索结果  list[Document]
prompt:
    - 输入：用户的提问 + 向量库的检索结果   dict
    - 输出：完整的提示词                 PromptValue
"""

对于这段代码，我有很困惑的一点，文件检索器如何获取用户的输入，我并没看到任何有关retriever相关的字符串传参。于是我去查资料，发现vector_store.as_retriever返回的是一个“待命”的函数，当chain进行调用后RunnablePassthrough()会拿到原有的用户输入传给input，而传到retriever时（因为是字典，所以是并行传入），会触发这个函数的调用，也就是retriever.invoke(用户输入)，将用户输入变成向量拿去向量库中进行匹配，返回k个匹配度最高的文档，但是需要先把这文档变成字符串类型（format_func函数实现）才能传入到提示词模板中（prompt）。

 

 

 

 

 

 

 

# python # pycharm
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

PyTorch LSTM层输入维度不匹配怎么办？教你一招避坑

avatar AtomGit开源社区
cover

过度授权 - 大语言模型 OWASP TOP 10系列

avatar AtomGit开源社区

智能识别告警系统完整方案

本文提出了一套完整的智能识别告警系统方案，通过AI视觉技术实现违规行为自动检测。系统采用分层架构设计，包含前端采集、接入服务、AI推理、规则判定、告警推送等模块，支持厨师帽佩戴、违规洗澡、人员闯入等多场景识别。技术选型上采用YOLO系列目标检测模型，结合自定义规则引擎实现精准判定。系统具备闭环调优机制，通过难样本收集、模型微调实现持续优化。方案优势在于轻量易部署、场景可扩展、识别准确率自提升等特点

avatar AtomGit开源社区