名词解释

1. RAG - 检索增强生成

        一种将外部知识库与大模型相结合的技术架构。简单说，就是先“检索”相关资料，再让模型基于这些资料“生成”答案。

  • 核心流程：用户提问 -> 从数据库（通常是向量数据库）中检索相关文档片段 -> 将问题和检索到的片段一起交给大模型 -> 模型生成基于可靠来源的答案。

  • 类比：就像开卷考试。相比于让模型只凭记忆（闭卷）回答，RAG允许它查阅指定的资料（开卷），从而提供更准确、更新且可追溯来源的答案。

  • 解决什么问题：缓解大模型的“幻觉”（编造信息）、知识过时、以及无法处理私有/专业数据的问题。

2. Agent - 智能体

        一个能感知环境、进行决策并执行动作来完成给定目标的AI系统。在大模型语境下，通常指以大模型为“大脑”，能够调用各种工具（如搜索引擎、计算器、API、代码解释器）来完成复杂任务的程序。

  • 核心能力：

    • 规划：将复杂任务拆解为步骤。

    • 工具使用：学习调用外部工具。

    • 记忆：保留对话和结果的历史。

  • 类比：像一个人类助理。你告诉它“帮我规划一个三天的北京行程”，它会自己决定先去搜索景点、查天气、再计算预算、最后整理成表格给你。它不只是聊天，而是能“动手做事”。

RAG与Anget与RAG的关系

        Agent可以调用RAG作为它的一个“工具”，来获取专业知识。例如，一个客服Agent在接到专业问题时，会先调用RAG工具从产品手册中检索信息，再生成回复。

3. Ollama

        一个专注于在本地（如个人电脑、自有服务器）快速部署和运行开源大模型的框架和工具。它提供了简单的命令行和API，让用户无需复杂的配置就能下载、运行和管理各种模型（如 Llama 3、Mistral、Gemma 等）。

  • 核心特点：轻量、易用、本地化。是让开发者和爱好者快速上手体验和测试开源大模型的利器。

  • 与上述概念的关系：Ollama提供了模型本身。你可以用Ollama部署的模型作为RAG系统的“生成大脑”，也可以作为Agent的“决策大脑”。它是底层的基础设施。

大模型接入

方式一：本地部署：Ollama，1.5b、7b、8b、32b…受我们电脑硬件的限制

方式二：云端大模型：阿里云百炼平台，满参数 + 免费额度大模型服务平台百炼控制台

阿里云百炼平台

登录

1.登录网站2.认证

认证之后重新打开网站，就会弹出免费赠送额度，这个额度够我们学习使用

如果经常使用一个模型，并且担心这个模型的额度会用完扣费，可以通过下面这个方式把“免费额度用完即停”打开

创建API key

虽然用下面这个方式可以像deepseek一样进行问答，但是不能通过代码访问

如果想通过代码访问就要创建一个API key

点击确定，就创建好了

只需要有这个API key这个“钥匙”我们就可以通过代码联网正常的访问到阿里云百炼的模型

代码调用阿里云百炼平台上的模型

仅需要： 阿里云百炼平台上，创建API-KEY（如上）

                通过pip为Python程序提供 OpenAI 库: pip install openai

                编写代码测试（可从百炼平台官网复制代码，直接执行，如下）

刚刚我们装了openai库，可以看到这里有个openai兼容的选项，选择这个，将对应示例代码复制下来在代码里就可以用了

编写测试代码，在api_key里面传入自己的可以，然后就可以通过改变content里面的内容来进行问答

用环境变量隐藏明文APIKEY

        将APIKEY明文显示在代码中，是有很大的安全隐患的。我们可以通过环境变量来隐藏明文APIKEY。

我们将OPENAI_API_KEY和DASHSCOPE_API_KEY（后面学习langchain的时候要用）一起配了

之后重启电脑，打开pycharm，将那一行代码删了会发现还是可以运行的

但是我们一般还是按照从网站直接复制的代码用，直接复制的完全可以运行，并且更具有健壮性

原因解释：

   os.getenv("DASHSCOPE_API_KEY")来从环境变量中获取API Key。如果你在系统的环境变量中已经配置了DASHSCOPE_API_KEY，那么即使你删除了这一行，代码仍然可以从环境变量中读取到API Key，所以可以正常运行。

        但是，如果你删除了这一行，并且没有在环境变量中配置DASHSCOPE_API_KEY，那么代码将无法获取API Key，会导致运行错误。

        所以，为了代码的健壮性和可移植性，建议保留这一行，这样代码会首先尝试从环境变量中获取API Key。如果没有配置环境变量，你还可以按照注释中的方式，直接使用api_key参数传入API Key。

Ollama

        为了避免未来阿里云免费额度到期或不提供免费活动导致的无法开发，额外补充： 基于Ollama部署本地模型，供代码调用。学习主体还是基于阿里云百炼平台做开发。

ollama：

        是一款旨在简化大型语言模型本地部署和运行过程的开源软件。 ollama提供了一个轻量级、易于扩展的框架，让开发者能够在本地机器上轻松构建和管理LLMs（大型语言模型） 通过ollama，开发者可以导入和定制自己的模型，无需关注复杂的底层实现细节。

        ollama 支持多种开源模型，涵盖文本生成、代码生成、多模态推理等场景。用户可以根据需求选择合适的模型，并通过简单的命令行操作在本地运行。

        网址：https://ollama.com

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

Windows/Mac系统部署Ollama

        只需要进入官方网站点击Download按钮下载安装。

        安装完成后，通过命令： ollama run 模型名称 即可运行对应的模型，并在命令行内做交互。

下载完成后打开这个exe文件，然后直接点击安装就可以了

安装完成后就进入到这个界面

选择模型时我们可以看到后面写有4B、8B...因为我们在Ollama里面运行的模型都是蒸馏模型，不是标准的完整版的大模型，因为完整版的大模型跑起来对硬件要求非常高，要几百万。这个相当于瘦身版的大模型，虽然没有大模型那么强，但是也可以完成八九不离十。4B、8B就是代表参数的量，显存多大就跑多大的就可以，要不然吐字会慢。如：显存8GB,就跑8B的大模型

第一次运行这个模型的时候需要下载模型

代码调用Ollama本地模型

只需要将原有代码（之前在阿里云上下载的）进行简单改动即可。

  • 将base_url改为：http://localhost:11434/v1（注意：Ollama 默认使用 HTTP 服务，并不提供 HTTPS）
  • 将model改为对应本地模型名称，如：qwen3:4b（如果你还没有下载模型，需要先拉取）

补充：如果Ollama没有启动，代码是不能调用Ollama本地模型的

Ollama的启动，模型的拉取：

# 启动 Ollama 服务（Windows会自动启动）
ollama serve

# 常见输出：
# >>> Starting Ollama...
# >>> Listening on http://localhost:11434

# 列出官方模型库
ollama list

# 搜索特定模型
ollama search llama
ollama search qwen
ollama search gemma

# 下载模型（7B参数量的常见选择）
ollama pull llama3.2:3b         # 3B参数，轻量级
ollama pull qwen2.5:7b          # 7B参数，中文优秀

OpenAI库的基础使用

OpenAI库是一个Python库，它允许开发者通过编程方式与OpenAI的服务进行交互，比如GPT-3、GPT-4、DALL-E等。但是，由于OpenAI的API接口是开放的，其他一些兼容OpenAI API的服务（如Ollama、vLLM、FastChat等）也可以使用这个库进行调用。

OpenAI库的核心是提供了一种简便的方法来构造请求并处理响应，使得开发者可以轻松地集成AI功能到他们的应用中。

        在之前使用的代码中，我们实际上是将OpenAI库用于与Ollama服务进行交互，因为Ollama提供了与OpenAI兼容的API接口。所以，我们只需要将基础URL改为Ollama服务的地址（如http://localhost:11434/v1），并使用一个虚拟的API密钥（因为Ollama通常不需要认证）即可。

之前代码有关OpenAI库的讲解

获取客户端对象主要是用如上2个参数：

  • api_key：模型服务商提供的APIKEY密钥
  • base_url：模型服务商的API接入地址 主要基于此参数来切换不同的模型服务商（如OpenAI、阿里云、腾讯云等，都是云服务商，它们提供了各种云服务，其中包括大模型服务）

因此，通过改变base_url和api_key，你可以切换不同的模型服务商，只要它们提供了与OpenAI兼容的API。

client.chat.completions.create创建ChatCompletion对象 主要参数有2个：

  • model：选择所用模型，如代码的qwen3-max
  • messages：提供给模型的消息，list类型，可以包含多个字典消息。

                         每个字典消息包含2个关键字： role和 content

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

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

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

                          

OpenAI库的流式输出：

可以设定结果输出为stream模式（流式输出），获得更好的使用体验。 开启流式输出主要就2步：

  • 在client.chat.completions.create()调用模型的时候设定参数：stream=True
  • for循环response对象，并在循环内输出内容

OpenAI库 附带历史消息调用模型

        通过调用模型传入的参数messages，因为其要求是list对象，这表明其支持非常多的消息在内。 我们可以基于此，将历史消息填入，让模型知晓对话的上下文，更好的回答。

        所以我们可以在调用模型的时候，通过： 在messages的list内，组织历史消息提供给模型

PS：当前的历史消息是一次性的，如果是生产系统可以将消息保存到文件、数据库等持久化工具内，需要的时候提取使用 后续学习LangChain库，会学习短期记忆和长期记忆的使用方法。

提示词实战案例

案例一

案例背景

当前金融领域信息化发展的时代,金融数据大量激增,许多投资者和研究者试图通过对这些数据进行深度分析而获得一些有效的决策和帮助,尽可能减少决策失误带来的损失。 所以,针对金融数据的分析方法研究是目前十分有益且热门的话题。

当前案例主要有三大业务场景实现：

  • 基于大模型完成：金融文本分类
  • 基于大模型完成：金融文本信息抽取
  • 基于大模型完成：金融文本匹配

大模型选择：Qwen在线大模型（阿里云通义千问 qwen3-max）

采用方法：基于Few-Shot + Zero-Shot的思想，设计prompt（提示词）, 进而应用大模型完成相应的任务

Zero-shot学习（Zero-shot Learning）

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

在提示词优化中： Zero-shot思想用于基于已训练的能力，不提供任何示例，仅通过语言去描述任务的要求、目标和约束，让模型直接生成结果。 简单来说就是“用语言定义任务，解放（信任）模型的预训练知识”

例如：

在模型训练中： 已知马（四脚兽）、虎（有条纹）、熊猫（黑白色）的特征，但未训练过斑马的数据（不认识）

告知模型：斑马是四脚兽、有黑白色的条纹 模型可以在已知数据中进行推理，从而识别斑马。

模型在马中提取四脚兽什么样、在虎中知道有条纹是什么样、在熊猫中知道黑白色什么样，进而识别出斑马

Few-shot学习（Few-shot Learning）

        是指少样本学习，当模型在学习了一定类别的大量数据后，对于新的类别，只需要少量的样本就能快速学习，对应的有one-shot learning，单样本学习，也算样本少到为一的情况下的一种few-shot learning。

例如：

在模型训练中（相似度判断方法）： 基于少量企鹅样本并结合相识度判断，推论未知图片内含“企鹅”

在提示词优化中： Few-shot主要用于基于少量示例，让模型参考示例回答。 简单来说就是“用示例定义任务，在模型的预训练知识的基础上，提升模型回答的对齐精度（比如参考示例的格式）”

文本分类

为了让模型知道什么叫做「文本分类」，我们借用 FewShot 的方式，给模型展示一些正确的例子：

User: "今日，股市经历了一轮震荡，受到宏观经济数据和全球贸易紧张局势的影响。投资者密切关注美联储可能的政策调整，以适应市场的不确定性。" 是['新闻报道', '公司公告', '财务公告 '分析师报告']里的什么类别？

Bot: 新闻报道

User: "本公司年度财务报告显示，去年公司实现了稳步增长的盈利，同时资产负债表呈现强劲的状况。经济环境的稳定和管理层的有效战略执行为公司的健康发展奠定了基础。"是['新闻报道', '公司公告', '财务公告 '分析师报告']里的什么类别？

Bot: 财务报告

分类任务代码实现
from openai import OpenAI
import os

# 1. 获取client对象，OpenAI类对象
client = OpenAI(
    # 若没有配置环境变量，请用百炼API Key将下行替换为：api_key="sk-xxx",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

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

####数据信息#######################################################################################################################
examples_data = {       # 示例数据
    '新闻报道': '今日，股市经历了一轮震荡，受到宏观经济数据和全球贸易紧张局势的影响。投资者密切关注美联储可能的政策调整，以适应市场的不确定性。',
    '财务报告': '本公司年度财务报告显示，去年公司实现了稳步增长的盈利，同时资产负债表呈现强劲的状况。经济环境的稳定和管理层的有效战略执行为公司的健康发展奠定了基础。',
    '公司公告': '本公司高兴地宣布成功完成最新一轮并购交易，收购了一家在人工智能领域领先的公司。这一战略举措将有助于扩大我们的业务领域，提高市场竞争力',
    '分析师报告': '最新的行业分析报告指出，科技公司的创新将成为未来增长的主要推动力。云计算、人工智能和数字化转型被认为是引领行业发展的关键因素，投资者应关注这些趋势'
}
# 分类列表
examples_types = ['新闻报道', '财务报道', '公司公告', '分析师报告']
# 提问数据
questions = [
    "今日，央行发布公告宣布降低利率，以刺激经济增长。这一降息举措将影响贷款利率，并在未来几个季度内对金融市场产生影响。",
    "ABC公司今日发布公告称，已成功完成对XYZ公司股权的收购交易。本次交易是ABC公司在扩大业务范围、加强市场竞争力方面的重要举措。据悉，此次收购将进一步巩固ABC公司在行业中的地位，并为未来业务发展提供更广阔的发展空间。详情请见公司官方网站公告栏",
    "公司资产负债表显示，公司偿债能力强劲，现金流充足，为未来投资和扩张提供了坚实的财务基础。",
    "最新的分析报告指出，可再生能源行业预计将在未来几年经历持续增长，投资者应该关注这一领域的投资机会",
    "小明喜欢小新哟"
]

#################################################################################################################




#组装提问数据依次放到messages中
for key, value in examples_data.items():
    messages.append({"role": "user", "content": value})
    messages.append({"role": "assistant", "content": key})

#将问题也添加进来
#注意不能这样写 messages=messages.append({"role": "user", "content": f"按照示例，回答这段文本的分类类别：{q}"}),
# messages.append(...)，这个append方法会修改原messages列表，并且返回的是None，因此实际上我们传递的messages参数是None。
for q in questions:
    completion = client.chat.completions.create(
        model="qwen3-max",
        messages=messages+ [{"role": "user", "content": f"按照示例，回答这段文本的分类类别：{q}"}],
        stream=True # 启用流式输出功能
    )
    for chunk in completion:
        print(chunk.choices[0].delta.content, end="", flush=True)

案例二：

补充：Json数据格式

JSON（JavaScript Object Notation）是一种轻量级的数据交换格式，易于人阅读和编写，同时也易于机器解析和生成。 Json是带有格式的字符串，主要用于数据交换，即程序和程序之间的信息互传，使用Json会更加方便，如下示例：

.txt文本非结构化 抽取信息不方便

.csv（固定分隔符）文本

                      结构化 抽取信息方便 数据不含Schema，有一定风险

JSON文本：结构化 抽取信息方便 每条数据都包含Schema 空间占用大

Json主要有2种结构：Json对象和Json数组
Json对象

Json数组

Json对象 => Python字典

Json数组 => Python列表内含多个字典

Json在Python中，就是 字典和列表套字典的字符串表现形式。

Json与Python之间的转化

主要使用Python内置的json库

json.dumps(字典或列表, ensure_ascii=False)：将字典或列表转换为Json字符串
  • ensure_ascii=False参数确保中文能正常显示
  • 返回值：Json字符串

注意我们输出的Json中字符串都是用双引号包围，这个是固定的，要不然Java识别不出来。

但是在python中单引号双引号都可以

import json
#转化为json字符串
d = {
    'name': '周杰轮',
    "age": 11,
    "gender": "男"
}

s = json.dumps(d, ensure_ascii=False)
print(s)#{"name": "周杰轮", "age": 11, "gender": "男"} 用的是标准的双引号

json.loads(json字符串)：将Json字符串转换为Python字典或列表
  • 返回值：Python字典 或 Python列表

json字符串要用单引号包围,因为里面是双引号

#用单引号包围,因为里面是双引号
json_str = '{"name": "周杰轮", "age": 11, "gender": "男"}'

res_dict = json.loads(json_str)
print(res_dict, type(res_dict))

案例背景

1.'2023-02-15，寓意吉祥的节日，股票佰笃[BD]美股开盘价10美元，虽然经历了波动，但最终以13美元收盘，成交量微幅增加至460,000，投资者情绪较为平稳。'

2.'2023-04-05，市场迎来轻松氛围，股票盘古(0021)开盘价23元，尽管经历了波动，但最终以26美元收盘，成交量缩小至310,000，投资者保持观望态度。',

我们想要的json格式：

 { "金融": ["日期":"XXX",  "股票名称":"XXXX",  "开盘价":"XXX",  "收盘价":"XXX",  "成交量":"XXX"], }

Prompt设计

根据所给代码，以及要求进行提示词设计

from openai import OpenAI
import json

client = OpenAI(
    # base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
    base_url="http://localhost:11434/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"按照上述示例，现在抽取这个句子的信息：{要抽取的句子文本}"}]}
]
"""
#根据上面内容完善代码

设计完成后的代码：

from openai import OpenAI
import json

client = OpenAI(
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
    # base_url="http://localhost:11434/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人民币收盘。"
]

messages = [
    {"role": "system", "content": "你是一个金融专家，你帮我完成信息抽取，我给你句子，你抽取{schema}信息，按JSON字符串输出，如果某些信息不存在，用'原文未提及'表示，请参考如下示例："},
]
for i in examples_data:
    messages.append({"role": "user", "content": i["content"]})
    messages.append({"role": "assistant", "content": json.dumps(i["answers"], ensure_ascii=False)})

for i in questions:
    completion = client.chat.completions.create(
        model="qwen3-max",
        messages=messages+[{"role": "user", "content": i}],
        stream=True  # 启用流式输出功能
    )
    for chunk in completion:
        print(chunk.choices[0].delta.content, end="", flush=True)

案例三

案例背景

Prompt设计

根据所给代码，以及要求进行提示词设计

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: [...]"}
"""

完整代码

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

messages=[{"role": "system", "content": "你帮我完成文本匹配，我给你2个句子，被[]包围，你判断它们是否匹配，回答是或不是，请参考如下示例："}]
for key, value in examples_data.items():
    for i in value:
        messages.append({"role": "user", "content": f"句子1：[{i[0]}],句子2：[{i[1]}]"})
        messages.append({"role": "assistant", "content":key})


for q in questions:
    completion = client.chat.completions.create(
        model="qwen3-max",
        messages=messages+ [{"role": "user", "content": f"[句子1：[{q[0]},句子2：[{q[1]}]]"}],
        stream=True # 启用流式输出功能
    )
    for chunk in completion:
        print(chunk.choices[0].delta.content, end=" ", flush=True)

LangChain

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

为什么不用OpenAI SDK直接调用，反而用LangChain？

OpenAI SDK只能做一件事：发送请求，接收回复，然后就结束了，需要自己处理所有逻辑，就向上面我们练习的一样，需要手敲。

completion = client.chat.completions.create(
    model="xxx",
    messages=[...],
    ...
)

LangChain安装

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

这些包的功能

langchain：核心包

langchain-community：社区支持包，提供了更多的第三方模型调用（我们用的阿里云千问模型就需要这个包）

langchain-ollama：Ollama支持包，支持调用Ollama托管部署的本地模型

langchain-chroma：ChromaDB支持包，支持调用ChromaDB

dashscope：阿里云通义千问的Python SDK

chromadb：轻量向量数据库（后续使用）

bs4：BeautifulSqop4库，协助解析HTML文档（后续学习文档加载器使用）

RAG

为什么用RAG？

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

  • LLM的知识不是实时的，模型训练好后不具备自动更新知识的能力，会导致部分信息滞后
  • LLM领域知识是缺乏的，大模型的知识来源于训练数据，这些数据主要来自公开的互联网和开源数据集，无法覆盖特定领域或高度专业化的内部知识
  • 幻觉问题，LLM有时会在回答中⽣成看似合理但实际上是错误的信息
  • 数据安全性

RAG（Retrieval（检索）-Augmented（增强） Generation（生成））即检索增强生成，为大模型提供了从特定数据源检索到的信息，以此来修正和补充生成的答案。可以总结为一个公式：RAG = 检索技术 + LLM 提示

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

离线准备线是从你能获得到信息的来源一直获取各种各样的知识，将这些知识进行处理分割向量化存入向量数据库中，是默默的在背后一直在做的

ps:向量数据库是自己的

向量的基础概念

RAG流程中，向量库是一个重要的节点。

  • 离线流程：知识和信息 -->向量嵌入（向量化）--> 存入向量库
  • 在线流程：用户的提问 -->向量嵌入（向量化）--> 在向量库中匹配

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

文本嵌入模型（如text-embedding-v1）通过深度学习等技术，从文本提取语义特征并映射为固定长度的数字序列。

向量匹配模拟示例：

生成向量的维度：

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

如text-embedding-v1模型，可以生成1536维的向量（一段文本固定得到1536个数字序列），比较实用。

  • 1536个数字表示，这段文本在1536个主题（抽象的语义特征）方向上的得分（强度）                             
  • 生成向量的维度越多，就更好的记录文本的语义特征，做语义匹配会更加精准。
  • 更多的向量会在计算、存储和匹配过程中，带来更大的压力。

选择合适的向量维度需要在精确和性能之间做平衡。 一般1536维算是比较好的选择。

余弦相似度：

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

                                        

LangChain

LangChain目前支持三种类型的模型：

LLMs（大语言模型）、Chat Models(聊天模型)、Embeddings Models(嵌入模型）.

  • LLMs:是技术范畴的统称，指基于大参数量、海量文本训练的 Transformer 架构模型，核心能力是理解和生成自然语言，主要服务于文本生成场景
  • 聊天模型:是应用范畴的细分，是专为对话场景优化的 LLMs，核心能力是模拟人类对话的轮次交互，主要服务于聊天场景。如果我们用Chat Model，我们会将输入构造为一个消息列表，每个消息都有角色（如“user”、“assistant”、“system”），然后模型会根据这个对话历史生成回复。
  • 文本嵌入模型: 文本嵌入模型接收文本作为输入, 得到文本的向量.

三种类型的模型的使用举例

假设我们要构建一个金融问答系统，它可以根据提供的金融文档来回答问题。
步骤：

  1.  我们有一些金融文档（比如公司财报、新闻报道等）。
  2. 我们想要问一个问题，比如“公司去年的利润是多少？”
  3. 我们需要从文档中找到相关信息，然后结合这些信息来回答问题。

这里我们会用到两种模型：

  • 嵌入模型：将文档分割成小块，然后将每一块转换为向量，并存储到向量数据库中。
  • 聊天模型（或LLM）：当我们提问时，先从向量数据库中检索出与问题相关的文档块，然后将这些文档块和问题一起发送给模型，让它生成答案。

为什么需要嵌入模型？

        因为我们需要快速找到与问题相关的文档。我们可以将问题也转换为向量，然后在向量数据库中搜索与问题向量最相似的文档向量（即语义相似）。
整个流程如下：

  • 文档 --> 分割成小块 --> 用嵌入模型将每个块转换为向量 --> 存储到向量数据库
  • 提问 --> 用嵌入模型将问题转换为向量 --> 在向量数据库中搜索相似向量 --> 得到相关文档块
  • 将相关文档块和问题组合成提示词 --> 发送给聊天模型（或LLM）--> 得到答案

在这个流程中，我们用了两种模型：嵌入模型和聊天模型（或LLM）。而聊天模型和LLM的区别在于我们如何构造提示词。在LangChain中，聊天模型更适合多轮对话，而LLM更适合单轮文本生成。

误解1：Chat Models 比 LLMs 更高级

不对！Chat Models 底层还是调用 LLMs，只是做了消息格式的转换和优化。

误解2：一个模型只能是一种类型

不对！像 qwen3:7b 这样的模型，既可以当 LLM 用，也可以当 Chat Model 用（通过不同包装器）。

误解3：Embeddings 模型很小

不一定！虽然参数量可能小（如 100M），但有些嵌入模型也很大（如 1B+）。

误解4：必须用 Chat Models 做对话

不一定！LLMs 通过精心设计的提示词也能做对话，但 Chat Models 更方便。

大语言模型的使用

LangChain 访问 阿里云大语言模型、ollama本地大语言模型

我们使用 LangChain 的 Tongyi 类时，它默认调用的是阿里云的通义千问 API。
在 LangChain 中，Tongyi 类已经内置了 API 的基础 URL，因此我们不需要显式指定 base_url。
但是，我们需要提供 API 密钥（通过环境变量 DASHSCOPE_API_KEY 设置）来验证身份。

from langchain_community.llms.tongyi import Tongyi
model = Tongyi(model="qwen-max")
res=model.invoke(input="你是谁？")
print(res)
from langchain_ollama import OllamaLLM
model = OllamaLLM(model="qwen3:4b")
res=model.invoke(input="你是谁？")
print(res)

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

  • invoke方法：一次型返回完整结果
  • stream方法：逐段返回结果，流式输出 
    from langchain_community.llms.tongyi import Tongyi
model = Tongyi(model="qwen-max")
res=model.stream(input="你是谁？")
for chunk in res:
    print(chunk, end="", flush=True)

    聊天模型的使用

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

model=ChatTongyi(model="qwen3-max")
messages=[
    SystemMessage(content="你是一个边塞诗人"),
    HumanMessage(content="写一首唐诗"),
    AIMessage(content="锄禾日当午，汗滴禾下土，谁知盘中餐，粒粒皆辛苦"),
    HumanMessage(content="按照上一首的格式再写一首")
]
res=model.stream(input=messages)
for chunk in res:
    print(chunk.content,end="",flush=True)
from langchain_ollama.chat_models import ChatOllama
from langchain_core.messages import HumanMessage,SystemMessage,AIMessage

model=ChatOllama(model="qwen3:4b")
messages=[
    SystemMessage(content="你是一个边塞诗人"),
    HumanMessage(content="写一首唐诗"),
    AIMessage(content="锄禾日当午，汗滴禾下土，谁知盘中餐，粒粒皆辛苦"),
    HumanMessage(content="按照上一首的格式再写一首")
]
res=model.stream(input=messages)
for chunk in res:
    print(chunk.content,end="",flush=True)

消息的简写形式

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

model=ChatTongyi(model="qwen3-max")
messages=[
    ("system","你是一个边塞诗人"),
    ("human","写一首唐诗"),
    ("ai","锄禾日当午，汗滴禾下土，谁知盘中餐，粒粒皆辛苦"),
    ("human","按照上一首的格式再写一首"),
]
res=model.stream(input=messages)
for chunk in res:
    print(chunk.content,end="",flush=True)

嵌入模型Embeddings Models的使用

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

  • embed_query方法的作用：将单个文本转换为向量表示（嵌入）。通常，embed_query接受一个字符串，返回一个浮点数列表（向量）。

  • embed_documents方法的作用：用于多个文本的批量嵌入。

访问阿里云嵌入模型

from langchain_community.embeddings import DashScopeEmbeddings
#不传model默认使用模型是：text-embedding-v1
model=DashScopeEmbeddings()
print(model.embed_query("你好"))
print(model.embed_documents(["你叫什么名字","你妈妈叫什么名字？"]))

访问Ollama嵌入模型

先在官网找到嵌入模型模型的名称，复制下来，

然后在命令行下载下来ollama pull qwen3-embedding,就可以使用了

from langchain_ollama.embeddings import OllamaEmbeddings

model=OllamaEmbeddings(model="qwen3-embedding:4b")
print(model.embed_query("你好"))
print(model.embed_documents(["你叫什么名字","你妈妈叫什么名字？"]))

Prompts

通用Prompts

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

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

prompt_template = PromptTemplate.from_template("我邻居的性是{lastname},刚生了一个{gender}，你帮我起个名字,简单回答")
text=prompt_template.format(lastname="王",gender="男")

model=Tongyi(modle="qwen-max")
res=model.stream(input=text)
for chunk in res:
    print(chunk, end="", flush=True)

基于chain链的写法

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

prompt_template = PromptTemplate.from_template("我邻居的性是{lastname},刚生了一个{gender}，你帮我起个名字,简单回答")

#基于chain的写法
model=Tongyi(modle="qwen-max")
chain=prompt_template | model
res=chain.stream(input={"lastname":"王","gender":"男"})
for chunk in res:
    print(chunk, end="", flush=True)

组装FewShotPromptTemplate对象并获得最终提示词

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

# 提示模板
prompt_template=PromptTemplate.from_template("我邻居的姓是{lastname},刚生了一个{gender},姓名{name}")
# 示例数据
examples_data=[
    {"lastname":"王","gender":"男","name":"王伟"},
    {"lastname":"张","gender":"女","name":"张三"}]

# 用上面的示例模板 组装创建 few-shot 提示
few_shot_prompt = FewShotPromptTemplate(
    example_prompt=prompt_template,           #示例提示模板示例数据
    examples=examples_data,                   #示例模板的填充数据，list，内套字典
    prefix="告知我应该起什么名字，我提供如下示例：",#示例之前的提示词
    suffix="基于前面的示例告知我，我邻居的姓是{input_lastname},刚生了一个{input_gender},应该起什么名字",#示例之后的提示词
    input_variables=["input_lastname","input_gender"],#声明在前缀或后缀中所需要注入的输入变量名.列表，注入的变量列表

)
# 获得最终提示词
prompt_text=few_shot_prompt.invoke(input={"input_lastname":"李","input_gender":"男"}).to_string()


model=Tongyi(modle="qwen-max")
res=model.stream(input=prompt_text)
for chunk in res:
    print(chunk, end="", flush=True)

ChatPromptTemplate的使用

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

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

chat_prompt_template=ChatPromptTemplate.from_messages(
    [
    ("system","你是一个边塞诗人"),
    MessagesPlaceholder("history"),
    ("human","请写一首关于{topic}的诗"),
    ]
)
history_data=[
    ("human","你来写一首诗"),
    ("ai","床前明月光，疑是地上霜，举头望明月，低头思故乡"),
    ("human", "很好再来一首"),
    ("ai", "锄禾日当午，汗滴禾下土，谁知盘中餐，粒粒皆辛苦")
]
#组装提示词模板
prompt_text=chat_prompt_template.invoke({"history":history_data,"topic":"唐诗"}).to_string()

model=ChatTongyi(model="qwen3-max")
res=model.invoke(prompt_text)
print(res.content)

format和invoke

PromptTemplate、FewShotPromptTemplate、ChatPromptTemplate（后续学习）都拥有format和invoke这2类方法。

format和invoke的区别在于：

chains 链的基础使用

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

chain = prompt_template | model 核心前提：即Runnable子类对象才能入链（以及Callable、Mapping接口子类对象也可加入（后续了解用的不多））。 我们目前所学习到的组件，均是Runnable接口的子类，如下类的继承关系：

在下面代码中：

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

chat_prompt_template = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个边塞诗人，可以作诗。"),
        MessagesPlaceholder("history"),
        ("human", "请再来一首唐诗，无需额外输出"),
    ]
)
history_data = [
    ("human", "你来写一个唐诗"),
    ("ai", "床前明月光，疑是地上霜，举头望明月，低头思故乡"),
    ("human", "好诗再来一个"),
    ("ai", "锄禾日当午，汗滴禾下锄，谁知盘中餐，粒粒皆辛苦"),
]

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

chain: RunnableSerializable = chat_prompt_template | model
print(type(chain))

# Runnable接口，invoke执行
res = chain.invoke({"history": history_data})
print(res.content)

# Runnable接口，stream执行
for chunk in chain.stream({"history": history_data}):
    print(chunk.content, end="", flush=True)

chain: RunnableSerializable = chat_prompt_template | model  #组成链
print(type(chain))

# Runnable接口，invoke执行
res = chain.invoke({"history": history_data})#组成的链在执行上有：上一个组件的输出作为下一个                                                                       #组件的输入的特性。
print(res.content)

res = chain.invoke({"history": history_data})完成的工作流程

补充运算符|的重写

前文代码中： chain = chat_prompt_template | model

在语法上使用了|运算符的重写 在 Python 中，运算符（如 +、|）的行为由类的魔法方法（内置方法）决定。例如：

  • a + b 本质调用的是 a.__add__(b)
  • a | b 本质调用的是 a.__or__(b) 只需要自行实现类的__or__方法，即可对|符号的功能进行重写。

示例：

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

    def __str__(self):
        return f"Test({self.name})"

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


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 arg in self.sequence:
            print(arg)

if __name__ == "__main__":
    a = Test("a")
    b = Test("b")
    c = Test("c")
    
    d = a | b | c
    d.run()#a b c
    print(type(d))# <class '__main__.MySequence'>

当a|b完成之后，返回的是一个 MySequence对象,之后a | b | c调用的就是MySequence中的def __or__(self, other)

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

并且在chain = prompt | model中 chain变量是RunnableSequence（RunnableSerializable子类，也继承了 Runnable 抽象基类）类型 ，因为Runnable基类内部对__or__魔术方法的改写。 同时，在后面继续使用|添加新的组件，依旧会得到RunnableSequence，这就是链的基础架构。

使用我们invoke或者stream的时候，输入输出就会变成链跑起来

StrOutputParser字符串输出解析器

为什么要用StrOutputParser？

这个代码：

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

错误的主要原因是： prompt的结果是PromptValue类型，输入给了model model的输出结果是：AIMassage.

答疑：为什么一会说prompt | model中 返回的是RunnableSequence类型，一会说返回的是AIMessage类型?

  1. 当我们说 prompt | model 返回的是 RunnableSequence 类型，这是从链的构建角度来说的。在LangChain中，使用 | 操作符连接两个Runnable对象（比如PromptTemplate和ChatTongyi）会返回一个RunnableSequence。这个RunnableSequence本身也是一个Runnable，它代表了一个序列化的调用流程。

  2. 当我们说 prompt | model 返回的是AIMessage类型，这是从调用（invoke）这个RunnableSequence的结果来说的。当我们调用 chain.invoke(...) 时，实际上会依次调用PromptTemplate和ChatTongyi，最终返回的是ChatTongyi的输出，即一个AIMessage。

所以，这两个说法并不矛盾，只是描述的是不同阶段的东西：

  • 在构建阶段，我们得到一个RunnableSequence对象（它还没有被执行）。

  • 在执行阶段（调用invoke），我们得到的是这个序列执行后的最终结果。

链的构建时（编译时/静态）

chain = prompt | model | model  # 这里返回 RunnableSequence 类型
# chain 是一个 RunnableSequence 对象
print(type(chain))  # <class 'langchain_core.runnables.base.RunnableSequence'>

链的执行时（运行时/动态）

res = chain.invoke({"lastname": "张", "gender": "女儿"})
# 这里的返回是 chain 的执行结果
print(type(res))  # <class 'langchain_core.messages.ai.AIMessage'>

模型（ChatTongyi）源码中关于invoke方法明确指定了input的类型是PromptValue,但prompt | model 返回的是AIMessage类型：

什么是StrOutputParser？

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

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

示例：直接将第一次大模型输出的结果再次传给model,再次进行大模型处理

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

history_data = [
    ("human","你来写一首诗"),
    ("ai","床前明月光，疑是地上霜，举头望明月，低头思故乡"),
    ("human", "很好再来一首"),
    ("ai", "锄禾日当午，汗滴禾下土，谁知盘中餐，粒粒皆辛苦")
]
chat_prompt_template=ChatPromptTemplate(
    [
        ("system","你是一个边塞诗人"),
        MessagesPlaceholder("history"),
        ("human","请写一首关于{topic}的诗"),
    ]
)
model=ChatTongyi(model="qwen3-max")
parser=StrOutputParser()
# clain=chat_prompt_template|model|parser|model
clain=chat_prompt_template|model|parser|model|parser#最后可以再转换一次，这样最后输出的就是字符串了，不需要通过res.content取到字符串
res=clain.invoke({"history":history_data,"topic":"边塞"})
# print(res.content)
print(res)

我们注意到

  • 在之前的代码中，我们使用了ChatPromptTemplate.from_messages方法，
  • 而在现在用的是ChatPromptTemplates方法。

实际上，这两种方式都可以，但是传入的messages格式要求略有不同。

  • 当我们使用ChatPromptTemplate.from_messages时：

        它可以接受多种消息格式，包括元组（tuple）格式，比如：
        ("system", "你是一个边塞诗人") 会被自动转换为SystemMessage
        ("human", "请写一首关于{topic}的诗") 会被自动转换为HumanMessage
        ("ai", "一些内容") 会被自动转换为AIMessage

  • 但是，如果我们直接使用ChatPromptTemplate构造函数：

        它要求传入的消息必须是BaseMessage的实例，或者已经是正确格式的列表（比如已经是消息对象）。

然而，在LangChain的较新版本中，ChatPromptTemplate的构造函数也可以接受类似from_messages的格式，但为了确保正确，我们通常使用from_messages。

JsonOutputParser&多模型执行链

在上面我们构造的链：

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

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

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

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

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

所以，我们需要完成： 将模型输出的AIMessage--> 转为字典 --> 注入第二个提示词模板中，形成新的提示词（PromptValue对象）

“AIMessage--> 转为字典”需要JsonOutputParser转化

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

history_data = [
    ("human","你来写一首诗"),
    ("ai","床前明月光，疑是地上霜，举头望明月，低头思故乡"),
    ("human", "很好再来一首"),
    ("ai", "锄禾日当午，汗滴禾下土，谁知盘中餐，粒粒皆辛苦")
]
chat_prompt_template=ChatPromptTemplate(
    [
        ("system","你是一个边塞诗人"),
        MessagesPlaceholder("history"),
        ("human","请写一首关于{topic}的诗,并封装到JSON格式返回给我，包含两个字段：'title'（诗的名字）和'content'（诗的内容），请严格按照格式返回给我"),
    ]
)
new_chat_prompt_template=PromptTemplate.from_template("诗的名字是{title},诗的内容是{content}")

model=ChatTongyi(model="qwen3-max")
str_output_parser=StrOutputParser()
json_output_parser=JsonOutputParser()

clain=chat_prompt_template|model|json_output_parser|new_chat_prompt_template|model|str_output_parser#最后可以再转换一次，这样最后输出的就是字符串了，不需要通过res.content取到字符串
res=clain.invoke({"history":history_data,"topic":"边塞"})
print(res)
  • chat_prompt_template | model: 第一个提示词传入model,model返回的就是AIMessage("{title:"XXX"，"content":"XXX"}")
  • chat_prompt_template | model | json_output_parser:变成字典{title:"XXX"，"content":"XXX"}传入new_chat_prompt_template

RunnableLambda&函数加入链

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

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

  •  RunnableLambda类是LangChain内置的，将普通函数等转换为Runnable接口实例，方便自定义函数加入chain。
  • RunnableLambda包装了这个函数，并实现了invoke方法

        语法： RunnableLambda(函数对象或lambda匿名函数)

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

str_parser = StrOutputParser()
my_func = RunnableLambda(lambda ai_msg: {"name": ai_msg.content})
model = ChatTongyi(model="qwen3-max")

first_prompt = PromptTemplate.from_template(
    "我邻居姓：{lastname}，刚生了{gender}，请起名，仅告知我名字，不要额外信息"
)
second_prompt = PromptTemplate.from_template(
    "姓名{name}，请帮我解析含义。"
)

chain = first_prompt | model | my_func | second_prompt | model | str_parser
res: str = chain.invoke({"lastname": "张", "gender": "女儿"})
print(res)
print(type(res))

函数直接入链

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

跳过RunnableLambda类，直接让函数加入链也是可以的。 因为Runnable接口类在实现__or__的时候，支持Callable接口的实例。 函数就是Callable接口的实例

如上代码示例，|符号（底层是调用__or__）组链，是支持函数加入的。

并且入链之后将函数自动转换为RunnableLambda

Memory 临时会话记忆（InMemoryChatMessageHistory）

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

model=ChatTongyi(model="qwen3-max")
prompt_template=PromptTemplate.from_template("你需要根据会话历史回答用户问题。对话历史：{chat_history},用户的提问：{input},请回答")
parser=StrOutputParser()
base_chain=prompt_template | model | parser

store={}
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,
    input_messages_key="input",
    history_messages_key="chat_history"
)

session_config={
    "configurable":{
        "session_id":"user_001"
    }
}

res=conversation_chain.invoke({"input":"小明有两个猫"},session_config)
print("第一次")
res=conversation_chain.invoke({"input":"小明有一个小狗"},session_config)
print("第二次")
res=conversation_chain.invoke({"input":"小明有一个小猪，请问一共有几只宠物"},session_config)
print(res)

1.当调用conversation_chain.invoke(input, config)时，RunnableWithMessageHistory会做以下事情：

  • 从配置中获取session_id
  • 然后调用get_history(session_id)获取到ChatMessageHistory对象（即InMemoryChatMessageHistory实例）。
  • RunnableWithMessageHistory会自动将这个历史记录对象中的消息传递给提示模板的chat_history变量
  •  然后，它会把历史记录（chat_history变量）和用户输入（input）一起传递给基础链（base_chain）。

2. 在基础链中，首先由PromptTemplate接收这两个变量（chat_history和input），并填充模板。
3. 填充后的模板被传递给模型，模型生成回答。
4. 最后，输出解析器将模型的输出解析为字符串。
5. 在调用结束后，RunnableWithMessageHistory会将当前轮次的用户输入和模型输出添加到历史记录中（即保存到store[session_id]中）。

我们稍微写改一下代码，将每一次的提示词模板打印出来

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

model=ChatTongyi(model="qwen3-max")
prompt_template=PromptTemplate.from_template("你需要根据会话历史回答用户问题。对话历史：{chat_history},用户的提问：{input},请回答")
parser=StrOutputParser()

def print_prompt(full_template):
    print("*"*50)
    print(full_template.to_string())
    print("*"*50)
    return full_template

base_chain=prompt_template |print_prompt | model | parser

store={}
def get_history(session_id):
    if session_id not in store:
        store[session_id]=InMemoryChatMessageHistory()
    return store[session_id]   #get_history返回的是一个InMemoryChatMessageHistory对象，

conversation_chain=RunnableWithMessageHistory(
    base_chain,
    get_history,
    input_messages_key="input",
    history_messages_key="chat_history"
)

session_config={
    "configurable":{
        "session_id":"user_001"
    }
}

res=conversation_chain.invoke({"input":"小明有两个猫"},session_config)
print("第1次执行：", res)
res=conversation_chain.invoke({"input":"小明有一个小狗"},session_config)
print("第2次执行：", res)
res=conversation_chain.invoke({"input":"小明有一个小猪，请问一共有几只宠物"},session_config)
print("第3次执行：", res)

输出：

**************************************************
你需要根据会话历史回答用户问题。对话历史：[],用户的提问：小明有两个猫,请回答
**************************************************
第1次执行： 小明有两只猫。
**************************************************
你需要根据会话历史回答用户问题。对话历史：[HumanMessage(content='小明有两个猫', additional_kwargs={}, response_metadata={}), AIMessage(content='小明有两只猫。', additional_kwargs={}, response_metadata={}, tool_calls=[], invalid_tool_calls=[])],用户的提问：小明有一个小狗,请回答
**************************************************
第2次执行： 小明有一只小狗。
**************************************************
你需要根据会话历史回答用户问题。对话历史：[HumanMessage(content='小明有两个猫', additional_kwargs={}, response_metadata={}), AIMessage(content='小明有两只猫。', additional_kwargs={}, response_metadata={}, tool_calls=[], invalid_tool_calls=[]), HumanMessage(content='小明有一个小狗', additional_kwargs={}, response_metadata={}), AIMessage(content='小明有一只小狗。', additional_kwargs={}, response_metadata={}, tool_calls=[], invalid_tool_calls=[])],用户的提问：小明有一个小猪，请问一共有几只宠物,请回答
**************************************************
第3次执行： 小明有两只猫、一只小狗和一只小猪，一共有 2 + 1 + 1 = 4 只宠物。

进程已结束,退出代码0

我们注意到，在打印的提示词模板中，历史记录部分并不是我们通常见到的字符串格式（如"Human: ...  AI: ..."），而是一个Python列表的字符串表示，其中包含的是HumanMessage和AIMessage对象。这是因为默认情况下，RunnableWithMessageHistory在将历史记录传递给基础链时，并没有将消息列表转换为字符串，而是直接将消息列表传递给了提示模板。而提示模板在格式化时，将整个列表转换成了字符串（通过调用str()），所以我们就看到了列表的字符串表示。

所以我们使用聊天提示词模板（ChatPromptTemplate）会更加规范，功能上没有区别

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

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

def print_prompt(full_template):
    print("*"*50)
    print(full_template.to_string())#.to_string()转换格式: 将 ChatPromptValue 对象转换为人类可读的字符串
    print("*"*50)
    return full_template

base_chain=chat_prompt_template |print_prompt | model | parser

store={}
def get_history(session_id):
    if session_id not in store:
        store[session_id]=InMemoryChatMessageHistory()
    return store[session_id]   #get_history返回的是一个InMemoryChatMessageHistory对象，

conversation_chain=RunnableWithMessageHistory(
    base_chain,
    get_history,
    input_messages_key="input",
    history_messages_key="chat_history"
)

session_config={
    "configurable":{
        "session_id":"user_001"
    }
}

res=conversation_chain.invoke({"input":"小明有两个猫"},session_config)
print("第1次执行：", res)
res=conversation_chain.invoke({"input":"小明有一个小狗"},session_config)
print("第2次执行：", res)
res=conversation_chain.invoke({"input":"小明有一个小猪，请问一共有几只宠物"},session_config)
print("第3次执行：", res)

输出：

**************************************************
System: 你需要根据会话历史回答用户问题
Human: 用户的提问：小明有两个猫,请回答
**************************************************
第1次执行： 小明有两只猫。
**************************************************
System: 你需要根据会话历史回答用户问题
Human: 小明有两个猫
AI: 小明有两只猫。
Human: 用户的提问：小明有一个小狗,请回答
**************************************************
第2次执行： 小明有一只小狗。
**************************************************
System: 你需要根据会话历史回答用户问题
Human: 小明有两个猫
AI: 小明有两只猫。
Human: 小明有一个小狗
AI: 小明有一只小狗。
Human: 用户的提问：小明有一个小猪，请问一共有几只宠物,请回答
**************************************************
第3次执行： 小明有两只猫、一只小狗和一只小猪，一共有：

2（猫） + 1（狗） + 1（小猪） = **4只宠物**。

进程已结束,退出代码0

Memory 长期会话记忆 自实现FileChatMessageHistory

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

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

import os, json
from typing import Sequence

from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import BaseMessage, message_to_dict, messages_from_dict


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


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

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

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

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

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

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

完整代码：

import os, json
from typing import Sequence

from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import BaseMessage, message_to_dict, messages_from_dict

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


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

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

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

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

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

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

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

def print_prompt(full_template):
    print("*"*50)
    print(full_template.to_string())#.to_string()转换格式: 将 ChatPromptValue 对象转换为人类可读的字符串
    print("*"*50)
    return full_template

base_chain=chat_prompt_template |print_prompt | model | parser

store={}
def get_history(session_id):
    if session_id not in store:
        store[session_id]=FileChatMessageHistory(session_id, "./chat_history")
    return store[session_id]   #get_history返回的是一个InMemoryChatMessageHistory对象，

conversation_chain=RunnableWithMessageHistory(
    base_chain,
    get_history,
    input_messages_key="input",
    history_messages_key="chat_history"
)

session_config={
    "configurable":{
        "session_id":"user_001"
    }
}

# res=conversation_chain.invoke({"input":"小明有两个猫"},session_config)
# print("第1次执行：", res)
# res=conversation_chain.invoke({"input":"小明有一个小狗"},session_config)
# print("第2次执行：", res)
res=conversation_chain.invoke({"input":"小明有一个小猪，请问一共有几只宠物"},session_config)
print("第3次执行：", res)

当我们运行之后，会自动创建文件夹保存记录

Document loaders: 文档加载器

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

Class Document数据类型，是LangChain内文档的统一载体，所有文档加载器最终返回此类的实例。Document类其核心记录了： page_content：文档内容 metadata：文档元数据（字典）

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

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

文档加载器（内置或自行实现）需实现BaseLoader接口中的load()和lazy_load()

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

简单的学习如下几个常用的文档加载器： CSVLoader JSONLoader PDFLoader

CSVLoader

了解CSVLoader里面的参数：

 “delimiter”: "|" 指定每个属性的分隔符是|

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

这个时候就要指定分隔符是“|”，因为默认是“,”

“quotechar”: ' " '     指定包裹有分割符的引号是双引号

"fieldnames": ["name", "age", "gender"]  当文件信息中无表头使用字段列表

        有表头的时候不能使用

CSVLoader的使用：

from langchain_community.document_loaders import CSVLoader
loader=CSVLoader(file_path="./data/stu.csv",
                 encoding="utf-8")

# .load()批量加载数据, 返回[Ducument,Ducument,Ducument...]
ducuments=loader.load()
print(ducuments)

可以看出用返回的数据是列表里面套Document类型[Ducument,Ducument,Ducument...]

使用lazy_load()：

from langchain_community.document_loaders import CSVLoader
loader=CSVLoader(file_path="./data/stu.csv",
                 encoding="utf-8")


for ducument in loader.lazy_load():
    print(ducument)

JSONLoader

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

 跨平台的json解析工具jq

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

  • jq是一个跨平台的json解析工具，LangChain底层对JSON的解析就是基于jq工具实现的。
  • 将JSON数据的信息抽取出来，封装为Document对象，抽取的时候依赖jq_schema语法。

  jq_schema语法：

        

抽取上面的json数组

   规则        ：                    .表示整个JSON对象（根）              []表示数组

.name表示抽取                           周杰轮

.hobby表示抽取                        爱好数组

.hobby[1]或.hobby.[1]表示抽取  跳

.other.addr表示抽取                  地址深圳

.[].得到  3个字典

.[].name 表示抽取 全部的name，即得到3个name信息

这是一个Jsonlines文件

.得到  3个字典

.name 表示抽取 全部的name，即得到3个name信息

了解JsonLoader里面的参数

#文件./data/stus.json
#[
#   {"name": "周杰轮", "age": 11, "gender": "男"},
#    {"name": "蔡依临", "age": 12, "gender": "女"},
#    {"name": "王力鸿", "age": 11, "gender": "男"}
#]

from langchain_community.document_loaders import JSONLoader
loader = JSONLoader(
    file_path="./data/stus.json"
    , jq_schema=".[].name",
    text_content=False,

)
documents = loader.load()
print(documents)
#输出：
[Document(metadata={'source': 'C:\\Program Files (x86)\\ruanjian\\PycharmProjects\\AI大
模型RAG与智能体开发\\LangChainRAG开发\\data\\stus.json', 'seq_num': 1}, page_content='周杰
轮'), Document(metadata={'source': 'C:\\Program Files 
(x86)\\ruanjian\\PycharmProjects\\AI大模型RAG与智能体开发\\LangChainRAG开发
\\data\\stus.json', 'seq_num': 2}, page_content='蔡依临'), Document(metadata={'source': 
'C:\\Program Files (x86)\\ruanjian\\PycharmProjects\\AI大模型RAG与智能体开发\\LangChainRAG
开发\\data\\stus.json', 'seq_num': 3}, page_content='王力鸿')]


#文件./data/stus.json
#{
#    "name": "周杰轮",
#    "age": 11,
#    "hobby": ["唱", "跳", "RAP"],
#    "other": {
#        "addr": "深圳",
#        "tel": "12332112321"
#    }
#}

from langchain_community.document_loaders import JSONLoader
loader=JSONLoader(
    file_path="./data/stu.json"
    ,jq_schema=".name"
)
documents=loader.load()
print(documents)
#输出：
[Document(metadata={'source': 'C:\\Program Files (x86)\\ruanjian\\PycharmProjects\\AI大模
型RAG与智能体开发\\LangChainRAG开发\\data\\stu.json', 'seq_num': 1}, page_content='周杰
轮')]

TextLoader

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

使用方式和上面加载器类似：

from langchain_community.document_loaders import TextLoader

loader = TextLoader(
    "xxx.txt",
    encoding="utf-8",
)
docs = loader.load()
print(docs)
print(len(docs))

如果文档很大 加载到一个Document对象中是否不太合适？

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

被分割之后就有很多Document对象了，[Document、Document、Document...]

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

loader=TextLoader("./data/Python基础语法.txt",encoding="utf-8")
doc=loader.load()
text_splitter=RecursiveCharacterTextSplitter(
    #分段的最大字符数
    chunk_size=500,
    # 分段之间允许重叠的字符数
    chunk_overlap=50,
    # 允许的分隔符列表(可以根据自己的文本添加一些其他的)
    separators=["\n\n", "\n", "。", "！", "？", " ", "!", ".", ""],
)
spilt_docs=text_splitter.split_documents(doc)
for i in spilt_docs:
    print(i)
    print("*"*50)

PyPDFLoader

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

所以，需要安装它： pip install pypdf

使用如下：

from langchain_community.document_loaders import PyPDFLoader
loader=PyPDFLoader(
    file_path="./data/pdf2.pdf",
    mode="single",#读取模式，可选page（按页面划分不同Document）
                           # 和single（单个Document）
    password="itheima"# 如果这个pdf文件双击打开需要密码
)
for i in loader.lazy_load():
    print(i)
#page_content='天天开心，健康快乐。' metadata={'producer': '', 
# 'creator': 'WPS 文字', 'creationdate': '2026-01-13T18:35:32+08:00',
# 'author': '曹宇[囧]', 'comments': '', 'company': '', 'keywords': '', 
# 'moddate': '2026-01-13T18:36:36+08:00', 'sourcemodified': "D:20260113183532+08'00'",
# 'subject': '', 'title': '', 'trapped': '/False', 'source': './data/pdf2.pdf', 'total_pages': 1}

Vector stores 向量存储

如图，这是一个典型的向量存储应用，也即是典型的RAG流程。

这部分开发主要涉及到：

  • 如何文本转向量（前文已经学习）

                                           

  • 创建向量存储，基于向量存储完成：

                        存入向量 add_documents

                         删除向量 delete

                        向量检索 similarity_search

内置向量存储InMemoryVectorStore（官方提供的，存在内存里，程序一停就没了）

           

使用如下：

CSV文件

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

# 准备要加载数据
loader=CSVLoader(file_path="./data/info.csv",
                 encoding="utf-8")
documents=loader.load()
# 创建向量存储器
inmemory_vector_store=InMemoryVectorStore(
    embedding=DashScopeEmbeddings()
)


#添加文档到向量存储
inmemory_vector_store.add_documents(
    documents=documents,#被添加的文件[document,document,document...]
    ids=["id"+str(i) for i in range(1,len(documents)+1)],#给文件添加id
)
#检索向量
result=inmemory_vector_store.similarity_search(
    "python是不是简单易学呀？",
    k=2 #返回最相似的k个文件
)
for i in result:
    print(i)
# 输出：
# page_content='source: 传智教育
# info: Python学起来很简单的' metadata={'source': './data/info.csv', 'row': 4}
# page_content='source: 黑马程序员
# info: 学习Python键盘敲烂月薪过万' metadata={'source': './data/info.csv', 'row': 5}

 外部向量存储Chroma的使用（是持久化的存储）

要确保已经安装了chroma和chroma-db，没有安装的模块，请使用pip安装

          

使用如下：

from langchain_chroma import Chroma
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.document_loaders import CSVLoader
#要确保已经安装了chroma和chroma-db，没有安装的模块，请使用pip安装

# 准备要加载数据
loader=CSVLoader(file_path="./data/info.csv",
                 encoding="utf-8",
                 source_column="source",#指定本条数据的来源
                 )
documents=loader.load()
# 创建向量存储器
vector_store=Chroma(
    embedding_function=DashScopeEmbeddings(),
    collection_name="test",#给当前向量存储起个名字，类似于数据库的表名
    persist_directory="./chroma_db"#指定数据库存放的路径
)


#添加文档到向量存储
vector_store.add_documents(
    documents=documents,#被添加的文件[document,document,document...]
    ids=["id"+str(i) for i in range(1,len(documents)+1)],#给文件添加id
)
#检索向量
result=vector_store.similarity_search(
    "python是不是简单易学呀？",
    k=2 #返回最相似的k个文件
)
for i in result:
    print(i)

运行完了之后会多一个文件夹chroma.sqlite3是向量存储的数据库

双击chroma.sqlite3在弹出来的框框下载驱动，然后点击测试连接，就可以看到里面的数据

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

里面的source_column 参数在 CSVLoader 中用于指定 CSV 文件中的某一列作为文档的“来源”（source）元数据。

        当你加载 CSV 文件时，每一行会生成一个 Document 对象，其内容（page_content）默认由所有列（或指定的内容列）拼接而成，而 source_column 指定的列值会被存入该文档的元数据（metadata）字典中，键名为 "source"。

作用与好处：

  1. 记录来源信息：例如，如果 CSV 中有一列是 source，里面存放了文档的 URL、文件路径或数据来源名称，那么加载后，每个文档的 metadata 中就会包含 "source": "实际值"。

  2. 便于溯源与过滤：在检索时，你可以通过元数据过滤结果，或者查看文档来自哪个具体文件/来源，方便后续处理，如下：

#检索向量
result=vector_store.similarity_search(
    "python是不是简单易学呀？",
    k=2, #返回最相似的k个文件
    filter={"source": "黑马程序员"}
)

这样我们检索的内容就不会给其他来源的

基于向量检索构建提示词

from langchain_community.chat_models import ChatTongyi
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings


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

prompt_template=ChatPromptTemplate.from_messages([
    ("system","以我提供的已知的参考资料为主.参考资料:{context}"),
    ("human","用户的提问：{input},请回答")
])


#准备向量数据库的内容
vector_store=InMemoryVectorStore(
    embedding=DashScopeEmbeddings()
)
vector_store.add_texts(["Python是简单易学的语言","Python是好用的","Python是好的语言"])#大文件还是要用.add_doucuments()方法

input_text="Python好学吗？"

result=vector_store.similarity_search(
    query=input_text,
    k=2
)
reference_text=[i.page_content for i in result]

def print_prompt(prompt_template):
    print(prompt_template.to_string())
    print("*"*50)
    return prompt_template

chain=prompt_template |print_prompt| model | parser
res=chain.invoke({"context":reference_text,"input":input_text})
print(res)
输出：
System: 以我提供的已知的参考资料为主.参考资料:['Python是简单易学的语言', 'Python是好的语言']
Human: 用户的提问：Python好学吗？,请回答
**************************************************
是的，Python好学。根据参考资料，Python是简单易学的语言，同时也被认为是一门好的语言，非常适合初学者入门编程。

向量检索入链

from langchain_community.chat_models import ChatTongyi
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.runnables import RunnablePassthrough


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

prompt_template=ChatPromptTemplate.from_messages([
    ("system","以我提供的已知的参考资料为主.参考资料:{context}"),
    ("human","用户的提问：{input},请回答")
])

def print_prompt(prompt_template):
    print(prompt_template.to_string())
    print("*"*50)
    return prompt_template

#准备向量数据库的内容
vector_store=InMemoryVectorStore(
    embedding=DashScopeEmbeddings()
)
vector_store.add_texts(["Python是简单易学的语言","Python是好用的","Python是好的语言"])#大文件还是要用.add_doucuments()方法

input_text="Python好学吗？"


#langchain中向量存储对象，中的as_retriever方法，可以将向量存储对象转换成Runnable接口的子类实例对象
retriever=vector_store.as_retriever(
    search_type="similarity",
    search_kwargs={"k":2}
)

#将检索到的内容组成字符串
def format_func(docs:list[Document]):
    if not docs:
        return "无相关资料"
    formatted_str="["
    for doc in docs:
        formatted_str += f"{doc.page_content},"
    formatted_str += "]"
    return formatted_str

chain=(
        {"input":RunnablePassthrough(),"context":retriever | format_func} | prompt_template |print_prompt| model | parser
)
res=chain.invoke(input_text)
print(res)

与上面所作出的改变有：

1.通过向量存储对象（vector_store）中的as_retriever方法，可以将向量存储对象转换成Runnable接口的子类实例对象

2.查看VectorStoreRetriever和ChatPromptTemplate期望输入输出

用完as_retriever方法后返回的是VectorStoreRetriever对象，查看VectorStoreRetriever类里的方法

可以看到invoke函数期望

  • 输入：用户的提问 str
  • 输出：向量库的检索结果 list[Document]

查看ChatPromptTemplate类的方法

可以看到invoke函数期望

  • 输入：用户的提问 + 向量库的检索结果 dict
  • 输出：完整的提示词模板 PromptValue

3.将查询到的文档转换成字符串

    4.构造链chain

    RunnablePassthrough用来接收输入，并原样返回输入，本身不做任何改变。

    所以输入的数据不但传入了retriever，也传入了RunnablePassthrough（）

    # python # pytorch
    Logo
    2048 AI社区

    有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

    更多推荐

    cover

    [ 开源 ] FastAPI + LangGraph 实战智能客服 Agent：从工单分类到自动回复与业务回写 附github

    avatar 2048 AI社区

    大厂Java面试实战：Spring Boot/Cloud、Redis/Kafka、JVM调优与Spring AI RAG（内容社区UGC+AIGC客服场景）

    以“内容社区+UGC+AIGC智能客服”为业务背景，模拟大厂Java面试：严肃面试官循序渐进提问Spring Boot/Cloud、JVM、Redis、Kafka、数据库、观测、CI/CD与Spring AI/RAG/Agent等，小Y简单题会答复杂题含糊。文末给出每题详细答案与落地方案，便于小白系统学习。

    avatar 2048 AI社区
    cover

    一文分清Agent与Skill

    avatar 2048 AI社区