RAG：

介绍：

通⽤的基础⼤模型存在一些问题：

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

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

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

数据安全性等

RAG（Retrieval-Augmented Generation，检索增强生成）是一种大语言模型增强技术，它通过从指定数据源中检索相关信息，为大模型提供外部知识，从而修正、补充模型的生成内容，有效解决幻觉问题。其核心逻辑可概括为：RAG = 检索技术 + LLM 提示工程

工作原理：

工作流程图解：

RAG 标准流程：

纯 LLM 模式：用户直接提问，模型仅靠自身训练数据生成回答。面对未训练过的新知识、私有数据，要么无法回答，要么凭空编造（幻觉），就像图中对 Sam Altman 事件的回答，模型因无对应知识只能表示无法评论

LLM + RAG 模式：给大模型加装 "外挂知识库"，先检索相关资料，再结合资料生成回答，从根源上保证回答的准确性和可信度

简单来说，RAG 工作分为两条线： 离线准备线 / 在线服务线

离线流程是检索准备阶段，在用户交互前完成：先加载各类文档与私有知识，将长文本分割为语义完整的文本块，再通过嵌入模型向量化，最终存入向量数据库，搭建起可快速检索的结构化知识库

在线流程是用户交互的实时链路：用户提问后，先将问题向量化，在向量库中匹配召回相关参考片段；再把问题与参考资料融合成 Prompt，输入大模型；最终由模型基于真实资料生成准确、无幻觉的回答

核心价值：

时效性保障：解决大模型知识 "滞后" 问题，可实时接入最新财报、行业政策、动态新闻，让回答始终精准且贴合最新业务场景

降低幻觉风险：强制基于检索到的真实事实生成，所有结论均有明确依据，从根源上减少凭空捏造的可能，保证回答可信度

高效低成本更新：无需模型微调（Fine-tuning）重构整个模型，仅需更新外部知识库即可快速补充新信息，大幅降低运维与迭代成本，适配业务快速变化需求

标准流程：

RAG 标准流程核心分为索引、检索、生成三个阶段，是大模型精准回答的关键链路：

索引阶段（离线准备）：

这是所有工作的基础，核心是完成数据的标准化处理与向量化转换；首先对原始知识库（如书籍、文档、报告等）进行清洗和分割，将长文本拆解为若干个短文本片段（Chunk），每个片段作为后续检索的基本单位；随后通过文本嵌入模型，将这些文本片段逐一转换为计算机可识别的数字向量（Embedding），最终将这些向量数据有序存储在向量数据库（Vector database）中，形成完整的 "向量知识库"，为后续精准匹配做好准备

检索阶段（精准匹配）：

当用户提出具体问题后，首先将用户问题（Query）转换为与上述流程一致的向量形式；随后在向量数据库中进行相似性匹配，通过计算向量之间的相似度，精准召回与用户问题语义最相关的 Top-K 个文本片段；这一步是核心枢纽，直接决定了后续生成内容的准确性与针对性

生成阶段（内容输出）：

将用户原始问题、召回的相关文本片段整合，组合成完整的提示词（Prompt），输入到大语言模型（LLM）中；模型基于这些真实参考资料，生成逻辑严谨、内容准确的回答，彻底避免 "凭空臆造" 的幻觉问题，同时适配私有数据、行业专属知识等个性化场景

Chunk（文本片段）:

指将原始长文本分割后的短文本单元，是 RAG 系统中存储、检索的基本单位；分割后的片段长度需适中，过短会导致信息不完整，过长则会降低匹配精度，通常根据业务场景控制在几百字到上千字不等

文本嵌入（Embedding）:

也叫向量嵌入，是将自然语言文本转换为数字向量的技术过程；通过预训练的嵌入模型，让语义相近的文本对应向量空间中距离相近的位置，从而实现计算机对文本语义的理解与匹配，是 RAG 实现语义检索的核心基础

向量数据库（Vector database）：

专门用于存储向量数据，并支持快速、高效相似度检索的数据库；区别于传统数据库的结构化查询，向量数据库能基于语义匹配精准召回相关内容，是 RAG 系统的核心存储与检索载体

目前主流、热门的向量数据库：

向量的基础概念：

向量库在 RAG（检索增强生成）流程中的两个关键角色：知识存储中心和实时检索引擎

离线流程：知识入库准备：

在用户提问前，系统会预先处理所有可用的知识和信息；这些信息会通过向量嵌入（向量化）技术，被转换成计算机能理解的数字向量；随后，这些向量被统一存入向量库中，形成一个结构化的、可随时检索的知识库

在线流程：实时匹配检索：

当用户提出问题时，系统会立即将这个问题也进行向量化处理；然后，利用这个问题向量，在已建立的向量库中进行快速的相似度匹配，精准地找到与问题最相关的知识片段

向量（Vector）就是文本的 "数学身份证"：它把一段文字的语义信息，转换成一串固定长度的数字列表，让计算机能 "看懂" 文字的含义并做相似度计算；简单来说，就是让计算机更方便的理解不同的文本内容，是否表述的是一个意思

向量的计算：

文本嵌入是 RAG 流程中实现语义理解的核心技术，通过 text-embedding-v1 这类深度学习模型，将自然语言文本提取语义特征，映射为固定长度的数字向量，让计算机能够 "读懂" 文本的含义

在实际应用中，会用合适的文本嵌入模型完成向量化转换，再通过余弦相似度等算法，计算不同文本向量之间的相似程度，以此判断两段文本是否语义相近；如语义相同的 "如何快速学习 RAG" 和 "RAG 如何快速学会"，生成的向量高度相似，相似度接近 1；而语义无关的 "歪比巴卜"，与前者的向量相似度则极低

如何更为精准的完成语义匹配，生成向量的维度是一个很重要的指标

如 text-embedding-v1 模型，可以生成 1536 维的向量（一段文本固定得到 1536 个数字序列），比较实用；（1536 个数字表示这段文本在 1536 个主题（抽象的语义特征）方向上的得分（强度））

生成向量的维度越多，就更好的记录文本的语义特征，做语义匹配会更加精准；但更多的向量会在计算、存储和匹配过程中，带来更大的压力；所以选择合适的向量维度需要在精确和性能之间做平衡，一般 1536 维算是比较好的选择

余弦相似度：

向量的数字序列，共同决定了向量在高维空间中的方向和长度；而余弦相似度主要就是撇除长度的影响，得到方向的夹角；夹角越小越相似，即方向相同

以一维和二维向量为例体现向量的方向和长度：

余弦相似度主要匹配的就是：同向（无所谓长度）

在文本向量语义匹配中，余弦相似度是衡量两个向量方向相似程度的核心算法，即判断两段文本语义是否相近

代码实现：

import numpy as np
"""
计算两个向量的余弦相似度（衡量方向相似性，剔除长度影响）

参数：
    vec_a (np.array): 向量A
    vec_b (np.array): 向量B
返回：
    float: 余弦相似度结果（范围[-1,1]，越接近1方向越一致）
公式：
    cos_sim = (vec_a · vec_b) / (||vec_a|| × ||vec_b||)
    拆解：
    1. 点积：vec_a · vec_b = vec_a[0]×vec_b[0] + vec_a[1]×vec_b[1] + ... + vec_a[n]×vec_b[n]
    2. 模长：||vec_a|| = √(vec_a[0]² + vec_a[1]² + ... + vec_a[n]²)
    3. 模长：||vec_b|| = √(vec_b[0]² + vec_b[1]² + ... + vec_b[n]²)

A: [0.5, 0.5]
B: [0.7, 0.7]
C: [0.7, 0.5]
D: [-0.6, -0.5]
"""


def get_dot(vec_a, vec_b):
    """计算2个向量的点积，2个向量同维度数字乘积之和"""
    if len(vec_a) != len(vec_b):
        raise ValueError("2个向量必须维度数量相同")

    dot_sum = 0
    for a, b in zip(vec_a, vec_b):
        dot_sum += a * b

    return dot_sum


def get_norm(vec):
    """计算单个向量的模长：对向量的每个数字求平方在求和在开根号"""
    sum_square = 0
    for v in vec:
        sum_square += v * v

    # numpy sqrt函数完成开根号
    return np.sqrt(sum_square)


def cosine_similarity(vec_a, vec_b):
    """余弦相似度：2个向量的点积 除以 2个向量模长的乘积"""

    result = get_dot(vec_a, vec_b) / (get_norm(vec_a) * get_norm(vec_b))
    return result


if __name__ == '__main__':
    vec_a = [0.5, 0.5]
    vec_b = [0.7, 0.7]
    vec_c = [0.7, 0.5]
    vec_d = [-0.6, -0.5]

    print("ab:", cosine_similarity(vec_a, vec_b))
    print("ac:", cosine_similarity(vec_a, vec_c))
    print("ad:", cosine_similarity(vec_a, vec_d))

LangChain 基础：

介绍：

LangChain 由 Harrison Chase 创建于2022年10月，它是围绕 LLMs（大语言模型）建立的一个框架

LangChain 自身并不开发 LLMs，它的核心理念是为各种 LLMs 实现通用的接口，把 LLMs 相关的组件 "链接" 在一起，简化 LLMs 应用的开发难度，方便开发者快速地开发复杂的 LLMs 应用

官方文档：https://docs.langchain.com/

它是一个开发 LLM 相关业务功能的集大成者，是一个 Python 的第三方库，提供了各种功能的 API：提示词优化的相关功能 API，调用各类模型的功能 API，会话记忆的相关功能 API，各类文档管理分析的功能 API，各类功能链式执行的能力，构建 Agent 智能体的相关功能 API

环境部署：

pip install langchain langchain-community langchain-ollama langchain-chroma dashscope chromadb bs4 jq

模型调用：

调用大语言模型：

现在市面上各种各样的模型不断出现，LangChain 模型组件提供了与各种模型的集成，并为所有模型提供一个精简的统一接口

LangChain 目前支持三种类型的模型：LLMs（大语言模型）、Chat Models (聊天模型)、Embeddings Models (嵌入模型）

大语言模型：是技术范畴的统称，指基于大参数量、海量文本训练的 Transformer 架构模型，核心能力是理解和生成自然语言，主要服务于文本生成场景；

聊天模型：是应用范畴的细分，是专为对话场景优化的 LLMs，核心能力是模拟人类对话的轮次交互，主要服务于聊天场景

文本嵌入模型：文本嵌入模型接收文本作为输入, 得到文本的向量

LangChain 支持的三类模型，它们的使用场景不同，输入和输出不同，开发者需要根据项目需要选择相应

当前学习过程中所用的阿里云通义千问系列主要来自于：langchain_community 包

LLMs 使用场景最多，常用大模型的下载库：

https://huggingface.co/models

https://modelscope.cn/models

同时 LangChain 支持对许多模型的调用，以通义千问为例：

阿里云大语言模型的访问：

# langchain_community
from langchain_community.llms.tongyi import Tongyi

# qwen3-max是聊天模型，qwen-max是大语言模型
model = Tongyi(model="qwen-max")

# 调用invoke向模型提问
res = model.invoke(input="你是谁呀能做什么？")

print(res)

Ollama 本地大语言模型的访问：

# langchain_ollama
from langchain_ollama import OllamaLLM

model = OllamaLLM(model="qwen3.5:4b")

res = model.invoke(input="你是谁呀能做什么？")

print(res)

Models 的流式输出：

如果需要流式输出结果，需要将模型的 invoke 方法改为 stream 方法即可；

invoke 方法：一次性返回完整结果

stream 方法：逐段返回结果，流式输出

代码实现：

from langchain_community.llms.tongyi import Tongyi

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

# 通过stream方法获得流式输出
res = model.stream(input="你是谁呀能做什么？")
for chunk in res:
    print(chunk, end="", flush=True)

# from langchain_ollama import OllamaLLM
#
# model = OllamaLLM(model="qwen3.5:4b")
#
# res = model.stream(input="你是谁呀能做什么？")
#
# for chunk in res:
#     print(chunk, end="", flush=True)

调用聊天模型：

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

AIMessage：就是 AI 输出的消息，可以是针对问题的回答（OpenAI 库中的 assistant 角色） 

HumanMessage：人类消息就是用户信息，由人给出的信息发送给聊天模型的提示信息，比如 "实现一个快速排序方法" (OpenAI 库中的 user 角色）

SystemMessage：可以用于指定模型具体所处的环境和背景，如角色扮演等；可以在这里给出具体的指示，比如 "作为一个代码专家"，或者 "返回 json 格式"（OpenAI 库中的 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)

from langchain_ollama import ChatOllama
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

# 得到模型对象
model = ChatOllama(model="qwen3.5:4b")

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

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

消息的简写形式：

类对象写法：是最标准、最直接的方式；这种写法需要显式导入 SystemMessage、HumanMessa ge 和 AIMessage 等类，直接实例化对应的对象并传入内容；它是一种 "静态" 的直接定义方式，运行时无需额外转换，一步到位直接得到结构化的消息对象，逻辑清晰且类型明确

简写形式：这种写法更简洁便捷；它不需要导入具体的消息类，而是用二元元组 (角色名，内容) 的形式来表示消息，例如用 "system" 对应系统消息、"human" 对应用户消息、"ai" 对应 AI 消息；这是一种 "动态" 的写法，LangChain 会在运行时内部自动将这些元组转换为对应的消息类对象，省去了导包的步骤，代码更简洁

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)

简写形式的核心优势在于它对动态模板的天然支持；因为它是运行时动态转换的，所以可以直接在字符串内容中使用 {变量名} 作为占位符，例如 ("system", "今天的天气是{weather}")；在后续使用提示词模板时，就能更灵活地在运行时为这些变量填充具体值，实现动态化的对话构建

调用文本嵌入模型：

Embeddings Models 嵌入模型的特点：将字符串作为输入，返回一个浮点数的列表（向量）在NLP 中，Embedding 的作用就是将数据进行文本向量化

阿里云千问模型访问方式：

from langchain_community.embeddings import DashScopeEmbeddings

# 创建模型对象 不传model默认用的是 text-embeddings-v1
model = DashScopeEmbeddings()

# 不用 invoke stream
# embed_query、embed_documents
print(model.embed_query("我喜欢你"))
print(model.embed_documents(["我喜欢你","你喜欢我","不喜欢你"]))

本地 Ollama 模型访问方式：

通过 langchain_ollama 导入 OllamaEmbeddings 使用，其余不变

from langchain_ollama import OllamaEmbeddings


model = OllamaEmbeddings(model="qwen3.5-embedding:4b")

# 不用invoke stream
# embed_query、embed_documents
print(model.embed_query("我喜欢你"))
print(model.embed_documents(["我喜欢你", "你喜欢我", "不喜欢你"]))

目前所提及的 LangChain API 如下：

提示词模版：

PromptTemplate：

提示词优化在模型应用中非常重要，LangChain 提供了 PromptTemplate 类，用来协助优化提示词；

PromptTemplate 表示提示词模板，可以构建一个自定义的基础提示词模板，支持变量的注入，最终生成所需的提示词

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")
# 调用.format方法注入信息即可
prompt_text = prompt_template.format(lastname="张", gender="女儿")
# 调用模型获取结果
res = model.invoke(input=prompt_text)
print(res)

# 基于chain链的写法
# 生成链
# chain = prompt_template | model
# 基于链, 调用模型获取结果
# res = chain.invoke(input={"lastname": "张", "gender": "女儿"})
# print(res)

在大型项目中，使用模板能更方便地做提示词的标准化、统一化管理，让代码结构更清晰、更易维护，避免手动拼接带来的混乱和错误

PromptTemplate 这类模板类是 LangChain 框架的原生组件，天然支持框架的 Runnable 接口和链式调用，能无缝融入 LangChain 的工作流，这是手动字符串无法做到的

同时，LangChain 还提供了多种模板类型，除了基础的 PromptTemplate，还有 FewShotPromptT emplate、ChatPromptTemplate 等，可满足少样本学习、对话场景等不同的提示词构建需求

FewShotPromptTemplate：

FewShotPromptTemplate 是带示例的提示词模版，它会在给模型的提示里，预先放入几组「问题 + 标准答案」作为示例，让模型照着格式、逻辑、风格来生成结果，不用复杂指令就能让输出更规范、更准确，特别适合需要固定格式、分类、翻译等场景：

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

# 示例的模版
example_prompt = PromptTemplate.from_template("单词: {word}, 反义词: {antonym}")

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

few_shot_template = FewShotPromptTemplate(
    example_prompt=example_prompt, # 示例数据的模版
    examples=example_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))

案例练习：

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

product_template = PromptTemplate.from_template(
    "产品名称: {product_name}, 产品卖点: {product_selling_points}"
)

example_prompt = PromptTemplate.from_template(
    "输入: {input}"
    "输出: {output}"
)

example_data = [
    {
        "input": "MacBookPro高效节能, 性能强大, 适合牛马工作使用",
        "output": product_template.format(
            product_name="MacBookPro",
            product_selling_points="高效节能, 性能强大"
        )
    },
    {
        "input": "联想笔记本拥有RTX4060独立显卡, 畅玩游戏, 丝滑流畅",
        "output": product_template.format(
            product_name="联想笔记本",
            product_selling_points="畅玩游戏, 丝滑流畅"
        )
    }
]

few_shot_template = FewShotPromptTemplate(
    example_prompt=example_prompt,
    examples=example_data,
    prefix="请抽取产品名称和核心卖点2个字段, 格式为Json, 我提供2个示例:",
    suffix="请处理: {input_message}",
    input_variables=["input_message"]
)

prompt_text = few_shot_template.invoke(input={"input_message": "华为MatepadPro, 高清大屏, 长效续航, 你的好帮手"}).to_string()
print(prompt_text)

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

print(model.invoke(input=prompt_text))

模板类的 format 和 invoke 方法：

PromptTemplate、FewShotPromptTemplate、ChatPromptTemplate 都拥有 format 和 invoke 这两类方法：

format 和 invoke 的区别在于：

区别formatinvoke
功能Python 原生方法，只做纯字符串替换，解析占位符生成提示词LangChain 的 Runnable 接口标准方法，解析占位符生成提示词
返回值字符串PromptValue 类对象
传参.format(k=v, k=v, ...).invoke({"k":v, "k":v, ...})
解析支持解析 {} 占位符支持解析 {} 占位符和 MessagesPlaceholder 结构化占位符

from langchain_core.prompts import PromptTemplate
from langchain_core.prompts import FewShotPromptTemplate
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.llms.tongyi import Tongyi
from langchain_community.chat_models.tongyi import ChatTongyi

"""
PromptTemplate -> StringPromptTemplate -> BasePromptTemplate -> RunnableSerializable -> Runnable
FewShotPromptTemplate -> StringPromptTemplate -> BasePromptTemplate  -> RunnableSerializable -> Runnable
ChatPromptTemplate -> BaseChatPromptTemplate -> BasePromptTemplate  -> RunnableSerializable -> Runnable
Tongyi -> BaseLLM -> BaseLanguageModel -> RunnableSerializable -> Runnable
ChatTongyi -> BaseChatModel -> BaseLanguageModel -> RunnableSerializable -> Runnable
"""


template = PromptTemplate.from_template("我的邻居是：{lastname}，最喜欢：{hobby}")

res = template.format(lastname="老王", hobby="来我家")
print(res, type(res))


res2 = template.invoke({"lastname": "杰哥", "hobby": "来我家"})
print(res2, type(res2))

ChatPromptTemplate：

PromptTemplate：通用基础模板，用于单条提示词的动态变量注入

FewShotPromptTemplate：在模板中嵌入多组示例，实现少样本学习，引导模型按示例输出

ChatPromptTemplate：专为对话场景设计，支持注入多轮历史会话，通过 from_messages 方法加载列表形式的多轮对话；区别于 from_template 仅支持单条消息，它可处理完整的对话上下文

因为历史会话不是固定的，会随着对话轮次不断累积，因此必须支持动态注入，不能写死在模板里

# 导入依赖
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 构建对话模板：固定角色 + 历史占位 + 当前用户提问
chat_template = ChatPromptTemplate.from_messages([
    ("system", "系统提示词..."),  # 固定的系统角色
    ("ai", "AI历史回复..."),     # 可选的固定AI回复
    MessagesPlaceholder("history"),  # 历史对话占位符
    ("human", "用户当前提问...")  # 当前用户的问题
])

# 准备动态历史数据（多轮对话列表）
history_data = [
    ("human", "用户历史提问1"),
    ("ai", "AI历史回复1"),
    ("human", "用户历史提问2"),
    ("ai", "AI历史回复2")
]

# 动态注入历史，生成完整提示
prompt_value = chat_template.invoke({"history": history_data})

    MessagesPlaceholder("history")：专门的占位符，用于预留历史对话的位置，history 是后续注入时的键名

    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", "清明时节雨纷纷，路上行人梳中分，借问背带何处有，牧童遥指蔡徐村"),
]

prompt_text = chat_prompt_template.invoke({"history": history_data}).to_string()

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

res = model.invoke(prompt_text)

print(res.content, type(res))

    LCEL：

    基础使用：

    LCEL 是一种声明式、链式组合语法，通过 | 管道符 把组件（加载器、检索器、模型、输出解析器）按顺序串联，直接构建可运行的 RAG / 智能体流程，代码极简、可读性极强

    将组件串联，上一个组件的输出作为下一个组件的输入是 LangChain 链（尤其是 | 管道链）的核心工作原理，这也是链式调用的核心价值：实现数据的自动化流转与组件的协同工作，如下：

    chain = prompt_template | model

    前提：即 Runnable子类对象才能入链（以及 Callable、Mapping 接口子类对象也可加入）

    Chain 通过链接提示词模板与模型对象生成，返回的 chain 对象属于 RunnableSerializable 类型，它是 Runnable 接口的直接子类，也是绝大多数组件的父类；支持通过 invoke 做阻塞执行，或用 stream 做流式执行，且链内组件按顺序执行，上一个组件的输出会作为下一个组件的输入，实现流程串联

    执行流程：

    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")

chain  = chat_prompt_template | model

# res = chain.invoke({"history": history_data})
# print(res.content)

for chunk in chain.stream({"history": history_data}):
    print(chunk.content, end="", flush=True)

    管道运算符 |：

    Python 里 | 这类运算符的行为由类的魔法方法决定，比如 a | b 本质调用 a.__or__(b)，LangChain 通过重写 __or__ 方法，让 | 运算符实现了组件串联的功能，把提示词模板和模型等对象拼接成执行链；

    示例：重写 | 即 __or__ 方法，让 a | b | c 的代码得到一个自定义的类对象（类似列表即 [a，b，c] ），调用 run 方法依次输出 a，b，c

    class Test(object):
    def __init__(self, name):
        self.name = name

    def __or__(self, other):
        return MySequence(self, other)

    def __str__(self):
        return self.name


class MySequence(object):
    def __init__(self, *args):
        self.sequence = []
        for arg in args:
            self.sequence.append(arg)

    def __or__(self, other):
        self.sequence.append(other)
        return self

    def run(self):
        for i in self.sequence:
            print(i)


if __name__ == '__main__':
    a = Test('a')
    b = Test('b')
    c = Test('c')

    d = a | b | c # a.__or__(b)
    d.run()
    print(type(d))

    Runnable 接口：

    LangChain 中的绝大多数核心组件都继承了 Runnable 抽象基类（位于 langchain_core.runnables. base）

    chain = prompt | model

    chain 变量是 RunnableSequence（RunnableSerializable 子类）类型；而得到这个类型的原因就是 Runnable 基类内部对 __or__ 魔术方法的改写；同时，在后面继续使用|添加新的组件，依旧会得到 RunnableSequence，这就是链的基础架构

    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)) # <class 'langchain_core.runnables.base.RunnableSequence'>

    StrOutputParser：

    StrOutputParser 是 LangChain 用于将大模型对话消息对象解包为原生字符串的标准输出（解析器）解析组件，实现模型输出到文本字符串的轻量化、标准化转换

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

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

prompt = PromptTemplate.from_template(
    "我的邻居姓{lastname}, 刚生了{gender}, 你帮我起个名字, 仅告知名字无需其他内容"
)

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

chain = prompt | model | model
res = chain.invoke({"lastname": "张", "gender": "女儿"})

    运行报错（ValueError: Invalid input type <class 'langchain_core.messages.ai.AIMessage'>. Must be a PromptValue, str, or list of BaseMessages.）

    错误的主要原因是： prompt 的结果是 PromptValue 类型，输入给了 model，model 的输出结果是 AIMessage，而模型（ChatTongyi）源码中关于 invoke 方法明确指定了 input 的类型：

    所以需要借助 LangChain 内置的解析器 StrOutputParser 字符串输出解析器做类型转换

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

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

parser = StrOutputParser()

prompt = PromptTemplate.from_template(
    "我的邻居姓{lastname}, 刚生了{gender}, 你帮我起个名字, 仅告知名字无需其他内容"
)

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

chain = prompt | model | parser | model | parser

res: str = chain.invoke({"lastname": "张", "gender": "女儿"})
print(res)
print(type(res))

    JsonOutputParser：

    JsonOutputParser 是 LangChain 核心结构化输出解析器，基于 Runnable 接口实现，自动将大模型返回的文本 / 消息对象解析为标准 Python 字典（JSON 对象），完成非结构化文本到结构化数据的标准化转换，支持无缝接入 | 调用链，无需手动解析 JSON 字符串

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

    在初始输入 -> 提示词模板 -> 模型 -> 数据处理 -> 提示词模板 -> 模型 -> 解析器 -> 结果的执行链中，第一个模型输出是 AIMessage 对象，而第二个提示词模板要求输入字典格式，需要用 JsonOutputParser，把AIMessage 解析为 JSON 字典，才能注入第二个提示词模板，完成链路的格式衔接

    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()

# 第一个提示词模版
first_prompt = PromptTemplate.from_template(
    "我的邻居姓{lastname}, 刚生了{gender}, 你帮我起个名字, 并封装为 JSON 格式返回, 要求是 key 是 name, value 是起的名字, 请严格遵守格式要求"
)

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

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

# 构建链
chain = first_prompt | model | json_parser | second_prompt | model | str_parser

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

    构建 LangChain 链时，必须保证前后组件的输入输出格式匹配：

    模型可接收 PromptValue、字符串、序列等多种输入，输出固定为 AIMessage 对象；

    提示词模板要求输入为字典，输出为 PromptValue 对象；

    StrOutputParser 将 AIMessage 转为字符串，JsonOutputParser 将 AIMessage 转为字典，用于适配下游不同格式需求

    RunnableLambda：

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

    除了 JsonOutputParser 这类固定功能的解析器之外，也可以基于 RunnableLambda 类实现自己编写 Lambda 匿名函数来完成自定义逻辑的数据转换

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

    还可以直接用 lambda 匿名函数插入链中，无需手动封装 RunnableLambda 类（因为 Runnable 接口重写的 __or__ 方法，原生支持 Callable（可调用对象，函数 / lambda 都属于此类），会自动将函数转换为 RunnableLambda，实现模型输出到下一个提示词模板的格式适配，本质是语法糖简化了链的编写）

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

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

# 第一个提示词模版
first_prompt = PromptTemplate.from_template(
    "我的邻居姓{lastname}, 刚生了{gender}, 你帮我起个名字, 并封装为 JSON 格式返回, 要求是 key 是 name, value 是起的名字, 请严格遵守格式要求"
)

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

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

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

chain =  first_prompt | model | my_func | second_prompt | model | str_parser
# 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)

    记忆模块：

    临时会话记忆：

    在 LangChain 中，除了手动维护历史消息，还可借助内置的 History 功能让模型具备对话记忆：
    核心通过 RunnableWithMessageHistory，在原有无记忆链的基础上，封装出带历史记录功能的新 Runnable 链；
    搭配 InMemoryChatMessageHistory 可实现对话历史的内存存储，适合临时、快速的对话场景，无需手动管理历史数据

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

str_parser = StrOutputParser()

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

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你需要根据会话历史回应用户问题, 对话历史: "),
        MessagesPlaceholder("chat_history"),
        ("human", "请回答如下问题: {input}")
    ]
)

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

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": "小黑有两只猫"}, session_config)
    print("第一次执行: ", res)
    res = conversation_chain.invoke({"input": "小黑还有一只狗"}, session_config)
    print("第二次执行: ", res)
    res = conversation_chain.invoke({"input": "总共有几个宠物"}, session_config)
    print("第三次执行: ", res)

    长期会话记忆：

    使用 InMemoryChatMessageHistory 仅可以在内存中临时存储会话记忆，一旦程序退出，则记忆丢失；InMemoryChatMessageHistory 类继承自 BaseChatMessageHistory，在官方注释中给出了相关实现的指南，并给出了基于文件的历史消息存储示例代码

    FileChatMessageHistory 基于文件实现对话历史存储，核心是以 session_id 为文件名区分不同会话的记录文件；它通过继承 BaseChatMessageHistory，并重写 add_messages（同步添加消息）、messages（同步获取消息）、clear（同步清除消息）这三个方法，完成对话历史的持久化存储与管理

    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  # session_id 用来区分不同会话各自的历史记录
        self.storage_path = storage_path  # storage_path 表示所有会话历史统一存放的目录
        self.file_path = os.path.join(self.storage_path, self.session_id)  # file_path 是当前会话最终读写的文件路径
        os.makedirs(os.path.dirname(self.file_path), exist_ok=True)  # 目录不存在时先创建，避免后面写文件报错

    def add_messages(self, messages: Sequence[BaseMessage]) -> None:
        all_messages = self.messages  # self.messages 会先从文件中读出当前会话已有的消息列表
        all_messages.extend(messages)  # 把这次新增的消息追加到已有历史后面 List[消息, 消息...]
        new_messages = [message_to_dict(message) for message in all_messages]  # 先把消息对象转成普通字典，json 才能处理 List[字典, 字典...]
        with open(self.file_path, "w", encoding="utf-8") as f:
            json.dump(new_messages, f)  # json.dump 会把 Python 对象（字典）写成 JSON 文本保存到文件中

    @property
    def messages(self) -> list[BaseMessage]:
        try:
            with open(self.file_path, "r", encoding="utf-8") as f:
                messages_data = json.load(f)  # json.load 会把文件里的 JSON 文本解析成 Python 的 list[dict]
                return messages_from_dict(messages_data)  # 再把字典列表还原成 LangChain 的消息对象列表
        except FileNotFoundError:
            return []  # 文件不存在说明还没有历史记录，所以直接返回空列表

    def clear(self) -> None:
        with open(self.file_path, "w", encoding="utf-8") as f:
            json.dump([], f)  # 清空历史的本质就是把文件内容重置为空列表


str_parser = StrOutputParser()

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

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你需要根据会话历史回应用户问题, 对话历史: "),
        MessagesPlaceholder("chat_history"),
        ("human", "请回答如下问题: {input}")
    ]
)

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

base_chain = prompt | print_prompt | model | str_parser

store = {} # key是session,value是InMemoryChatMessageHistory类对象

# 实现通过会话id获取InMemoryChatMessageHistory类对象
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": "小黑有两只猫"}, session_config)
    print("第一次执行: ", res)
    res = conversation_chain.invoke({"input": "小黑还有一只狗"}, session_config)
    print("第二次执行: ", res)
    res = conversation_chain.invoke({"input": "总共有几个宠物"}, session_config)
    print("第三次执行: ", res)

    RAG 组件开发：

    文档加载：

    CSVLoader：

    文档加载器（Document Loaders）是 LangChain 生态中数据接入的核心入口，需实现 BaseLoader 接口，可将 CSV、PDF、JSON 等不同来源的数据，全部转换为 LangChain 统一的 Document 对象，为后续的检索增强生成（RAG）和智能体开发打下基础

    CSV（Comma-Separated Values，逗号分隔值） 是一种纯文本格式的表格数据文件，用逗号（或其他分隔符）来区分不同列，用换行区分不同行，是最通用、最基础的表格数据存储格式之一

    Document 类是 LangChain 的统一文档载体，所有加载器最终都返回该类实例，核心包含两部分：page_content 存储文档正文内容，metadata 存储来源、路径等元数据字典，实现不同数据源的一致性处理，如一个基础的 Document 类实例：

    from langchain_core.documents import Document

document = Document(
    page_content="Hello, world!", metadata={"source": "https://example.com"}
)

    所有文档加载器（无论数据源是 CSV、PDF 还是 JSON）都必须实现统一的标准接口，通过两种方法保证数据读取的一致性：
    load()：同步加载所有文档到内存，可能出现内存溢出问题，适合小文件
    lazy_load()：得到生成器对象，流式延迟加载，逐行 / 逐段返回数据，避免大文件占用过多内存

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

    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(type(document), document)

    JSONLoader：

    JSONLoader 是 LangChain 专门用于加载 JSON/JSONL 格式文件的文档加载器；它基于 jq（一款跨平台 JSON 解析工具） 解析语法 精准抽取文本数据，并自动封装为标准 Document 对象，数据抽取逻辑通过 jq_schema 语法定义，支持灵活提取嵌套字段、数组内容，是处理结构化 JSON 数据的核心工具

    pip install jq

    对于单个 JSON 对象（用.表示），使用 .字段名 抽取属性，用 .父字段.子字段 获取嵌套数据，通过 .字段名[索引] 或 .字段名.[索引] 访问数组元素

    对于 JSON 数组，使用 [ ] 指代整个数组集合，配合 .字段名 可批量抽取数组中所有对象的同名字段数据（.[]. 得到所有字典，.[].字段名 得到所有全部该字段名的信息）

    from langchain_community.document_loaders import JSONLoader

loader = JSONLoader(
    file_path="./data/stu_json_lines.json",
    jq_schema=".name",  # jq schema 语法
    text_content=False, # 抽取到的是否是字符串, 默认 True, 非必填
    json_lines=True     # 是否是JsonLines文件, 默认 False, 非必填
)

document = loader.load()
print(document)

    PyPDFLoader：

    LangChain 提供了多款 PDF 文档加载器，其中 PyPDFLoader 是常用选择；该加载器依赖 pypdf 库，需先通过 pip install pypdf 完成安装，即可快速加载 PDF 中的文本内容，将其转换为 LangChain 标准的 Document 对象

    from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader(
    file_path="./data/pdf1.pdf",
    mode="single", # 默认是page模式, 每个页面形成一个 Document 文档对象; single模式, 不管有多少页都只返回一个 Document 对象
    # password=""
)
i = 0
for doc in loader.lazy_load():
    i += 1
    print(doc)
    print("="*20, i)

    TextLoader：

    TextLoader 是 LangChain 中最基础的纯文本加载器，专门用于读取 .txt 等文本格式文件；它会自动读取文件内容，将文本封装成 LangChain 标准的 Document 对象，是最简单、最常用的文档加载工具；返回值是只有一个 Document 的 List，但如果文档很大，把它全塞进一个 Document 里显然不合适

    RecursiveCharacterTextSplitter 是 LangChain 官方推荐的文本分割器；它采用递归逻辑，按段落拆分大文档，能有效平衡分块大小与上下文完整性，是处理大文档的最优工具（基于文本的自然段落分割大文档为小文档，可以指定小文档的最大字符数、重叠字符数，段落划分的依据（符号）以及字符数量统计函数）

    pip install langchain_text_splitters

    from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

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)

    向量存储：

    Vector stores 向量存储：

    Vector stores（向量存储）是 LangChain RAG 流程的核心，通过嵌入模型将文档 / 查询转为向量存入数据库；LangChain 提供 InMemoryVectorStore（内存存储）、Chroma（外部数据库存储）等实现，所有存储类统一支持 add_documents（存向量）、delete（删向量）、similarity_search（相似检索）三大接口，是检索增强生成的核心支撑

    内存向量存储：

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

# 创建一个内存版向量库，并指定使用 DashScope 的向量模型
vector_store = InMemoryVectorStore(
    embedding=DashScopeEmbeddings()
)

# 创建 CSV 加载器，用来把 CSV 文件中的每一行读取成一个 Document 对象
loader = CSVLoader(
    file_path="./data/info.csv",   # CSV 文件路径
    encoding="utf-8",              # 文件编码
    source_column="source",        # 指定 metadata 中的 source 来自 CSV 的 source 列
)

# 加载 CSV 文件，返回值是一个 Document 列表
documents = loader.load()

# 把文档批量加入向量库，并为每条文档手动指定一个唯一 id
vector_store.add_documents(
    documents=documents,  # 要加入向量库的文档列表
    ids=["id" + str(i) for i in range(1, len(documents) + 1)],  # 生成 id1、id2、id3...
)

# 按 id 删除向量库中的文档
vector_store.delete(["id1", "id2"])

# 按语义相似度检索文档
# 第一个参数是查询文本，第二个参数是返回结果的数量
result = vector_store.similarity_search(
    "Python是不是简单易学呀",
    3
)

# 打印检索结果，返回类型是 list[Document]
print(result)

    外部向量持久化存储：

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

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

# 创建 CSV 加载器，用来把 CSV 文件中的每一行读取成一个 Document 对象
loader = CSVLoader(
    file_path="./data/info.csv",   # CSV 文件路径
    encoding="utf-8",              # 文件编码
    source_column="source",        # 指定 metadata 中的 source 来自 CSV 的 source 列
)

# 加载 CSV 文件，返回值是一个 Document 列表
documents = loader.load()

# 把文档批量加入向量库，并为每条文档手动指定一个唯一 id
vector_store.add_documents(
    documents=documents,  # 要加入向量库的文档列表
    ids=["id" + str(i) for i in range(1, len(documents) + 1)],  # 生成 id1、id2、id3...
)

# 按 id 删除向量库中的文档
vector_store.delete(["id1", "id2"])

# 按语义相似度检索文档
# 第一个参数是查询文本，第二个参数是返回结果的数量
result = vector_store.similarity_search(
    "Python是不是简单易学呀",
    3
)

# 打印检索结果，返回类型是 list[Document]
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

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

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

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 += "]"

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

chain = prompt | print_prompt | model | StrOutputParser()
# 将用户提问和匹配信息一同封装到提示词模板中提问模型
res = chain.invoke({"input": input_text, "context": reference_text})
print(res)

    检索器 Retriever：

    RunnablePassthrough 是 LangChain 中用于透传数据的核心组件，用来在链中保留并透传用户原始问题，让向量检索只负责取上下文，而问题本身不被覆盖；

    配合检索器使用时，可以同时把 用户问题和检索到的文档上下文 一起传给提示词模板，从而完整构建 RAG 所需的输入结构，就能让向量检索加入链，是实现 "检索 + 生成" 串联的关键枢纽

    """
提示词：用户的提问 + 向量库中检索到的参考资料
"""
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 = ChatPromptTemplate.from_messages(
    [
        ("system", "以我提供的已知参考资料为主, 简洁和专业的回答用户问题, 参考资料: {context}"),
        ("user", "用户提问: {input}")
    ]
)

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

vector_store.add_texts(
    ["减肥就是要像良子一样少吃多餐", "在减脂期间吃东西很重要,清淡少油控制卡路里摄入并运动起来", "跑步是很好的运动哦"]
)
input_text = "怎么减肥?"

# as_retriever() 能将向量存储对象快速转换为支持链式调用的 Retriever（检索器），返回符合 Runnable 接口标准的实例
retriever = vector_store.as_retriever(search_kwargs={"k":2}) # 指定最多返回 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

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

# 构建整条 LCEL 链
chain = (
    {
        # 第一条分支：把用户原始输入原封不动保留下来，作为 prompt 里的 {input}
        "input": RunnablePassthrough(),

        # 第二条分支：把用户输入交给 retriever 做相似检索，
        # 得到 list[Document] 后，再交给 format_func 格式化成字符串，
        # 最终作为 prompt 里的 {context}
        "context": retriever | format_func
    }
    # 上面这个“并行字典”执行结束后，会得到：
    # {"input": 用户问题, "context": 检索后的参考资料字符串}

    # 把 input 和 context 填充进提示词模板，生成最终 prompt
    | prompt

    # 打印最终 prompt，便于观察模型到底看到了什么
    | print_prompt

    # 把 prompt 发给大模型生成回答
    | model

    # 把模型输出从消息对象解析成普通字符串
    | StrOutputParser()
)

# 执行整条链
# 这里传入的是一个普通字符串，因为最左侧的 RunnablePassthrough 和 retriever
# 都能直接接收这个字符串作为输入
res = chain.invoke(input_text)
print(res)
    # 人工智能 # python
    Logo
    码道开发者社区

    码道开发者社区，聚焦华为云码道 CodeArts 代码智能体，沉淀 Agent、Skill、鸿蒙开发实战内容，供开发者查阅资料、交流技术、分享工程实践

    更多推荐

    cover

    不想先学一堆 React？用 Codex 帮你完成第一支 Remotion 动效视频

    avatar 码道开发者社区
    cover

    突发！英伟达限制员工使用Claude；程序员一条代码删掉89TB数据获刑：被判五年十个月；曝iPhone Duo有望支持微信双开 | 极客头条

    avatar 码道开发者社区
    cover

    每日热评｜学术牛马的救星还是幻觉放大器？47k星学术研究Agent技能包深度评测

    avatar 码道开发者社区