RAG基础
·
视频：https://www.bilibili.com/video/BV1yjz5BLEoY/?spm_id_from=333.1387.homepage.video_card.click
1.大模型接入
from openai import OpenAI
import os
"""
python
# 获取客户端对象
# 主要是用如下2个参数:
# api_key:模型服务商提供的APIKEY密钥
# base_url:模型服务商的API接入地址 主要基于此参数来切换不同的模型服务商(如OpenAI、阿里云、腾讯云等)
"""
client = OpenAI(
# 如果没有配置环境变量，请用阿里云百炼API Key替换：api_key="sk-xxx"
#api_key=os.getenv("ALIYUN_BAILIAN_API_KEY"),也可以以环境变量的形式存储在OPENAI_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)
2.openAI
2.1基础使用
from openai import OpenAI
import os
"""
OpenAI库是OpenAI公司发布的Python SDK，便与编程调用其产品，现许多模型服务商都兼容OpenAISDK的调用。
使用主要就3个流程：
• 创建客户端对象（OpenAI类对象）
• 和模型对话（client.chat.completions.create），可以提供3个使
•system：设定模型的行为和规则
•assistant：设定模型的回答，由用户设定
•user：用户的提问
•处理结果：response.choices[0].message.content
"""
# 1. 获取 client 对象，OpenAI 类对象
"""
# 获取客户端对象
# 主要是用如下2个参数:
# api_key:模型服务商提供的APIKEY密钥
# base_url:模型服务商的API接入地址 主要基于此参数来切换不同的模型服务商(如OpenAI、阿里云、腾讯云等)
"""
client = OpenAI(
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 2. 调用模型
"""
调用模型
client.chat.completions.create创建ChatCompletion对象
主要参数有2个:
model:选择所用模型，如代码的qwen3-max
messages:提供给模型的消息
类型:list，可以包含多个字典消息每个字典消息包含2个key
role:角色
system角色:设定助手的整体行为、角色和规则，为对话提供上下文框架(如指定助手身份、回答风格、核心要求)，是全局的背景设定，影响后续所有交互。
assistant角色:代表AI助手的回答，可以在代码中认为设定
user角色:代表用户，发送问题、指令或需求
content:内容
"""
response = client.chat.completions.create(
model="qwen3-max",
messages=[
{"role": "system", "content": "你是一个python编成专家，并且不说废话"},
{"role": "assistant", "content": "我是编程专家，请问有什么需要帮助的吗？"},
{"role": "user", "content": "请帮我写一个python代码，输出1-100之间的偶数"}
]
)
# 3. 处理结果
print(response.choices[0].message.content)2.2 OpenAI库的流式输出
"""
OpenAI库的流式输出
可以设定结果输出为stream模式（流式输出），获得更好的使用体验。开启流式输出主要就2步：
1.在client.chat.completions.create（）调用模型的时候设定参数：stream=True
2.for循环response对象，并在循环内输出内容
"""
from openai import OpenAI
import os
# 1. 获取 client 对象，OpenAI 类对象
client = OpenAI(
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 2. 调用模型
response = client.chat.completions.create(
model="qwen3-max",
messages=[
{"role": "system", "content": "你是一个python编成专家，并且话非常多"},
{"role": "assistant", "content": "我是编程专家，请问有什么需要帮助的吗？"},
{"role": "user", "content": "请帮我写一个python代码，输出1-100之间的偶数"}
],
stream=True # 开启流式输出
)
# 3. 处理结果
for chunk in response:
print(
chunk.choices[0].delta.content,
end="", # 每一段之间以空格为分隔
flush=True # 立即输出，不进行缓冲
)
2.3 OpenAI库附带历史消息调用模型
"""
OpenAI库附带历史消息调用模型
调模型传的参数messages，其要求是list对象，即表明其支持非常多的消息在内。我们可以基于此，将历史消息填，让模型知晓对话的上下，更好的回答
"""
from openai import OpenAI
#1.获取client对象，OpenAI类对象
client = OpenAI(
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
#2.调用模型
response = client.chat.completions.create(
model="qwen3-max",
messages=[
{"role":"system" , "content":"你是AI助理,回答很简洁"},
{"role":"user","content":"小明有2条宠物狗"},
{"role":"assistant","content":"好的"},
{"role":"user","content":"小红有3只宠物猫"},
{"role":"assistant","content":"好的"},
{"role":"user","content":"总共有几个宠物？"}
],
stream=True
)
#开启了流式输出的功能
#3.处理结果
# print(response.choices[0].message.content)
for chunk in response:
print(
chunk.choices[0].delta.content,
end="",
flush=True
)
#每一段之间以空格分隔# 立刻刷新缓冲区3.LLM 提示词
当前融领域信息化发展的时代,金融数据量激增,许多投资者和研究者试图通过对这些数据进深度分析而获得一些有效的决策和帮助，尽可能减少决策失误带来的损失。
所以，针对金融数据的分析方法研究是目前十分有益且热门的话题。
当前案例主要有三大业务场景实现：
基于大模型完成：金融文本分类
基于大模型完成：金融文本信息抽取
基于大模型完成：金融本匹配
大模型选择：Qwen在线大模型（阿里云通义千问qwen3-max)
采用方法：基于Few-Shot+Zero-Shot的思想，设计prompt（提示词），进而应用大模型完成相应的任务
3.1 Zero-shot学习
Zero-shotLearning 是指在训练阶段不存在与测试阶段完全相同的类别，但是模型可以使用训练过的知识来推广到测试集中的新类别上。
这种能被称为“零样本”学习，因为模型在训练时从未见过测试集中的新类别，在模型训练和提示词优化中均有体现。
在模型训练中：
已知马（四脚兽）、虎（有条纹）、熊猫（）的特征，但未训练过斑马的数据（不认识）
告知模型：斑马是四脚兽、有黑白的条纹
模型可以在已知数据中进推理，从识别斑马。
在提示词优化中：
Zero-shot思想用于基于已训练的能力，不提供任何示例，仅通过语去描述任务的要求、标和约束，让模型直接生成结果。
简单来说就是“用语言定义任务，解放（信任）模型的预训练知识"
比如:
请判断”"包围的用户评论中的情感倾向，输出正面或负面。”这款代餐鸡胸肉饱腹感很强，吃起来也不柴，很推荐！”
3.2 Few-shot学习
Few-shotLearning 是指少样本学习，当模型在学习了一定类别的量数据后，对于新的类别，只需要少量的样本就能快速学习，对应的有one-shot learning，单样本学习，也算样本少到为一的情况下的一种few-shot learning。
在模型训练中（相似度判断方法）：
基于少量企鹅样本并结合相识度判断，推论未知图内含“企鹅'
在提示词优化中：
• Few-shot主要于基于少量示例，让模型参考示例回答。
简单来说就是“例定义任务，在模型的预训练知识的基础上，提升模型回答的对齐精度（如参考例的格式）"
比如：
请抽取产品名称和核卖点2个字段，格式为Json，我提供2个示例。示例I：MacBookPro高效节能，性能强大，适合牛马工作使用
输出：{“产品名称”：“MacBookPro”，“产品卖点”：“高效节能，性能强大”】示例2：联想笔记本拥有RTX4060独立显卡，畅玩游戏，丝滑流畅
输出：{“产品名称”：“联想笔记本”，“产品卖点”：“畅玩游戏，丝滑流畅””请处理：华为MatepadPro，清屏，效续航，你的好帮。
总结：
在模型训练层面：
Zero-shot：零样本，基于模型训练阶段学习的属性/语义关联，去迁移到未知的新类别
Few-shot：少样本，基于少量样本，快速泛化识别新样本
在提示词优化层面：
Zero-shot：提示，语描述任务，依赖模型预训练知识回答
Few-shot：给与模型少量示例，引导模型对齐示例输出结果
3.3 提示词实战案例 ：LLM实现金融文本分类
掌握Fewshot式下prompt的设计式
掌握利用LLM实现文本分类的代码
LLM文本分类任务介绍
下面几段文本来自某平台发布的金融领域文本：
1.“今日，央行发布公告宣布降低利率，以刺激经济增长。这一降息举措将影响贷款利率，并在未来几个季度内对金融市场产生影响。”
2."ABC公司今日发布公告称，已成功完成对XYZ公司股权的收购交易。本次交易是ABC公司在扩大业务范围、加强市场竞争力方面的重要举措。据悉，此次收购将进一步巩固ABC公司在行业中的地位，并为未来业务发展提供更广阔的发展空间。详情请见公司官方网站公告栏”
3.“公司资产负债表显示，公司偿债能力强劲，现金流充足，为未来投资和扩张提供了坚实的财务基础。”
4.“最新的分析报告指出，可再生能源行业预计将在未来几年经历持续增长，投资者应该关注这一领域的投资机会”
我们的目的是期望模型能够帮助我们识别出这4段话中，每一句话描述的是一个什么类型的报告。
即期望的输出结果为：[新闻报道，公司公告，财务公告分析师报告]
3.4 Prompt设计
对于大模型来讲，prompt 的设计非常重要，一个 明确 的 prompt 能够帮助我们更好从大模型中获得我们想要的结果。在该任务的 prompt 设计中，我们主要考虑 2点：
• 需要向模型解释什么叫作「文本分类任务」
• 需要让模型按照我们指定的格式输出
为了让模型知道什么叫做「文本分类」，我们借用FewShot的方式，给模型展示一些正确的例子：
User："今日，股市经历了一轮震荡，受到宏观经济数据和全球贸易紧张局势的影响。投资者密切关注美联储可能的政策调整，以适应市场的不确定性。“是[新闻报道，公司公告，财务公告分析师报告门里的什么类别？
Bot:新闻报道
User："本公司年度财务报告显示，去年公司实现了稳步增长的盈利，同时资产负债表呈现强劲的状况。经济环境的稳定和管理层的有效战略执行为公司的健康发展奠定了基础。"是[新闻报道，公司公告，财务公告分析师报告里的什么类别？
Bot：财务报告
其中，User 代表我们输入给模型的句子，Bot 代表模型的回复内容。
注意：上述例子中Bot的部分也是由人工输入的，其目的是希望看到在看到类似User中的句子时，模型应当做出类似Bot的回答。
3.5 Json数据格式
"""
Json数据格式
JSON（JavaScript Object Notation)是一种轻量级的数据交换格式，易于人阅读和编写，同时也易于机器解析和生成。
Json是带有格式的字符串，主要用于数据交换，即程序和程序之间的信息互传，使用Json会更加方便，如下示例：
Text文本
• 非结构化
• 抽取信息不方便
1.周杰轮11岁是个男孩
2.蔡依临12岁是个女孩
3.小明16岁是一个男孩子
CSV（固定分隔符）文本
结构化
抽取信息方便
数据不含Schema，有一定风险
1.周杰轮，11，男
2.蔡依临，12，女
3.小明，16,男
Json文本
1.{
2. "name"："周杰轮"
3. "age":11,
4 "gender":"男"
5.}
6.{
7. "name"："蔡依临"
8. "age":12,
9. "gender":"女"
10.}
11.{
12. "name":"小明",
13. "age":16,
14. "gender":"男"
15.}
Json数据格式
Json主要有2种结构：Json对象和Json数组
Json对象：由键值对组成，键必须是字符串，值可以是字符串、数字、布尔值、数组、对象或null
key必须是字符串
value可以是：
• 数字
• 字符串
• 列表
• Json对象或Json数组
{
"key1": "value1",
"key2": 123,
"key3": true,
"key4": ["item1", "item2"],
"key5": {
"nestedKey": "nestedValue"
},
"key6": null
}
Json数组：由一系列值组成，值可以是字符串、数字、布尔值、数组、对象或null
结构：[{}, {}, {},...]
[
"item1",
123,
true,
["nestedItem1", "nestedItem2"],
{
"nestedKey": "nestedValue"
},
null
]
Json对象=>Python字典
Json数组=>Python列表内含多个字典
Json在Python中，就是字典和列表套字典的字符串表现形式。
Python中使用Json主要完成：
• 将Python字典、列表转换为json字符串
•读取json字符串，转换为Python字典或列表
主要使用Python内置的json库
•json.dumps(字典或列表，ensure_ascii=False)：将字典或列表转换为json字符串
•ensure_ascii参数确保中文能正常显示
•返回值：Json字符串
•json.loads(json字符串)：将json字符串转换为Python字典或列表
•返回值：Python字典或Python列表
"""
import json
# 将Python字典转换为Json字符串
data_dict = {
"name": "周杰轮",
"age": 11,
"gender": "男"
}
json_str = json.dumps(data_dict, ensure_ascii=False)
print(json_str)
#将Python字典转换为Json数组
data_list = [
{
"name": "周杰轮",
"age": 11,
"gender": "男"
},{
"name": "蔡依临",
"age": 12,
"gender": "女"
},{
"name": "小明",
"age": 16,
"gender": "男"
}
]
json_str = json.dumps(data_list, ensure_ascii=False)
print(json_str)
#将Json字符串转换为Python字典
json_str = '{"name": "周杰轮", "age": 11, "gender": "男"}'
data_dict = json.loads(json_str)
print(data_dict)
#将Json字符串转换为Python列表
json_str = '[{"name": "周杰轮", "age": 11, "gender": "男"}, {"name": "蔡依临", "age": 12, "gender": "女"}, {"name": "小明", "age": 16, "gender": "男"}]'
data_list = json.loads(json_str)
print(data_list)
4.基础知识合集
4.1 Langchain
LangChain是个开发LLM相关业务功能的集成者，是个Python的第三库，提供了各种功能的API。
提供：
提示词优化的相关功能API
调各类模型的功能API
会话记忆的相关功能API
各类档管理分析的功能API
构建Agent智能体的相关功能API
各类功能链式执行的能能力
LangChain是后续学习RAG开发的主框架
pip install Langchain langchain-community Langchain-ollama dashscope chromadb
•langchain：核心包
• langchain-community：社区持包，提供了更多的第三模型调（我们的阿云千问模型就需要这个包）
•langchain-ollama：ollama支持包，支持调用ollama托管部署的本地模型
•dashscope：阿云通义千问的Python SDK
• chromadb：轻量向量数据库（后续使用）4.2 RAG
通用的基础大模型存在一些问题：
• LLM的知识不是实时的，模型训练好后不具备自动更新知识的能力，会导致部分信息滞后
• LLM领域知识是缺乏的，大模型的知识来源于训练数据，这些数据主要来自公开的互联网和开源数据集，无法覆盖特定领域或高度专业化的内部知识
• 幻觉问题，LLM有时会在回答中生成看似合理但实际上是错误的
• 信息数据安全性
RAG（Retrieval-AugmentedGeneration）即检索增强生成，为大模型提供了从特定数据源检索到的信息，以此来修正和补充生成的答案。可以总结为一个公式：
RAG= 检索技术+LLM 提示
模型本质上就是用户输入，模型给出输出，用户能做的就是在输入上做功夫。
RAG就是在向模型提问之前基于已有的知识库或文档内容做检索，确保向模型提问的内容更精准以及包含足够的信息量用以提供给模型。
RAG标准流程由索引l（Indexing）、检索（Retriever）和生成（Generation）三个核心阶段组成。
•索引阶段，通过处理多种来源多种格式的文档提取其中文本，将其切分为标准长度的文本块（chunk），并进行嵌入向量化
(embedding），向量存储在向量数据库（vector database）中。
。加载文件
。内容提取
。文本分割，形成chunk文本
。向量化
。存向量数据库
•检索阶段，用户输入的查询（query）被转化为向量表示，通过相似度匹配从向量数据库中检索出最相关的本块。
。query向量化
。在文本向量中匹配出与问句向量相似的top_k个
• 生成阶段，检索到的相关本与原始查询共同构成提示词（Prompt），输语模型（LLM），成精确且具备上下关联的回答。
。匹配出的本作为上下和问题一起添加到prompt中
。提交给LLM生成答案：
RAG的核心价值：
解决知识实效性问题：大模型的训练数据有截时间，RAG可以接最新档（如公司财报、政策件），让模型输出“与时俱进”。
降低模型幻觉：模型的回答基于检索到的事实性资料，而非纯靠自身记忆，幅减少编造信息的概率。
需重新训练模型：相比微调（Fine-tuning），RAG只需更新知识库，成本更低、效率更。
4.3 向量库
RAG流程中，向量库是一个重要的节点。
离线流程：知识和信息 向量嵌入（向量化） 存入向量库
在线流程：用户的提问 向量嵌入（向量化） 在向量库中匹配
商量的基础概念
向量（Vector）就是文本的“数学身份证”：它把一段文字的语义信息，转换成一串固定长度的数字列表，让计算机能“看懂”文字的含义并做相似度计算。
简单来说，就是让计算机更方便的理解不同的文本内容，是否表述的是一个意思。
文本嵌入模型（如text-embedding-vl）通过深度学习等技术，从文本提取语义特征并映射为固定长度的数字序列。
如何更为精准的完成语义匹配，生成向量的维度是一个很重要的指标。
如text-embedding-v1模型，可以成1536维的向量（段本固定得到1536个数字序列），比较实。
•1536个数字表示，这段文本在1536个主题（抽象的语义特征）方向上的得分（强度）
• 生成向量的维度越多，就更好的记录本的语义特征，做语义匹配会更加精准。
• 更多的向量会在计算、存储和匹配过程中，带来更大的压力。
选择合适的向量维度需要在精确和性能之间做平衡。一般1536维算是比较好的选择。
向量（Vector）就是本的“数学份证”
它把段字的语义信息，转换成串固定度的数字列表，让计算机能“看懂”字的含义并做相似度计算。
向量的计算（文本嵌入过程），可借助文本嵌模型实现，如text-embedding-v1
向量的匹配通过算法实现，如余弦相似度
向量的维度表示段本在多个抽象语义特征的强度
• 维度数代表模型用多少个抽象语义特征来描述本
• 维度越多，做语义匹配越精准
• 但性能压力也会增
4.4 余旋相似度
"""
向量的数字序列，共同决定了向量在高维空间中的方向和长度.而余弦相似度主要就是撇除长度的影响，得到方向的夹角。夹角越小越相似，即方向相同。
余弦相似度主要匹配的就是：同向（所谓度）
我们能直接发现
[0.5，0.5]和[0.7，0.7]是同向不同长那计算机如何判定就依赖余弦相似度算法了。
在本向量语义配中，余弦相似度是衡量两个向量向相似程度的核算法，即判断两段本语义是否相近。
"""
import math
def cosine_similarity(vec1, vec2):
"""
手动计算两个向量的余弦相似度
:param vec1: 列表/数组形式的向量1
:param vec2: 列表/数组形式的向量2
:return: 余弦相似度值
"""
# 1. 计算点积
dot_product = sum(a * b for a, b in zip(vec1, vec2))
# 2. 计算向量模长
norm1 = math.sqrt(sum(a ** 2 for a in vec1))
norm2 = math.sqrt(sum(b ** 2 for b in vec2))
# 3. 避免除0错误
if norm1 == 0 or norm2 == 0:
return 0.0
# 4. 计算余弦相似度
return dot_product / (norm1 * norm2)
# 测试示例
if __name__ == "__main__":
v1 = [1, 2, 3]
v2 = [4, 5, 6]
v3 = [-1, -2, -3] # 与v1完全相反
print(f"v1和v2相似度: {cosine_similarity(v1, v2):.4f}")
print(f"v1和v3相似度: {cosine_similarity(v1, v3):.4f}")5.model
5.1 大预言模型model
现在市面上的模型多如牛毛，各种各样的模型不断出现，LangChain模型组件提供了与各种模型的集成，并为所有模型提供一个精简的统一接口。
LangChain目前支持三种类型的模型：LLMs（大语言模型）、ChatModels（聊天模型）、EmbeddingsModels（嵌入模型)。
• LLMs:是技术范畴的统称，指基于参数量、海量本训练的Transformer架构模型，核能是理解和成然语，主要服务于本成场景
• 聊天模型:是应范畴的细分，是专为对话场景优化的LLMs，核能是模拟类对话的轮次交互，主要服务于聊天场景
• 文本嵌入模型：文本嵌入模型接收文本作为输入，得到文本的向量。
LangChain支持的三类模型，它们的使场景不同，输和输出不同，开发者需要根据项需要选择相应。
我们所的阿云通义千问系列主要来自于：langchain_community包
LLMs(阿里云大语言模型的访问）
LLMs使场景最多，常模型的下载库：https://huggingface.co/modelshttps://modelscope.cn/models
同时LangChain支持对许多模型的调用，以通义千问为例：
from langchain_community.llms.tongyi import Tongyi### 实例化模型
llm = Tongyi(model='qwen-max')### 模型推理
res = llm.invoke("帮我讲个笑话吧")
print(res)
LLMs（ottama本地大语言模型的访问）
如果要访问本地ollama的模型，简单更改一下代码。
通过langchain_ollama包导ollamaLLM类即可（请确保0llama已经启动并提前下载好要使的模型）。
from langchain_ollama import OllamaLLM
model = OllamaLLM (model="qwen3:4b")
#通过invoke方法去调用模型
res = model.invoke(input="你是谁呀能做什么？")
print (res)
通过：
from langchain_community.llm.tongyiimport Tongyi导入通义千问系列的支持
from langchain_ollama import 0ttamaLLM导0ttama系列的支持
创建好模型对象后，通过invoke对模型发起提问并可以直接打印输出结果
from langchain_community.llms.tongyi import Tongyi
### 实例化模型
llm = Tongyi(model='qwen-max')
### 调用Invoke方法像模型提问
res = llm.invoke("帮我讲个笑话吧")
print(res)5.2 model的流式输出
"""
如果需要流式输出结果，需要将模型的invoke法改为stream法即可。
• invoke法：次型返回完整结果
• stream法：逐段返回结果，流式输出
这两个方法是新版LangChain(1.0版本后）中基于Runnable接口的通用核心方法。
绝多数组件(如提示词模板、链、向量检索、具调等，后续学习）都支持这两个方法，这也是LangChain设计的核统范式。
"""
from langchain_community.llms.tongyi import Tongyi
### 实例化模型
llm = Tongyi(model='qwen-max')
### 调用Stream方法像模型提问
res = llm.stream("你是谁，你能做什么")
for chunk in res:
print(chunk, end="", flush=True)5.3聊天模型
"""
聊天模型
聊天消息包含下种类型，使时需要按照约定传合适的值：
• AIMessage：就是 AI输出的消息，可以是针对问题的回答。(OpenAI库中的assistant)
• HumanMessage： 类消息就是用户信息，由人给出的信息发送给LLMs的提信息，如“实现1个快速排序法”。（OpenAI库中的user角色）
• SystemMessage： 可以用于指定模型具体所处的环境和背景，如角色扮演等。你可以在这给出具体的指示，比如"作为一个代码专家"，或者"返回json格式"。（OpenAI库中的system角色）
演示单独使用HumanMessage
from Langchain_community.chat_models.tongyi import ChatTongyi
from Langchain_core.messages import HumanMessage
#初始化模型
chat = ChatTongyi(model="qwen3-max")
#准备消息listmessages = [
HumanMessage(content="给我写一首唐诗")
1
#流式输出
for chunk in chat.stream(input=messages):print(chunk.content,end="",flush=True)
"""
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.messages import HumanMessage , SystemMessage , AIMessage
#初始化模型
chat = ChatTongyi(model="qwen3-max")
#准备消息list
messages = [
SystemMessage(content="你是一个诗人，擅长写唐诗"),
AIMessage(content="模仿白居易"),
HumanMessage(content="给我写一首唐诗")
]
#流式输出
for chunk in chat.stream(input=messages):
print(chunk.content,end="",flush=True)5.4 从写法上简化chatmodel的使用
"""
从写法上简化chatmodel的使用
通过2元元组封装信息；
• 第一个元素为角色
字符串：system/human/ai
• 第2个元素为内容
---
### 方式一(原始写法)：使用 Message 类对象的写法
这种写法是直接调用 LangChain 提供的消息类（如 `SystemMessage`、`HumanMessage`、`AIMessage`）来构建对话列表。
它是**静态的**，在代码运行前就已经是完整的 `Message` 对象了，一步到位，不需要额外转换。
优点是类型清晰、代码规范；缺点是需要手动导入这些类，并且不支持直接写变量占位符。
**示例代码：**
```python
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
# 用类对象方式定义对话
messages = [
SystemMessage(content="你是一个乐于助人的AI助手。"),
HumanMessage(content="你好，今天天气怎么样？"),
AIMessage(content="你好！作为AI我无法感知天气，但可以帮你查询相关信息。")
]
```
---
### 方式二：使用元组的简写写法
这种写法用 `(角色, 内容)` 的元组来表示对话，比如 `("system", "...")`、`("human", "...")`、`("ai", "...")`。
它是**动态的**，运行时才会由 LangChain 内部机制自动转换成对应的 `Message` 对象。
优点是写法更简洁，不需要导入额外的类；最重要的是，它支持直接在内容里写 `{变量名}` 占位符，能在运行时动态填充值，这是提示词模板的基础。
**示例代码：**
```python
# 用元组简写方式定义对话
messages = [
("system", "你是一个乐于助人的AI助手。"),
("human", "你好，今天天气怎么样？"),
("ai", "你好！作为AI我无法感知天气，但可以帮你查询相关信息。")
]
```
**支持变量占位符的进阶示例：**
```python
# 简写方式的核心优势：支持 {变量} 占位符
messages = [
("system", "你是{role}，说话风格像{style}。"),
("human", "我的名字是{name}，今天的天气是{weather}，请和我打个招呼。")
]
# 这些变量后续可以通过 LangChain 的提示词模板在运行时填充具体值
```
---
简单来说，两种方式最终效果是一样的，都是生成 LangChain 能识别的对话消息列表。区别在于：
- 类对象写法更“显式”，适合静态对话；
- 元组简写写法更“隐式”，更简洁，并且天生支持动态变量，是后续学习提示词工程的基础。
"""
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.messages import HumanMessage , SystemMessage , AIMessage
#初始化模型
chat = ChatTongyi(model="qwen3-max")
#准备消息list
#（角色，内容） 角色：System/human/ai
messages = [
("system", "你是一个诗人，擅长写唐诗"),
("ai", "模仿宋代苏轼,只输出唐诗"),
("human", "给我写一首唐诗")
]
#流式输出
for chunk in chat.stream(input=messages):
print(chunk.content,end="",flush=True)5.5 EmbeddingsModets(文本嵌入模型）
"""
EmbeddingsModets(文本嵌入模型）
EmbeddingsModels嵌入模型的特点：将字符串作为输入，返回1个浮点数的列表（向量）
在NLP中，Embedding的作用就是将数据进行文本向量化。
阿云千问模型访问式：
from Langchain_community.embeddings import DashScopeEmbeddings
#DashScopeEmbeddings是阿云千问系列的嵌入模型类，提供了对阿云千问系列嵌入模型的访问支持。
#初始化嵌入模型对象，其默认使用模型是：text-embedding-v1
embed = DashScopeEmbeddings()
#测试
print(embed.embed_query("我喜欢你"))
#embed_query方法是DashScopeEmbeddings类中的一个方法，用于将输入的字符串进行嵌入向量化，返回一个浮点数列表（向量）。这个方法通常用于将单个查询文本转换为向量表示，以便在后续的相似度计算或其他任务中使用。
print(embed.embed_documents（['我喜欢你'，'我稀饭你'，'晚上吃啥'])）
#embed_documents方法也是DashScopeEmbeddings类中的一个方法，用于将输入的字符串列表(批量)进行嵌入向量化，返回一个包含多个浮点数列表（向量）的列表。这个方法通常用于将多个文档文本转换为向量表示，以便在后续的相似度计算或其他任务中使用。
"""
from langchain_community.embeddings import DashScopeEmbeddings
#初始化嵌入模型对象，其默认使用模型是：text-embedding-v1
embed = DashScopeEmbeddings()
#测试
print(embed.embed_query("我喜欢你")[:10])
#[:10]限制输出的维度，例如前10个元素
print(embed.embed_documents(['我喜欢你', '我稀饭你', '晚上吃啥']))5.6 model总结
1. 阿里云千问 LLM（大语言模型）
- **导入方式**：
from langchain_community.llms.tongyi import Tongyi
- **调用方法**：
- `invoke()`：批量调用，一次性获取完整回复
- `stream()`：流式调用，逐字/逐段输出内容
2. 阿里云千问 聊天模型（Chat Model）
- **导入方式**：
from langchain_community.chat_models.tongyi import ChatTongyi
- **调用方法**：
- `invoke()`：批量调用，一次性获取完整回复
- `stream()`：流式调用，逐字/逐段输出内容
3. 阿里云千问 文本嵌入模型（Embedding Model）
- **导入方式**：
from langchain_community.embeddings import DashScopeEmbeddings
- **调用方法**：
- `embed_query()`：单次文本向量化（用于用户查询）
- `embed_documents()`：批量文本向量化（用于文档/语料库）
- 前两种模型（LLM、聊天模型）都支持 `invoke/stream`，但**聊天模型**是更常用的形式，支持多轮对话和消息列表格式。
- 嵌入模型的两个方法分工明确：`embed_query` 专门处理用户的查询语句，`embed_documents` 用来批量处理大量文档文本。
6.prompt模板
6.1 通用prompt(zero-shot)
"""
通用prompt(zero-shot)
提示词优化在模型应用中非常重要，LangChain提供了PromptTemplate类，用来协助优化提示词。
PromptTemplate表示提示词模板，可以构建一个自定义的基础提示词模板，支持变量的注入，最终生成所需的提示词。
标准写法
from langchain_core.prompts import PromptTemplate
from langchain_community.llms.tongyi import Tongyi
prompt_template = PromptTemplate.from_template(“我的邻居姓{lastname}，刚生了{gender}，帮忙起名字，请简略回答。")
#变量注入，生成提示词文本
prompt_text = prompt_template.format (lastname="张", gender="女L")
model = Tongyi (model="qwen-max")#创建模型对象
res=model.invoke(input=prompt_text)# 调用模型获取结果
print (res)
基于chain链的写法
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":"女L"})) #通过链的invoke方法获取结果，输入参数是一个字典，键是提示词模板中的变量名，值是要注入的具体内容
print(res)
基于PromptTemplate类可以得到提示词模板，持基于模板注变量得到最终提示词。
• zero-shot思想下，可以基于PromptTemplate直接完成。
• few-shot思想下，需要更换为FewShotPromptTemplate。(后续学习)
使PromptTemplate还不如动拼接字符串？
1. 代码更清晰，提示词模板和变量分离，结构更清晰。
2. 支持变量注入，避免了字符串拼接的繁琐和错误。
3. 基于链的写法更符合LangChain的设计范式，后续学习中会大量使用链式调用。
4. 后续学习中会介绍更多基于PromptTemplate的高级用法，如FewShotPromptTemplate、ChatPromptTemplate等，提前熟悉PromptTemplate是基础。
使Template模板构建提示词，在型程中更容易做标准化模板，后续也更容易迁移到FewShotPromptTemplate、ChatPromptTemplate等更复杂的提示词模板类。
"""
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="女L")
res1=model.invoke(input=prompt_text)# 调用模型获取结果
print (res1)
chain = prompt_template | model #生成链，链的输入是提示词模板，输出是模型的结果
#基于链，调用模型获取结果
res2=chain.invoke(input={"lastname":"曹","gender":"女L"}) #通过链的invoke方法获取结果，输入参数是一个字典，键是提示词模板中的变量名，值是要注入的具体内容
print(res2)
6.2 FewShotPromptTemplate
"""
FewShotPromptTemplate
from Langchain_core.prompts import FewShotPromptTemplate
FewShotPromptTempLate(
examples=None,
example_prompt=None,
prefix=None,
suffix=None,
input_variables=None
)
参数：
•examples：示例数据，list，内套字典
• example_prompt：示例数据的提示词模板
• prefix：组装提示词，示例数据前内容
• suffix：组装提示词，示例数据后内容
• input_variables：列表，注入的变量列表
组装FewShotPromptTemplate对象井获得最终提示词
"""
from langchain_core.prompts import FewShotPromptTemplate, PromptTemplate
# 提示词示例
example_template = PromptTemplate.from_template("单词:{word}，反义词:{antonym}")
# 数据示例，list内套字典
example_data = [
{"word": "大", "antonym": "小"},
{"word": "上", "antonym": "下"}
]
# FewShot提示词模板对象
few_shot_prompt = FewShotPromptTemplate(
example_prompt=example_template,
examples=example_data,
prefix="给出给定词的反义词，有如下示例：",
suffix="基于示例告诉我：{input_word}的反义词是？",
input_variables=['input_word']
)
# 获得最终提示词
prompt_text = few_shot_prompt.invoke(input={"input_word": "左"}).to_string()
print(prompt_text)6.3 ChatPromptTemplate
"""
ChatPromptTemplate
PromptTemplate：通用提示词模板，支持动态注入信息。
FewShotPromptTemplate：支持基于模板注入任意数量的示例信息。
ChatPromptTemplate：支持注入任意数量的历史会话信息。
• 通过from_messages方法，从列表中获取多轮次会话作为聊天的基础模板
• PS:前面PromptTemplate类用的from_template仅能接入一条消息，而from_messages可以接入一个list的消息
历史会话信息并不是静态的（固定的），是随着对话的进不停地积攒，即动态的。
所以，历史会话信息需要持动态注。
•MessagePlaceholder作为占位
•提供history作为占位的key
•基于invoke动态注历史会话记录
必须是invoke，format无法注入动态历史会话记录
"""
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.chat_models.tongyi import ChatTongyi
chat_template=ChatPromptTemplate.from_messages(
[
("system","你是我的人工智能助手，协助我解答问题。"),
MessagesPlaceholder(variable_name="history"),#历史会话占位
("user","我的邻居姓{lastname}"),
("assistant","好的，请问他刚生了什么？"),
("user","刚生了{gender}，帮忙起名字，请简略回答。"),
("assistant","好的，我会帮你起名字。")
]
)
history_data=[
("user","我想让你帮我起名字。"),
("assistant","好的，请问你邻居姓什么？")
]
prompt_text=chat_template.invoke(input={"lastname":"张","gender":"女L","history":history_data}).to_string()
api_key = "sk-f9171bc4ee51428cada2834833a24ad8"
model=ChatTongyi(model="qwen3-max")
res=model.invoke(prompt_text)
print(res.content)
6.4 chain链的使用
"""
「将组件串联，上一个组件的输出作为下一个组件的输入」是LangChain链（尤其是|管道链）的核心工作原理，这也是链式调的核价值：实现数据的动化流转与组件的协同作，如下。
chain = prompt_template I model
核心前提：即Runnable子类对象才能入链（以及Callable、Mapping接口子类对象也可加入（后续了解用的不多）)。我们前所学习到的组件，均是Runnabe接的类，如下类的继承关系：
• 通过|链接提示词模板对象和模型对象
。返回值chain对象是RunnableSerializable对象
•是Runnable接口的直接子类
• 也是绝大多数组件的父类
•通过invoke或stream进阻塞执或流式执
组成的链在执上有：上个组件的输出作为下个组件的输入的特性。
所以有如下执行流程：
"""
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.runnables.base import RunnableSerializable
chat_template=ChatPromptTemplate.from_messages(
[
("system","你是我的人工智能助手，协助我解答问题。"),
MessagesPlaceholder(variable_name="history"),#历史会话占位
("user","我的邻居姓{lastname}"),
("assistant","好的，请问他刚生了什么？"),
("user","刚生了{gender}，帮忙起名字，请简略回答。"),
("assistant","好的，我会帮你起名字。")
]
)
history_data=[
("user","我想让你帮我起名字。"),
("assistant","好的，请问你邻居姓什么？")
]
model=ChatTongyi(model="qwen3-max")
chain=RunnableSerializable=chat_template|model
print(type(chain))
res=chain.invoke(input={"lastname":"张","gender":"女L","history":history_data})
print(res.content)
for chunk in chain.stream(input={"lastname":"张","gender":"女","history":history_data}):
print(chunk.content,end="",flush=True)7.字符串重载
7.1 运算符重载
前文代码中： chain = chat prompt template | model在语法上使用了运算符的重写
在Python中，运算符（如+、I）的行为由类的魔法方法决定。例如：
•a+b本质调用的是a.__add__(b)
•a|b本质调用的是a.__or__(b)
只需要自行实现类的__or__方法，即可对|符号的功能进行重写。
示例：
• 让a|b|c的代码得到一个自定义的类对象(类似列表即[a，b，c])调用run方法依次输出a、b、c
•我们需要重写|即__or__方法
7.2 StrOutputParser
"""
StrOutputParser是LangChain内置的简单字符串解析器
可以将AIMessage解析为简单的字符串，符合了模型invoke方法要求（可传入字符串，不接收AIMessage类型）
• 是Runnable接口的子类（可以加入链）
"""
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")
parser=StrOutputParser()
prompt=PromptTemplate.from_template(
"我邻居姓{last_name},请给他的女儿起个名字。仅告知姓名，无需其他内容"
)
chain=prompt | model | parser | model
res = chain.invoke({"last_name":"张"})
print(res.content)7.3 JsonOutputParser
"""
在构建链的时候要注意整体兼容性，注意前后组件的输和输出要求。
模型输入：PromptValue或字符串或序列(BaseMessage、list、tuple、str、dict)。模型输出：AIMessage
提词模板输：要求是字典
提示词模板输出：PromptValue对象
StrOutputParser：AIMessage输入、str输出JsonOutputParser:AIMessage输入、dict输出
"""
from langchain_core.output_parsers import JsonOutputParser,StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_community.chat_models.tongyi import ChatTongyi
str_parser=StrOutputParser()
json_parser=JsonOutputParser()
model=ChatTongyi(model="qwen3-max")
# 修改提示词，明确要求返回 JSON 格式的结果
first_prompt=PromptTemplate.from_template(
"我邻居姓{last_name}，请给他的女儿起个名字。"
"请以 JSON 格式返回结果， 格式如下：{{\"name\": \"名字\"}}。仅告知姓名，无需其他内容。"
)
sec_prompt=PromptTemplate.from_template(
"姓名：{name},的含义"
)
chain=first_prompt | model | json_parser |sec_prompt | model | str_parser
res= chain.invoke({"last_name":"张"})
print(res)7.4 RunnableLambda
""""
前我们根据JsonOutputParser完成了多模型执链条的构建。
• 除了JsonOutputParser这类固定功能的解析器之外
•我们也可以编写Lambda匿名函数来完成定义逻辑的数据转换，想怎么转换就怎么转换，更。想要完成这个功能，可以基于RunnableLambda类实现。
RunnableLambda类是LangChain内置的，将普通函数等转换为Runnable接口实例，便定义函数加chai
语法：
RunnableLambda（函数对象或lambda匿名函数）
如果像要在链中加定义函数，可以选择：
将函数封装入RunnableLambda类对象，其是Runnable接实例，可以直接入链
直接将函数入链，函数会自动转换为RunnableLambda对象
"""
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.runnables import RunnableLambda
str_parser=StrOutputParser()
# 定义一个Lambda函数，提取AIMessage中的姓名信息，并返回一个字典
#数据转入到ai_mes，输出一个字典，键为"name"，值为从AIMessage中提取的姓名信息
my_fun=RunnableLambda(lambda ai_mes:{"name": ai_mes.content})
model=ChatTongyi(model="qwen3-max")
# 修改提示词，明确要求返回 JSON 格式的结果
first_prompt=PromptTemplate.from_template(
"我邻居姓{last_name}，请给他的女儿起个名字，只给出名字，不要有其他信息。"
)
sec_prompt=PromptTemplate.from_template(
"姓名：{name},的含义"
)
chain=first_prompt | model | my_fun |sec_prompt | model | str_parser
res= chain.invoke({"last_name":"文"})
print(res)8.记忆功能 memory
8.1 短期记忆
"""
如果想要封装历史记录，除了维护历史消息外，也可以借助LangChain内置的历史记录附加功能。
LangChain提供了History功能，帮助模型在有历史记忆的情况下回答。
基于RunnableWithMesSageHistory在原有链的基础上创建带有历史记录功能的新链（新Runnable实例）
基于InMemoryChatMessageHistory为历史记录提供内存存储（临时）
RunnableWithMessageHistory是LangChain内Runnable接的实现，主要于：
创建一个带有历史记忆功能的Runnable实例（链）
它在创建的时候需要提供个BaseChatMessageHistory的具体实现（用来存储历史消息）
·
InMemoryChatMessageHistory可以实现在内存中存储历史
额外的，如果想要在invoke或stream执链的同时，将提词print出来，可以在链中加定义函数实现。
注意：函数的输应原封不动返回出去，避免破坏原有业务，仅在return之前，print所需信息即可。
"""
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_community.chat_models.tongyi import ChatTongyi
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} ，请回答"
)
str_parser=StrOutputParser()
base_chain=prompt | model | str_parser
#添加功能：历史消息附加功能
# 存储会话历史的字典
store = {}
# 通过会话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, # 被增强的原有链
get_history, # 通过会话ID获取InMemoryChatMessageHistory对象
input_messages_key="input", # 表示用户输入在模板中的占位符
history_messages_key="chat_history" # 表示历史消息在模板中的占位符
)
# 模拟多轮对话
if __name__ == "__main__":
session_id = "session_1"
# 第一次对话
res1 = conversation_chain.invoke(
{"input": "你是谁？"}, # 用户输入
{"configurable": {"session_id": session_id}} # 配置参数，包含 session_id
)
print("AI:", res1)
# 第二次对话
res2 = conversation_chain.invoke(
{"input": "你能做什么？"}, # 用户输入
{"configurable": {"session_id": session_id}} # 配置参数，包含 session_id
)
print("AI:", res2)8.2 长期记忆
""""
FileChatMessageHistory类实现，核心思路：
• 基于文件存储会话记录，以session_id为文件名，不同session_id有不同文件存储消息
继承BaseChatMessageHistory实现如下3个方法：
add_messages:同步模式，添加消息
messages:同步模式，获取消息
clear：同步模式，清除消息
如下侧代码，官在BaseChatMessageHistory类的注释中提供了一个基于文件存储的示例代码。
多会话隔离（session_id）
每个会话对应一个独立的 JSON 文件，文件名为 {session_id}.json，不同用户 / 会话之间互不干扰。
继承 BaseChatMessageHistory必须实现 LangChain 要求的三个核心同步接口：
@property messages：获取当前会话的所有消息
add_messages()：添加消息并持久化
clear()：清空会话并删除文件
LangChain 消息格式兼容
message_to_dict(msg)：将 LangChain 的 BaseMessage（HumanMessage/AIMessage 等）转为可序列化的字典
messages_from_dict(list_of_dict)：反序列化字典列表为 LangChain 消息对象，直接适配 LangChain 的链与模型
持久化与异常处理
自动创建存储目录，避免路径不存在报错
处理文件损坏、JSON 解析错误，保证程序稳定运行
每次读写前自动加载最新数据，避免多线程 / 并发场景下的数据不一致问题
"""
import os
import json
from langchain_core.messages import message_to_dict, messages_from_dict
from langchain_core.chat_history import BaseChatMessageHistory
class FileChatMessageHistory(BaseChatMessageHistory):
"""
基于文件的会话记忆存储，继承 LangChain 的 BaseChatMessageHistory
每个 session_id 对应一个独立的 JSON 文件，实现多会话隔离
"""
def __init__(self, session_id: str, storage_path: str):
self.session_id = session_id # 会话唯一标识
self.storage_path = storage_path # 存储目录路径
# 确保存储目录存在
os.makedirs(self.storage_path, exist_ok=True)
# 消息文件的完整路径：storage_path/session_id.json
self.file_path = os.path.join(self.storage_path, f"{self.session_id}.json")
# 加载历史消息到内存
self._load_messages()
def _load_messages(self) -> None:
"""从文件加载消息到内存"""
if os.path.exists(self.file_path):
try:
with open(self.file_path, "r", encoding="utf-8") as f:
data = json.load(f)
# 字典列表 -> LangChain 消息对象列表
self._messages = messages_from_dict(data)
except (json.JSONDecodeError, KeyError):
# 文件损坏/格式错误时重置为空
self._messages = []
else:
self._messages = []
def _save_messages(self) -> None:
"""将内存中的消息保存到文件"""
# LangChain 消息对象 -> 字典列表
data = [message_to_dict(msg) for msg in self._messages]
with open(self.file_path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
@property
def messages(self):
"""
同步获取所有消息（LangChain 要求的属性）
访问时自动从文件加载最新数据
"""
self._load_messages()
return self._messages
def add_messages(self, messages) -> None:
"""
同步添加消息（LangChain 要求的方法）
每次添加后自动保存到文件
"""
self._load_messages()
self._messages.extend(messages)
self._save_messages()
def clear(self) -> None:
"""
同步清空会话消息（LangChain 要求的方法）
同时删除文件中的记录
"""
self._messages = []
if os.path.exists(self.file_path):
os.remove(self.file_path)
from langchain_core.messages import HumanMessage, AIMessage
# 初始化存储目录
history_store_path = "./chat_history"
# 创建会话记忆（会话ID可自定义，比如用户ID+对话ID）
history = FileChatMessageHistory(
session_id="user_123_session_456",
storage_path=history_store_path
)
# 添加消息
history.add_messages([
HumanMessage(content="你好，介绍一下你自己"),
AIMessage(content="我是一个大语言模型，能帮你解决各种问题~")
])
# 获取所有消息
print("会话历史：")
for msg in history.messages:
print(f"{msg.type}: {msg.content}")
# 清空会话（会删除对应的JSON文件）
# history.clear()9.文档加载器
文档加载器提供了一套标准接口，用于将不同来源（如CSV、PDF或JSON等）的数据读取为LangChain的文档格式。这确保了无论数据来源如何，都能对其进行一致性处理。
文档加载器（内置或自行实现）需实现BaseLoader接口。
ClassDocument，是LangChain内文档的统一载体，所有文档加载器最终返回此类的实例。
不同的文档加载器可能定义了不同的参数，但是其都实现了统一的接口（方法）
•load()：一次性加载全部文档
• lazy_load()：延迟流式传输文档，对大型数据集很有用，避免内存溢出
LangChain内置了许多文档加载器，详细参见官方文档：
https://docs.langchain.com/oss/python/integrations/document loaders我们简单的学习如下几个常用的文档加载器：
CSVLoader
JSoNLoader
PDFLoader
9.1 CSVLoader、
"""
CSVLoader于加载CSV件，加载成功得到的即Document对象。
常见坑与解决方法
中文乱码：加上 encoding="utf-8" 或 encoding="gbk"。
无表头 CSV 读成数据：必须加上 fieldnames，否则第一行会被当成数据行。
分隔符不是逗号：比如 .tsv 文件，要设置 delimiter="\t"。
字段中包含逗号：要确保字段被引号包裹，并设置 quotechar='"'。
"""
from langchain_community.document_loaders.csv_loader import CSVLoader
# 自定义分隔符、引号、字段名等
loader = CSVLoader(
file_path="./xxx.csv",
csv_args={
"delimiter": ",", # 分隔符，常见的还有 "\t"（制表符）、";" 等
"quotechar": '"', # 字符串的引号包裹符，常见为双引号或单引号
# 无表头时手动指定字段名；有表头的 CSV 不要用这个，否则会把第一行也当成数据
"fieldnames": ["name", "age", "gender"],
},
# 可选：指定编码，处理中文乱码
encoding="utf-8",
# 可选：指定每一行 page_content 的格式模板
source_column=None, # 可以指定用某一列的值作为 metadata 的 source
)
data = loader.load()
print("加载的文档数量：", len(data))
for doc in data:
print("---")
print("内容：", doc.page_content)
print("元数据：", doc.metadata)9.2 JSONLoader
'''
要使用 JSONLoader，需要先安装 jq 库
pip install jq
JSONLoader 是一个用于加载 JSON 数据的工具类。它可以从文件、字符串或 URL 中加载 JSON 数据，并将其解析为 Python 对象。
jq 是一个跨平台的 JSON 解析工具，LangChain 的 JSONLoader 底层就是用它来实现 JSON 数据抽取的。
JSoNLoader使用jq的解析语法，常见如：
·表示根、表示数组
.name表示从根取name的值
.hobby[1]表示取hobby对应数组的第二个元素·表示将数组内的每个字典（JSON对象）都取到
.name表示取数组内每个字典（JSoN）对象的name对应的值
JS0NLoader初始化有4个主要参数：
file_path：文件路径，必填
jq_schema：jq解析语法，必填
text_content：抽取到的是否是字符串，默认True，非必填
json_lines：是否是JsonLines件，默认False，非必填
•jsonLines文件：每一行都是一个独的字典（Json对象）
常见使用场景
普通 JSON 文件：如配置文件、接口返回的 JSON 数据，用 jq_schema 抽取需要的字段。
JSON Lines 文件：如日志文件、批量数据，开启 json_lines=True 逐行解析。
嵌套 JSON 解析：通过 jq_schema 快速抽取深层嵌套的字段，无需手动遍历。
'''
from langchain_community.document_loaders import JSONLoader
# 假设 JSON 结构：[{"name": "周杰伦", "age": 11}, {"name": "蔡依临", "age": 12}]
loader = JSONLoader(
file_path="./data.json",
jq_schema=".[].name", # 抽取数组中每个对象的 name 字段
text_content=True
)
documents = loader.load()
for doc in documents:
print(doc.page_content) # 输出：周杰伦、蔡依临
# JSON Lines 文件示例：
# {"name": "周杰伦", "age": 11, "gender": "男"}
# {"name": "蔡依临", "age": 12, "gender": "女"}
# {"name": "王力宏", "age": 11, "gender": "男"}
loader = JSONLoader(
file_path="./data.jsonl",
jq_schema=".",
text_content=False,
json_lines=True # 开启 JSON Lines 模式，逐行解析
)
documents = loader.load()
print(len(documents)) # 输出 3，每个 JSON 对象对应一个 Document9.3 TextLoader
""""
除了前文学习的三个Loader以外，还有一个基本的加载器：TextLoader
作用：读取文本文件（如.txt），将全部内容放入一个Document对象中。
RecursiveCharacterTextSplitter 递归字符文本分割器
这是 LangChain 官方推荐的默认文本分割器，核心目标是在保持语义 / 上下文完整性和控制片段大小之间找到平衡，开箱即用效果佳。
基于文本的自然段落分割大文档为小文档
可以指定小文档的最大字符数、重叠字符数
可以动指定段落划分的依据（符号）以及字符数量统计函数
它的核心逻辑是递归式分割：
先尝试用最粗粒度的分隔符（比如段落换行 \n\n）分割文本
如果分割后的片段依然超过设定的 chunk_size，就递归使用更细粒度的分隔符（比如换行 \n、句号 。、感叹号 !、问号 ? 等）继续拆分
直到所有片段都不超过最大长度，或者所有分隔符都用完为止
同时通过 chunk_overlap 控制片段之间的重叠，避免语义断裂
适用场景与优势
适用场景：绝大多数通用文本分割场景，比如文档、文章、教程、对话记录等，尤其适合语义连贯的大文本。
核心优势：
自动优先按自然语义边界分割，减少语义断裂
递归兜底机制，保证无论文本格式如何都能完成分割
重叠片段设计，有效解决上下文丢失问题
支持自定义分隔符，适配中文、英文、混合文本等不同语言场景
比如一段超过 500 字符的 Python 教程文本，分割器会：
先按段落拆分，如果单个段落不超过 500 字符，就直接作为一个片段
如果段落太长，就按句子拆分，把完整句子优先保留在片段中
每个片段和前后片段有 50 个字符的重叠，比如前一段的结尾 “Python 中函数的参数分为位置参数和关键字参数”，会出现在后一段的开头，保证语义连贯
"""
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
# 修改路径为绝对路径，确保文件加载成功
loader = TextLoader(
"C:\\Users\\1\\Desktop\\MyPythonProject\\data\\Python基础语法.txt", # 使用绝对路径
encoding="utf-8",
)
docs = loader.load() # 得到完整的Document对象
# 2. 初始化递归分割器
splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # 单个片段的最大字符数
chunk_overlap=50, # 相邻片段之间允许重叠的字符数，用于保留上下文
# 分割优先级列表：先尝试用前面的分隔符，再用后面的
separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""], #用于分隔的一些常见符号
length_function=len, # 用于计算文本长度的函数，默认是len
)
# 3. 对Document进行分割
split_docs = splitter.split_documents(docs)
print(f"原始文本长度：{len(docs[0].page_content)} 字符")
print(f"分割后片段数量：{len(split_docs)}")
# 打印分割后的片段内容
for i, doc in enumerate(split_docs):
print(f"片段 {i + 1}：")
print(doc.page_content)
print("-" * 50) # 分隔线
"""
参数详解：
chunk_size：单个文本片段的最大字符数，控制每个片段的长度上限，避免过长导致向量库处理效率低或语义分散。
chunk_overlap：相邻片段之间的重叠字符数，比如设置为 50，前一个片段的最后 50 个字符会出现在下一个片段的开头，保证上下文不会被生硬切断，适合有连续语义的文本。
separators：分割优先级列表，递归分割的核心。从左到右优先级从高到低：
先尝试用 \n\n（段落）分割，优先按自然段落拆分
再用 \n（换行）拆分
再用中文标点 。 ！ ？ 或英文标点 . ! ? 拆分句子
最后用空格和空字符串兜底，保证文本一定能被分割
length_function：计算文本长度的函数，默认是 len，也可以自定义，比如按 token 数计算长度，适配不同大模型的 token 限制。
"""9.4 PyPDFLoader
"""
LangChain内支持许多PDF的加载器，我们选择其中的PyPDFLoader使用。
PyPDFLoader加载器，依赖PyPDF库，所以，需要安装它：
pip install pypdf
"""
from langchain_community.document_loaders import PyPDFLoader
# 1. 初始化PDF加载器
loader = PyPDFLoader(
"""
file_path="./data/encrypted.pdf",
mode="single", # 加载模式，默认为 single;不设置默认PAGE,每个页面形成一个Document文档对象；single模式将整个PDF作为一个Document对象加载
password="your_pdf_password"
"""
file_path="C:\\Users\\1\\Desktop\\MyPythonProject\\data\\Harness design for long-running application development _ Anthropic.pdf",
)
# 2. 方式一：一次性加载所有页（load）
docs = loader.load()
print("=== 一次性加载结果 ===")
print(f"PDF 总页数：{len(docs)}")
for doc in docs:
print(f"页码：{doc.metadata['page']}")
print(f"内容片段：{doc.page_content[:100]}...\n")
# 3. 方式二：惰性加载（lazy_load，逐页读取，适合大文件）
print("\n=== 惰性加载结果 ===")
i = 0
for doc in loader.lazy_load():
i += 1
print(doc)
print("=" * 20, i)10 向量 VECTOR
10.1 内部向量存储
# ==============================================================
# 向量存储（Vector Stores）核心概念与使用示例
# 功能：存储文本嵌入向量，并支持高效的相似性检索，是 RAG 流程的核心组件
# 典型流程：
# 1. 索引阶段：文档 → 嵌入模型 → 向量 → 存入向量库
# 2. 查询阶段：查询文本 → 嵌入模型 → 查询向量 → 向量库相似性搜索 → 相关文档
# LangChain 为所有向量库提供了统一接口：
# add_documents: 添加文档
# delete: 删除文档
# similarity_search: 相似性检索
# ==============================================================
# ==============================================================
# 内部向量存储（InMemoryVectorStore）完整示例
# 功能：将 CSV 数据加载为文档，存入内存向量库，并完成新增/删除/检索操作
# 依赖安装：
# pip install langchain langchain-community dashscope
# ==============================================================
# 1. 导入所需模块
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.document_loaders import CSVLoader
# 2. 初始化嵌入模型（DashScopeEmbeddings 是阿里通义的嵌入服务）
embedding = DashScopeEmbeddings()
# 3. 创建内存向量存储实例
vector_store = InMemoryVectorStore(
embedding=embedding
)
# 4. 加载 CSV 数据为 Document 对象
# file_path: 你的 CSV 文件路径
# encoding: 指定编码，避免中文乱码
# source_column: 指定哪一列作为文档的 source 元数据（比如记录数据来源）
loader = CSVLoader(
file_path="C:\\Users\\1\\Desktop\\MyPythonProject\\data\\Python基础语法.csv",
encoding="utf-8",
source_column="source", # 可选：如果 CSV 里有一列叫 source，就用它作为来源标识
)
# 加载所有行，每行转为一个 Document
documents = loader.load()
# 打印第一条文档，查看结构
print("=== 加载的第一条文档 ===")
print(documents[0])
print()
# 5. 将文档添加到向量存储，并生成自定义 ID
# ids 必须是字符串列表，长度和 documents 一致
ids = [f"id{i}" for i in range(1, len(documents) + 1)]
vector_store.add_documents(
documents=documents,
ids=ids
)
print(f"已添加 {len(documents)} 条文档到向量库，ID 示例：{ids[:3]}...")
print()
# 6. 删除示例：通过 ID 删除指定文档
# 传入要删除的 ID 列表
vector_store.delete(["id1", "id2"])
print("已删除 id1 和 id2 的文档")
print()
# 7. 相似性检索示例：查询与问题相关的 Top-k 文档
query = "Python是不是简单易学呀"
k = 3 # 返回前 3 个最相似的结果
result = vector_store.similarity_search(
query=query,
k=k
)
# 打印检索结果
print("=== 相似性检索结果 ===")
for idx, doc in enumerate(result):
print(f"结果 {idx+1}：")
print(" 内容：", doc.page_content[:100] + "..." if len(doc.page_content) > 100 else doc.page_content)
print(" 元数据：", doc.metadata)
print()10.2 Chroma 外部向量存储
# ==============================================================
# Chroma 外部向量存储完整示例（支持持久化存储）
# 功能：将 CSV 数据存入 Chroma 向量库，支持新增/删除/检索，数据重启后不丢失
# 依赖安装：
# pip install langchain langchain-community langchain-chroma chromadb dashscope
# ==============================================================
# 1. 导入所需模块
from langchain_chroma import Chroma
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.document_loaders import CSVLoader
# 2. 初始化 Chroma 向量存储
# collection_name: 集合名称，类似数据库表名，用来区分不同知识库
# embedding_function: 嵌入模型，用于将文本转为向量
# persist_directory: 数据持久化目录，重启程序数据不会消失
vector_store = Chroma(
collection_name="test",
embedding_function=DashScopeEmbeddings(),
persist_directory="./chroma_db"
)
# 3. 加载 CSV 数据为 Document 对象
# 这里使用之前给你生成的 info.csv 文件
loader = CSVLoader(
file_path="C:\\Users\\1\\Desktop\\MyPythonProject\\data\\Python基础语法.csv",
encoding="utf-8",
source_column="source" # 指定 CSV 中哪一列作为文档来源元数据
)
documents = loader.load()
# 打印第一条文档，确认加载成功
print("=== 加载的第一条文档 ===")
print(documents[0])
print()
# 4. 将文档添加到 Chroma 向量库，并生成自定义 ID
ids = [f"id{i}" for i in range(1, len(documents) + 1)]
vector_store.add_documents(
documents=documents,
ids=ids
)
print(f"已添加 {len(documents)} 条文档到 Chroma 向量库，ID 示例：{ids[:3]}...")
print()
# 5. 删除示例：通过 ID 删除指定文档
vector_store.delete(["id1", "id2"])
print("已删除 id1 和 id2 的文档")
print()
# 6. 相似性检索示例
query = "Python是不是简单易学呀"
k = 3 # 返回前 3 个最相似的结果
result = vector_store.similarity_search(
query=query,
k=k
)
# 打印检索结果
print("=== Chroma 向量库检索结果 ===")
for idx, doc in enumerate(result):
print(f"结果 {idx+1}：")
print(" 内容：", doc.page_content[:100] + "..." if len(doc.page_content) > 100 else doc.page_content)
print(" 元数据：", doc.metadata)
print()
# 可选：带相似度分数的检索
result_with_score = vector_store.similarity_search_with_score(query=query, k=3)
print("=== 带相似度分数的检索结果 ===")
for doc, score in result_with_score:
print(f"相似度分数：{score:.4f}")
print(f"内容片段：{doc.page_content[:100]}...")
print()10.3 向量检索构建提示词
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.output_parsers import StrOutputParser
model = ChatTongyi(model="qwen3-max")
prompt = ChatPromptTemplate.from_messages(
[
{"role": "system", "content": "按照参考资料回答用户的问题{context}"},
{"role": "user", "content": "用户提问：{input}"}
]
)
vector_store = InMemoryVectorStore(
embedding=DashScopeEmbeddings(model="text-embedding-v4"),
)
# 准备参考资料，即向量库数据，这里我选择把数据保存再txt文件中上传
# 导入txt
loader = TextLoader(
"C:\\Users\\1\\Desktop\\MyPythonProject\\data\\hermes.txt", # 使用绝对路径
encoding="utf-8",
)
docs = loader.load()
# 递归字符文本分割器，设置分割参数
txt_splitter = RecursiveCharacterTextSplitter(
chunk_size=200, # 单个片段的最大字符数
chunk_overlap=50, # 相邻片段之间允许重叠的字符数，用于保留上下文
# 分割优先级列表：先尝试用前面的分隔符，再用后面的
separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""], # 用于分隔的一些常见符号
length_function=len, # 用于计算文本长度的函数，默认是len
)
# 分割后的文档列表
split_docs = txt_splitter.split_documents(docs)
# 将分割后的文档添加到向量库中 vector_storeadd_documents
ids = [f"doc_{i}" for i in range(len(split_docs))]
vector_store.add_documents(documents=split_docs, ids=ids)
print("文档已成功存入向量库")
input_text = "请问Hermes是什么？"
# 从向量库中检索相关文档-相似度搜索
retrieved_docs = vector_store.similarity_search(input_text, top_k=3)
print("检索到的文档：", retrieved_docs) # 调试打印
#链式调用最后输出字符串
chain = prompt | model | StrOutputParser()
result = chain.invoke({"input": input_text, "context": retrieved_docs})
print("模型输出：", result) # 调试打印
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
12
0- 0
已为社区贡献1条内容
所有评论(0)