RAG介绍

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

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

RAG(Retrieval-AUgmented Generation)即检索增强生成，为大模型提供了从特定数据源检索到的信息，以此来修正和补充生成的答案。

  1. 可以总结为一个公式: RAG=检索技术+LLM 提示

  1. 工作流程图解:

  1. RAG标准流程:

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

RAG 标准流程由索引(Indexing)、检索(Retriever)和生成(Generation)三个核心阶段组成。

  1. 索引阶段，通过处理多种来源多种格式的文档提取其中文本，将其切分为标准长度的文本块(chunk)，并进行嵌入向量化(embedding)，向量存储在向量数据库(vector database)中。
  • 加载文件
  • 内容提取
  • 文本分割，形成chunk
  • 文本向量化
  • 存向量数据库
  1. 检索阶段，用户输入的查询(query)被转化为向量表示，通过相似度匹配从向量数据库中检索出最相关的文本块。
  • query向量化
  • 在文本向量中匹配出与问句向量相似的top_k个
  1. 生成阶段，检索到的相关文本与原始查询共同构成提示词(Prompt)，输入大语言模型(LLM)，生成精确且具备上下文关联的回答匹配
  • 出的文本作为上下文和问题一起添加到prompt中
  • 提交给LLM生成答案

模型本质上就是用户输入，模型给出输出，用户能做的就是在输入上做功夫。

  1. RAG就是在向模型提问之前基于已有的知识库或文档内容做检索，确保向模型提问的内容更精准以及包含足够的信息量用以提供给模型。
  2. RAG的核心价值:
  • 解决知识实效性问题: 大模型的训练数据有截止时间，RAG 可以接入最新文档(如公司财报、政策文件)，让模型输出“与时俱进”
  • 降低模型幻觉: 模型的回答基于检索到的事实性资料，而非纯靠自身记忆，大幅减少编造信息的概率。
  • 无需重新训练模型: 相比微调(Fine-tuning)，RAG只需更新知识库，成本更低、效率更高。

向量介绍

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

  1. 离线流程: 知识和信息 -> 向量嵌入(向量化) -> 存入向量库
  2. 在线流程: 用户的提问 ->向量嵌入(向量化) -> 在向量库中匹配

向量(Vector)就是文本的 “数学身份证”: 它把一段文字的语义信息，转换成一串固定长度的数字列表，让计算机能 “看懂”文字的含义并做相似度计算。

  1. 简单来说，就是让计算机更方便的理解不同的文本内容，是否表述的是一个意思。

向量的嵌入

向量的计算(文本嵌入过程)，我们一般选用合适的文本嵌入模型来完成。

  1. 文本嵌入模型(如text-embedding-v1)通过深度学习等技术，从文本提取语义特征并映射为固定长度的数字序列。
  2. 把文本计算为如下的数字序列, 让算机理解自然语言的语意

向量的匹配

在向量匹配的过程中，如何识别2段文本是否表述相似的含义，主要可以通过如余弦相似度等算法来完成。

  1. 比如(下列案例中向量为示例，仅描述概念，非真实向量):

  1. 通过余弦相似度算法可以计算得到: A和B相似度0.999789，A和C相似度0.361446
  2. 由此可通过精确的数学计算，去匹配2段文本是否描述同一个意思，提高语义匹配的效率和精度。

匹配精度的控制

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

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

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

余弦相似度

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

  1. 如何体现向量的方向和长度呢?
  2. 以一维向量为例:

  1. 以二维向量为例:

  • 余弦相似度主要匹配的就是: 同向(无所谓长度)
  • 我们能直接发现[0.5,0.5]和[0.7,0.7]是同向不同长
  • 那计算机如何判定就依赖余弦相似度算法了。
  1. 3维乃至更高纬度难以描述，但概念一致

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

以: A[0.5, 0.5]、B[0.7, 0.7]、C[0.7, 0.5]、D[-0.6,-0.5]为例

如AB之间的余弦相似度为: AB点积÷(A模长*B模长)

代码实现: 计算两个向量的余弦相似度

import numpy as np

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

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

测试数据:
A: [0.5, 0.5]
B: [0.7, 0.7]
C: [0.7, 0.5]
D: [-0.6, -0.5]

补充: numpy在安装langchain时自动安装
"""


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

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

    return dot_sum


def get_norm(vec):
    """
    计算单个向量的模长: 对向量的每个数字求平方再求和再开根号
    """
    sum_square = 0
    for v in vec:
        sum_square += v * v
    # numpy.sqrt()完成开根号
    return np.sqrt(sum_square)


def cosine_similarity(vec_a, vec_b):
    """
    计算两个向量的余弦相似度: 2个向量的点积 除以 2个向量模长的乘积
    """
    return get_dot(vec_a, vec_b) / (get_norm(vec_a) * get_norm(vec_b))


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

LangChain

安装

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

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

  1. LangChain是后续学习RAG开发的主力框架

安装命令

  1. 安装: pip install langchain langchain-community langchain-ollama dashscope chromadb
  • langchain: 核心包
  • langchain-community:社区支持包，提供了更多的第三方模型调用(我们用的阿里云千问模型就需要这个包)
  • langchain-ollama:Ollama支持包，支持调用Ollama托管部署的本地模型
  • dashscope:阿里云通义千问的Python SDK
  • chromadb:轻量向量数据库(后续使用)

  1. 验证

进入python环境, 通过import langchain验证安装结果, 没有报错说明安装成功

组件

Models

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

  1. LangChain目前支持三种类型的模型: LLMs(大语言模型)、Chat Models(聊天模型)、Embeddings Models (嵌入模型)
  • LLMs: 是技术范畴的统称，指基于大参数量、海量文本训练的Transformer架构模型，核心能力是理解和生成自然语言，主要服务于文本生成场景
  • 聊天模型: 是应用范畴的细分，是专为对话场景优化的LLMs，核心能力是模拟人类对话的轮次交互，主要服务于聊天场景
  • 文本嵌入模型: 文本嵌入模型接收文本作为输入，得到文本的向量.
  1. LangChain支持的三类模型，它们的使用场景不同，输入和输出不同，开发者需要根据项目需要选择相应。
  2. 我们所用的阿里云通义千问系列主要来自于langchain_community包

阿里云大语言模型的访问

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

  1. https://huggingface.co/models
  2. https://modelscope.cn/models
  3. 同时LangChain支持对许多模型的调用，以通义千问为例
"""
使用langChain访问千问大模型
"""
from langchain_community.llms.tongyi import Tongyi

# 不使用qwen3-max, 因为qwen3-max是聊天模型,qwen-max是大预言模型
model = Tongyi(model="qwen-max", api_key="")

# 向模型提问
res =  model.invoke(input="你是谁?你能做什么?")
print(res)

  1. 如果要访问本地Ollama的模型，简单更改一下代码。
  2. 通过langchain_ollama包导入OllamaLLM类即可 (请确保Ollama已经启动并提前下载好要使用的模型)

  1. models的流式输出
  • 如果需要流式输出结果，需要将模型的invoke方法改为stream方法即可。
  • invoke方法: 一次型返回完整结果
  • stream方法: 逐段返回结果，流式输出
"""
使用langChain访问千问大模型
"""
from langchain_community.llms.tongyi import Tongyi

# 不使用qwen3-max, 因为qwen3-max是聊天模型,qwen-max是大预言模型
model = Tongyi(model="qwen-max", api_key="")


# 向模型提问 (流式输出)
res =  model.stream(input="你是谁?你能做什么?")
for chunk in res:
    print(chunk, end="", flush=True)

Chat Models (聊天模型)

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

  1. AIMessage: 就是 AI 输出的消息，可以是针对问题的回答. (OpenAI库中的assistant角色)
  2. HumanMessage: 人类消息就是用户信息，由人给出的信息发送给LLMs的提示信息，比如”实现一个快速排序方法”. (OpenAI库中的user角色)
  3. SystemMessage: 可以用于指定模型具体所处的环境和背景，如角色扮演等。你可以在这里给出具体的指示，比如“作为一个代码专家”，或者“返回json格式”.(OpenAI库中的system角色)
  4. 演示单独使用HumanMessage

  1. 演示SystemMessage + HumanMessage

  1. 演示SystemMessage +HumanMessage + AIMessage

  1. 三种角色的使用
"""
演示聊天模型中 SystemMessage, HumanMessage, AIMessage 三种角色的使用
"""
from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage

# 初始化模型
chat = ChatTongyi(model="qwen3-max", api_key="")

# 准备消息list
messages = [
    SystemMessage(content="你是一名来自边塞的诗人"),
    HumanMessage(content="给我写一首唐诗"),
    AIMessage(content="锄禾日当午,汗滴禾下土,谁之盘中餐,粒粒皆辛苦."),
    HumanMessage(content="参考上首诗的格式,再写一首"),
]

for chunk in chat.stream(messages):
    print(chunk.content, end="", flush=True)
  1. 使用本地Ollama的聊天模型

  1. SystemMessage、HumanMessage、AIMessage可以有如下的简写形式

  • 通过2元元组封装信息: 第一个元素为角色(system/human/ai) 第二个元素为内容
  1. 区别和优势在于，使用类对象的方式，如下:

Embeddings Models(文本嵌入模型)

Embeddings Models嵌入模型的特点将字符串作为输入，返回一个浮点数的列表(向量)。

  1. 在NLP中，Embedding的作用就是将数据进行文本向量化。
  2. 阿里云千问模型访问方式:
"""
演示嵌入模型的访问
"""
from langchain_community.embeddings import DashScopeEmbeddings

# 创建模型对象,不传入model默认使用 text-embeddings-v1
model = DashScopeEmbeddings(dashscope_api_key="")

# 向模型提问
print(model.embed_query("我喜欢你"))
print(model.embed_documents(["我喜欢你","我稀饭你","晚上吃啥"]))

  1. 本地Ollama模型访问方式:
  • 通过langchain_ollama导入OllamaEmbeddings使用，其余不变。

模型使用总结

目前所掌握的LangChain API如下:

提示词模板

通用提示词模板

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

  1. PromptTemplate表示提示词模板，可以构建一个自定义的基础提示词模板，支持变量的注入，最终生成所需的提示词。
  2. 提示词思想分为两类, 预训练和后训练, 通过模板适用于预训练场景(zero-shot)
  • zero-shot思想下，可以基于PromptTemplate直接完成。
  • 可以理解为预训练思想, 核心就是信任模型的推理能力, 让模型发挥自己的能力, 生成效果完全依赖模型的能力
  • few-shot思想下，需要更换为FewShotPromptTemplate(后续学习)
  • 可以理解为后训练思想, 核心就是不信任模型的推理能力, 通过少量的示例来, 让模型的生成效果更可控
  1. 基础语法

  1. 代码示例
"""
标准写法
"""
from langchain_community.llms.tongyi import Tongyi
from langchain_core.prompts import PromptTemplate

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

# 调用.format方法注入信息
prompt_text = prompt_template.format(lastname="张", gender="男")
model = Tongyi(model="qwen-max", api_key="sk-0af497cdd462495aa2f08dddaead1ded")

# 传入信息,调用模型回答问题
res = model.invoke(input=prompt_text)
print(res)


"""
链式调用
"""
from langchain_community.llms.tongyi import Tongyi
from langchain_core.prompts import PromptTemplate

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

model = Tongyi(model="qwen-max", api_key="sk-0af497cdd462495aa2f08dddaead1ded")

# 创建链
chain = prompt_template | model

# 链式调用
res = chain.invoke(input = {"lastname": "张", "gender": "男"})

print(res)

  1. 使用PromptTemplate还不如自己手动拼接字符串?
  • 使用Template模板构建提示词，在大型工程中更容易做标准化模板
  • Template模板类，支持LangChian框架的链式调用 (Runnable接口，后续学习)
    • PromptTemplate
    • FewShotPromptTemplate(后续学习)
    • ChatPromptTemplate(后续学习)

示例提示词模板

few-shot思想下，需要使用FewShotPromptTemplate管理提示词

  1. 基础语法

  • 参数:说明
  • examples: 示例数据，list，内套字典
  • example_prompt: 示例数据的提示词模板
  • prefix: 组装提示词，示例数据前内容
  • suffix: 组装提示词，示例数据后内容
  • input_variables: 列表，注入的变量列表
  1. 代码示例
from langchain_core.prompts import PromptTemplate, FewShotPromptTemplate

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

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

# 提示词对象
few_shot_template = FewShotPromptTemplate(
    example_prompt=example_template, # 示例模板
    examples=example_data,           # 示例数据
    prefix="告诉我单词的反义词,示例如下",    # 示例前缀
    suffix="基于前面的示例, 告诉我{input_word}的反义词是?",  # 示例后缀
    input_variables=["input_word"],       # 输入变量
)

# 获得最终的提示词
prompt_text = few_shot_template.invoke(input={"input_word": "大"}).to_string()
print(prompt_text)

  1. 基于提示词模板向模型提问
from langchain_community.llms.tongyi import Tongyi
from langchain_core.prompts import PromptTemplate, FewShotPromptTemplate

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

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

# 提示词对象
few_shot_template = FewShotPromptTemplate(
    example_prompt=example_template, # 示例模板
    examples=example_data,           # 示例数据
    prefix="告诉我单词的反义词,示例如下",    # 示例前缀
    suffix="基于前面的示例, 告诉我{input_word}的反义词是?",  # 示例后缀
    input_variables=["input_word"],       # 输入变量
)

# 获得最终的提示词
prompt_text = few_shot_template.invoke(input={"input_word": "大"}).to_string()
print(prompt_text)


# 模型对象
model = Tongyi(model="qwen-max", api_key="sk-0af497cdd462495aa2f08dddaead1ded")
res = model.invoke(input=prompt_text)
print(res)

模板类的format和invoke方法

在PromptTemplate(通用提示词模板)和FewShotPromptTemplate (FewShot提示词模板)的使用中，我们使用了如下:

  1. PromptTemplate、FewShotPromptTemplate、ChatPromptTemplate (后续学习) 都拥有format和invoke这2类方法

  1. format和invoke的区别在于

from langchain_core.prompts import PromptTemplate

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

res = template.format(lastname="张三", hobby="看电影")
print(res, type(res))


# 如果把template.invoke().to_string()之后,res和res2就一样了
# 后面使用invoke方法会更多一些
res2 = template.invoke({"lastname": "张杰", "hobby": "听歌"})
print(res2, type(res2))

历史提示词模板

ChatPromptTemplate支持历史会话信息

  • PromptTemplate: 通用提示词模板，支持动态注入信息。
  • FewShotPromptTemplate: 支持基于模板注入任意数量的示例信息。
  • ChatPromptTemplate: 支持注入任意数量的历史会话信息。
  1. 通过from_messages方法，从列表中获取多轮次会话作为聊天的基础模板
  • 前面PromptTemplate类用的from_template仅能接入一条消息，而from_messages可以接入一个list的消息

  1. 历史会话信息并不是静态的(固定的)，而是随着对话的进行不停地积攒，即动态的。
  2. 所以，历史会话信息需要支持动态注入

  1. 示例代码
from langchain_community.chat_models import ChatTongyi
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

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

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

# 创建历史提示词模板
prompt_text = chat_prompt_template.invoke({"history": history_data}).to_string()
print(prompt_text)

# 创建模型对象
model = ChatTongyi(model="qwen3-max", api_key="sk-0af497cdd462495aa2f08dddaead1ded")
res = model.invoke(prompt_text)
print(res.content)

chain链

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

chain = prompt_template | model

  1. 核心前提: 即Runnable子类对象才能入链(以及Callable、Mapping接口子类对象也可加入(后续了解用的不多))。
  2. 我们目前所学习到的组件，均是Runnable接口的子类，如下类的继承关系:

  1. 语法

  • 通过 | 链接提示词模板对象和模型对象
  • 返回值chain对象是RunnableSerializable对象
    • 是Runnable接口的直接子类
    • 也是绝大多数组件的父类
  • 通过invoke或stream进行阻塞执行或流式执行
  • 组成的链在执行上有: 上一个组件的输出作为下一个组件的输入的特性。
  • 所以有如下执行流程:

  1. 代码示例
from langchain_community.chat_models import ChatTongyi
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 创建提示词模板
chat_prompt_template = ChatPromptTemplate.from_messages([
    ("system", "你是一个边塞诗人,可以作诗."),
    MessagesPlaceholder("history"),
    ("human", "再写一首唐诗"),
])

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

# 创建模型对象
model = ChatTongyi(model="qwen3-max", api_key="sk-0af497cdd462495aa2f08dddaead1ded")

# 创建链条
chain = chat_prompt_template | model

# 链式调用
# 组成链的要求: 每一个组件都是Runnable接口的子类
# 流程说明: 先把 {"history": history_data}作为参数给到 chat_prompt_template 得到提示词, 然后调用model得到模型结果
# res = chain.invoke({"history": history_data})
# print(res.content)


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

前文代码中: chain = chat_prompt_template | model

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

  1. 在Python中，运算符(如 +、|)的行为由类的魔法方法决定。例如:
  • a+b 本质调用的是a.__add__(b)
  • a | b 本质调用的是a.__or__(b)
  1. 只需要自行实现类的__or__方法，即可对|符号的功能进行重写。
  2. 示例:
  • 让a | b | c 的代码得到一个自定义的类对象 (类似列表即[a，b，c])
  • 调用run方法依次输出a、b、c
  • 我们需要重写 | 即__or__方法
"""
 |运算符的重载
"""

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

    def __str__(self):
        return self.name

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


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

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

    def run(self):
        for seq in self.sequences:
            print(seq)


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

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

Runnable接口

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

  1. 代码:
  • chain = prompt | model
  • chain变量是RunnableSequence (RunnableSerializable子类) 类型
  • 而得到这个类型的原因就是Runnable基类内部对 __or__ 魔术方法的改写。
  • 同时，在后面继续使用 | 添加新的组件，依旧会得到RunnableSequence，这就是链的基础架构。
  1. 源码查看

  1. 结论: 无论添加多少组件, 得到都是RunnableSequence类型, 该类是Runnable接口的子类, 当调用invoke方法或者stream方法, 数据就会在链条中流转, 最终完成链式调用
from langchain_community.llms.tongyi import Tongyi
from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate.from_template("你是ai助手")
model = Tongyi(model="qwen-max", api_key="sk-0af497cdd462495aa2f08dddaead1ded")

chain = prompt | model

print(type(chain))

StrOutputParser字符串输出解析器

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

  1. 链的构建完全符合要求(参与的组件)

  1. 运行报错

  1. 报错原因
  • prompt的结果是PromptValue类型，输入给了model
  • model的输出结果是 AIMessage
  • AIMessage作为参数传给第二个model时报错
  • 因为model的输入参数不支持AIMessage类型
  • model的输入支持的类型: PromptValue或字符串或序列(BaseMessage、 list、 tuple、 str、 dict).
  1. 模型(ChatTongyi)源码中关于invoke方法明确指定了input的类型

  1. 可以借助LangChain内置的解析器StrOutputParser字符串输出解析器进行类型转换
  2. StrOutputParser是LangChain内置的简单字符串解析器
  • 可以将AIMessage解析为简单的字符串，符合了模型invoke方法要求(可传入字符串，不接收AIMessage类型)
  • 是Runnable接口的子类(可以加入链)

  1. 代码示例
"""
字符串输出解析器
"""
from langchain_community.chat_models import ChatTongyi
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate


# 通过字符串输出解析器把AIMessage对象转成字符串, 如果直接把AIMessage对象传给模型会报错
parser = StrOutputParser()


model = ChatTongyi(model="qwen3-max", api_key="sk-0af497cdd462495aa2f08dddaead1ded")
prompt = PromptTemplate.from_template(
    "我的邻居姓: {lastname}, 刚生了{gender}, 请起名, 请告诉我名字, 无需其他内容, 数量1个, "
)

chain = prompt | model | parser | model

res = chain.invoke({"lastname": "张", "gender": "男孩"})
print(type(res))
print(res.content)

在前面我们完成了这样的需求去构建多模型链

不过这种做法并不标准，因为上一个模型的输出，没有被处理就输入下一个模型。

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

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

根据输出和输入的要求:

  • 模型的输出为: AIMessage类对象
  • 提示词模板要求输入如右侧代码:

所以，我们需要完成:

  • 将模型输出的AIMessage -> 转为字典 -> 注入第二个提示词模板中，形成新的提示词(PromptValue对象)

通过JsonOutputParser完成多模型链

示例代码

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

# 使用字符串输出解析器把AIMessage对象转成字符串
str_parser = StrOutputParser()
# 使用JSON输出解析器把AIMessage对象转成JSON对象
json_parser = JsonOutputParser()


model = ChatTongyi(model="qwen3-max", api_key="sk-0af497cdd462495aa2f08dddaead1ded")

# 示例: AIMessage({ "name": "张三" })
first_prompt = PromptTemplate.from_template(
    "我邻居姓: {lastname}, 刚生了{gender}, 请起一个名字, 并封装成JSON格式返回给我"
    "要求key是name, value是你起的名字, 请严格遵守格式要求"
)

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

chain = first_prompt | model | json_parser | second_prompt | model | str_parser
res:str = chain.invoke({"lastname": "张", "gender": "男孩"})
print(res)
print(type(res))

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

  • 模型输入: PromptValue或字符串或序列(BaseMessage、 list、 tuple, str、 dict)
  • 模型输出: AIMessage
  • 提示词模板输入: 要求是字典
  • 提示词模板输出: PromptValue对象
  • StrOutputParser: AIMessage输入、str输出
  • JsonOutputParser: AIMessage输入、dict输出

RunnableLambda&函数加入链

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

  • 除了JsonOutputParser这类固定功能的解析器之外
  • 我们也可以自己编写Lambda匿名函数来完成自定义逻辑的数据转换，想怎么转换就怎么转换，更自由。
  1. 想要完成这个功能，可以基于RunnableLambda类实现。
  2. RunnableLambda类是LangChain内置的，将普通函数等转换为Runnable接口实例，方便自定义函数加入chain。
  3. 语法: RunnableLambda(函数对象或lambda匿名函数)

  1. 示例代码
from langchain_community.chat_models import ChatTongyi
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import Runnable, RunnableLambda

str_parser = StrOutputParser()
# 使用自定义函数解析AIMessage对象
my_fun = RunnableLambda(lambda ai_msg: {"name": ai_msg.content})

model = ChatTongyi(model="qwen3-max", api_key="sk-0af497cdd462495aa2f08dddaead1ded")
first_prompt = PromptTemplate.from_template(
    "我邻居姓: {lastname}, 刚生了{gender}, 请起一个名字, 不需要额外信息"
)

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

chain = first_prompt | model | my_fun | second_prompt | model | str_parser

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

  1. 函数直接入链

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

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

临时记忆

如果想要封装历史记录，除了自行维护历史消息外，也可以借助LangChain内置的历史记录附加功能。LangChain提供了History功能，帮助模型在有历史记忆的情况下回答。

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

  1. 代码示例

  1. 完整代码
from langchain_community.chat_models import ChatTongyi
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate, ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableWithMessageHistory

# 模型对象
model = ChatTongyi(model="qwen3-max", api_key="sk-0af497cdd462495aa2f08dddaead1ded")
# 通过提示词模板
# prompt = PromptTemplate.from_template(
#     "你需要根据会话历史回应用户问题。对话历史:{chat_history}，用户提问:{input}，请回答"
# )
# 对话提示词模板
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你需要根据会话历史回应用户问题。对话历史:{chat_history}"),
        MessagesPlaceholder(variable_name="chat_history"),
        ("human", "请回答如下问题,{input}"),
    ]
)

# 字符串输出流
str_parser = StrOutputParser()


# 自定义方法, 用于调试
def print_prompt(full_prompt):
    print("=" * 20, full_prompt.to_string(), "=" * 20)
    return full_prompt


# 创建基础链
base_chain = prompt | print_prompt | model | str_parser

# 历史会话容器
# key就是session, value就是InMemoryChatMessageHistory类对象
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,  # 获取会话历史(通过会话id获取历史会话对象)
    input_messages_key="input",  # 用户在输入模板中的占位符
    history_messages_key="chat_history",  # 用户在输入模板中的占位符
)

if __name__ == '__main__':
    # 配置LangChain,为当前程序配置session_id
    session_config = {
        "configurable": {
            "session_id": "user_001"
        }
    }

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

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

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

  1. 核心说明
  • RunnableWithMessageHistory是LangChain内Runnable接口的实现，主要用于:
    • 创建一个带有历史记忆功能的Runnable实例(链)
  • 它在创建的时候需要提供一个BaseChatMessageHistory的具体实现(用来存储历史消息)
    • InMemoryChatMessageHistory可以实现在内存中存储历史
  • 额外的，如果想要在invoke或stream执行链的同时，将提示词print出来，可以在链中加入自定义函数实现。
    • 注意函数的输入应原封不动返回出去，避免破坏原有业务，仅在return之前，print所需信息即可

长期记忆

官方给出了基于文件的历史消息存储示例代码。我们可以自行实现一个基于Json格式和本地文件的会话数据保存。

FileChatMessageHistory类实现，核心思路:

  1. 基于文件存储会话记录，以session_id为文件名，不同session_id有不同文件存储消息
  2. 继承BaseChatMessageHistory实现如下3个方法:
  • add_messages: 同步模式，添加消息
  • messages: 同步模式，获取消息
  • clear: 同步模式，清除消息
  1. 官方在BaseChatMessageHistory类的注释中提供了一个基于文件存储的示例代码。

  1. 完整代码实现
import json
import os
from typing import Sequence

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


# 自定义会话存储的方法
class FileChatMessageHistory(BaseChatMessageHistory):
    def __init__(self, session_id, storage_path):
        self.session_id = session_id # 会话id
        self.storage_path = storage_path # 存储路径
        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]):   # Sequence序列 类似list/tuple
        all_messages = list(self.messages) # 已有的消息列表, messages来自父类
        all_messages.extend(messages) # 把新的消息和旧的消息合并为一个list

        # 将数据同步写入到本地文件
        # 类对象不能直接写入文件, 如果直接写入是一堆二进制数据
        # 要将BaseMessage对象转成字典, 再借助json模块以json格式写入文件
        # 完整代码
        # new_message = []
        # for message in all_messages:
        #     new_message.append(message.to_dict())

        # 列表推导式
        # 流程: 遍历all_messages, 赋值给message, 调用message_to_dict()方法, 把返回值放入list
        new_message = [message_to_dict(message) for message in all_messages]

        # 将数据写入文件
        with open(self.file_path, "w", encoding="utf-8") as f:
            json.dump(new_message, f)

    @property # 该注解不能省略, 作用是把该方法变成成员属性, LangChain调用该方法是以属性的方式调用的
    def messages(self) -> list[BaseMessage]:
        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", api_key="sk-0af497cdd462495aa2f08dddaead1ded")
# 对话提示词模板
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你需要根据会话历史回应用户问题。对话历史:{chat_history}"),
        MessagesPlaceholder(variable_name="chat_history"),
        ("human", "请回答如下问题,{input}"),
    ]
)

# 字符串输出流
str_parser = StrOutputParser()

# 自定义方法, 用于调试
def print_prompt(full_prompt):
    print("=" * 20, full_prompt.to_string(), "=" * 20)
    return full_prompt


# 创建基础链
base_chain = prompt | print_prompt | model | str_parser

# 历史会话容器 (由内存存储变成文件存储)
def get_history(session_id):
    return FileChatMessageHistory(session_id, "./19_chat_history")


# 创建增强链
conversation_chain = RunnableWithMessageHistory(
    base_chain,  # 被增强的链
    get_history,  # 获取会话历史(通过会话id获取历史会话对象)
    input_messages_key="input",  # 用户在输入模板中的占位符
    history_messages_key="chat_history",  # 用户在输入模板中的占位符
)

if __name__ == '__main__':
    # 配置LangChain,为当前程序配置session_id
    session_config = {
        "configurable": {
            "session_id": "user_001"
        }
    }

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

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

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

文档加载器

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

  1. 文档加载器(内置或自行实现)需实现BaseLoader接口。
  2. Class Document，是LangChain内文档的统一载体，所有文档加载器最终返回此类的实例。
  3. 一个基础的Document类实例，基于如下代码创建:

  1. 可以看到，Document类其核心记录了:
  • page_content: 文档内容
  • metadata: 文档元数据(字典)

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

  1. load(): 一次性加载全部文档
  2. lazy_load(): 延迟流式传输文档，对大型数据集很有用，避免内存溢出。
  3. 一个简单的CSVLoader的使用示例如下:

LangChain内置了许多文档加载器，

详细参见官方文档: https://docs.langchain.com/oss/python/integrations/document_loaders

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

  • CSVLoader
  • JSONLoader
  • PDFLoader

CSVLoader

CSVLoader用于加载CSV文件，加载成功得到的即Document对象。

  1. 基础使用

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

  1. 完整代码

"""
csv文档加载器
"""
from langchain_community.document_loaders import CSVLoader

loader = CSVLoader(
    file_path="./data/stu.csv",
    csv_args={
        "delimiter": ",", # 指定分隔符，默认,
        "quotechar": '"', # 指定引号字符，默认"
        "fieldnames": ["姓名", "年龄", "性别", "爱好"], # 自定义表头， 数据源中没有表头时使用
    },
    encoding="utf-8", # 指定编码字符集
)

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

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

JSONLoader

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

  1. 使用JSONLoader需要额外安装: pip install jq

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

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

  1. 抽取JSON对象

from langchain_community.document_loaders import JSONLoader

# 抽取字符串
# loader = JSONLoader(
#     file_path="./data/stu.json",
#     jq_schema=".other.addr",  # 把数据“深圳”抽取出来
# )

# 抽取对象
loader = JSONLoader(
    file_path="./data/stu.json",
    jq_schema=".",  # 收取对象
    text_content=False, # 告知文档加载器，抽取的内容不是字符串，是json对象
)

document = loader.load()
print(document)

  1. 抽取JSON数组

"""
json文档加载器
"""
from langchain_community.document_loaders import JSONLoader

# 抽取对象
# loader = JSONLoader(
#     file_path="./data/stus.json",
#     jq_schema=".",  # 收取对象
#     text_content=False, # 告知文档加载器，抽取的内容不是字符串，是json对象
# )

# 抽取字符串
loader = JSONLoader(
    file_path="./data/stus.json",
    jq_schema=".[].name",   # 收取字符串
    text_content=False, # 告知文档加载器，抽取的内容不是字符串，是json对象
)

document = loader.load()
print(document)

  1. 抽取JSONLines文件

JsonLines文件: 每一行都是一个独立的字典(Json对象)

"""
json文档加载器
"""
from langchain_community.document_loaders import JSONLoader

# 抽取JSONLines
loader = JSONLoader(
    file_path="./data/stu_json_lines.json",
    jq_schema=".name",   # 收取字符串
    text_content=False, # 告知文档加载器，抽取的内容不是字符串，是json对象
    json_lines=True,    # 告知文档加载器，是JSONLines格式
)

document = loader.load()
print(document)

TxtLoader

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

  1. 作用: 读取文本文件(如.txt)，将全部内容放入一个Document对象中。

  1. 如果文档很大， 加载到一个Document对象中不太合适
  2. RecursiveCharacterTextSplitter，递归字符文本分割器，主要用于按自然段落分割大文档。
  3. 是LangChain官方推荐的默认字符分割器。
  4. 它在保持上下文完整性和控制片段大小之间实现了良好平衡，开箱即用效果佳。

  1. 示例代码
"""
Text加载器和文档分割器
"""
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

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

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000, # 分段的最大字符数
    chunk_overlap=50, # 分段之间允许重叠的字符数
    separators=["\n\n", "\n", "。", ".", "!", "！", "?", "？", " ", ""], # 段落的分割符
    length_function=len, # 统计字符数
)

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

PDFLoader

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

  1. PyPDFLoader加载器，依赖PyPDF库，所以，需要安装它: pip install pypdf

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

  1. 完整代码
from langchain_community.document_loaders import PyPDFLoader

# 加载文档1
# loader = PyPDFLoader(
#     file_path="./data/pdf1.pdf",
# )

# 加载文档2
loader = PyPDFLoader(
    file_path="./data/pdf2.pdf",
    # 默认page模式, 每个页面形成一个document
    # single模式, 不管多少页面，只返回一个document
    mode="single",
    # 可以加载有密码的pdf文件
    password="itheima"
)


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

向量存储

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

如图，这是一个典型的向量存储应用，也即是典型的RAG流程。这部分开发主要涉及到:

  1. 如何文本转向量(前文已经学习)
  2. 创建向量存储，基于向量存储完成:

向量存储

  1. 内置向量存储的使用
  • 基础语法

  • 完整示例
from langchain_community.document_loaders import CSVLoader
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.vectorstores import InMemoryVectorStore

# 创建向量存储实例
vector_store = InMemoryVectorStore(
    embedding=DashScopeEmbeddings(  # 指定嵌入模型为DashScopeEmbeddings，它负责将输入文本转换为向量
        dashscope_api_key="sk-0af497cdd462495aa2f08dddaead1ded",
    )
)

loader = CSVLoader(
    file_path="./data/info.csv",
    encoding="utf-8",
    source_column="source",  # 指定数据源列
)

documents = loader.load()
print(documents)

# 增加向量
vector_store.add_documents(
    documents=documents,  # 被添加的文档，类型：list[Document]
    ids=["id" + str(i) for i in range(1, len(documents) + 1)]  # 给添加的文档设置id（字符串），类型：list[str]
)

# 删除向量
vector_store.delete(["id1", "id2"])

# 检索向量
res = vector_store.similarity_search(
    "Python是不是简单易学呀",
    k=2  # 返回的向量个数
)
print(res)

  1. 外部(Chroma)向量存储的使用
  • 基础语法

  • 完整示例
from langchain_chroma import Chroma
from langchain_community.document_loaders import CSVLoader
from langchain_community.embeddings import DashScopeEmbeddings

"""
Chroma 向量数据库（轻量级的）
确保 langchain-chroma chromadb 这两个库安装了，没有的话请pip install
"""
# 创建向量存储实例
vector_store = Chroma(
    collection_name="text", # 当前向量存储的名称，类似数据库的表名称
    embedding_function=DashScopeEmbeddings(  # 创建嵌入模型实例
        dashscope_api_key="sk-0af497cdd462495aa2f08dddaead1ded",
    ),
    persist_directory="./chroma_db", # 指定向量数据库的存储目录
)

loader = CSVLoader(
    file_path="./data/info.csv",
    encoding="utf-8",
    source_column="source",  # 指定数据源列
)

documents = loader.load()
print(documents)

# 增加向量
vector_store.add_documents(
    documents=documents,  # 被添加的文档，类型：list[Document]
    ids=["id" + str(i) for i in range(1, len(documents) + 1)]  # 给添加的文档设置id（字符串），类型：list[str]
)

# 删除向量
vector_store.delete(["id1", "id2"])

# 检索向量
res = vector_store.similarity_search(
    "Python是不是简单易学呀",
    k=2,    # 返回的向量个数
    filter={"source": "黑马程序员"} # 数据过滤，排除source是黑马程序员的结果
)
print(res)

  • 可视化界面管理数据库

向量检索构建提示词

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

# 模型对象
model = ChatTongyi(model="qwen3-max", api_key="sk-0af497cdd462495aa2f08dddaead1ded")
# 提示词模板
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "以我提供的已知参考资料为主，简洁和专业的回答用户问题，参考资料：{context}"),
        ("user", "用户提问：{input}")
    ]
)
# 基于内存的向量存储对象
vector_store = InMemoryVectorStore(
    embedding=DashScopeEmbeddings(model="text-embedding-v4", dashscope_api_key="sk-0af497cdd462495aa2f08dddaead1ded")
)

# 添加资料
vector_store.add_texts(["减肥就是少吃多练", "在减脂期间吃东西很重要，清淡少油控制卡路里摄入并运动起来", "跑步是很好的运动哦"])

# 模拟用户提问
input_text = "怎么减肥？"

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


# 调试方法
def print_prompt(prompt):
    print(prompt.to_string())
    print("="*20)
    return prompt

# 构建链
chain = prompt | print_prompt | model | StrOutputParser()

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

  1. 向量存储的实例，通过add_texts(list[str])方法可以快速添加到向量存储中。
  2. 流程:
  • 先通过向量存储检索匹配信息
  • 将用户提问和匹配信息一同封装到提示词模板中提问模型

LangChain组件

RunnablePassthrough的使用

  1. 回顾：上个案例的代码可以正常执行

  1. 目标： 让向量检索入链，使用RunnablePassthrough类
from langchain_community.chat_models import ChatTongyi
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.documents import Document
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.vectorstores import InMemoryVectorStore

# 模型对象
model = ChatTongyi(model="qwen3-max", api_key="sk-0af497cdd462495aa2f08dddaead1ded")
# 提示词模板
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "以我提供的已知参考资料为主，简洁和专业的回答用户问题，参考资料：{context}"),
        ("user", "用户提问：{input}")
    ]
)
# 基于内存的向量存储对象
vector_store = InMemoryVectorStore(
    embedding=DashScopeEmbeddings(model="text-embedding-v4", dashscope_api_key="sk-0af497cdd462495aa2f08dddaead1ded")
)

# 添加资料
vector_store.add_texts(["减肥就是少吃多练", "在减脂期间吃东西很重要，清淡少油控制卡路里摄入并运动起来", "跑步是很好的运动哦"])

# 模拟用户提问
input_text = "怎么减肥？"

# 调试方法
def print_prompt(prompt):
    print(prompt.to_string())
    print("="*20)
    return prompt

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

# 拼接向量检索结果
def format_func(docs: list[Document]):
    if not docs:
        return "无相关参考资料"

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

    return formatted_str

"""
分析链的数据类型
    retriever:
        -输入： 用户的提问          str
        -输出： 向量库的检索结果     list[Document]
    prompt:
        -输入： 用户的提问 + 向量库的检索结果   dict
        -输出： 完整的提示词                 PromptValue
"""

#构建链
"""
    执行流程：
    1. 先执行 chain.invoke(”怎么减肥？“) 启动链
    2，先执行 retriever 检索向量库
    3. 再执行 format_func  组装检索数据
    4. 得到提示词填充数据 {"input": "用户原始提问", "context": "向量检索结果（完整的参考资料）"}
    5. 再执行 prompt 得到完整的提示词
    6. 再执行 model 向模型提问
    7. 使用字符串解析器把AIMessage类型转成字符串
"""
chain = (
        {"input": RunnablePassthrough(), "context": retriever | format_func} | prompt | print_prompt | model | StrOutputParser()
)

# 调用链
res = chain.invoke(input_text)
print(res)

# python
Logo
AtomGit开源社区

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

更多推荐

cover

【Coze-AI智能体平台】低代码省时高效：Coze 应用开发全流程指南

avatar AtomGit开源社区
cover

【Coze-AI智能体平台】Coze智能体实操：翻译助手从工作流搭建到应用发布全流程详解

avatar AtomGit开源社区

GEO 学习路径指南——新手如何系统入门生成式引擎优化

传统 SEO：关注网页如何被搜索引擎收录与排序GEO（生成式引擎优化）：关注内容如何进入大模型的理解链路、引用链路和决策链路文章排名不错，但 AI 不引用流量稳定，但 AI 搜索带来的曝光几乎为零内容质量高，但在 AI 回答中"查无此人"为此，我维护了这个 GEO 中文资料库，并在 2026 年 4 月完成升级——新增快速使用指南和 3 条学习路径。本文就来详细说说，新手如何系统入门 GEO。

avatar AtomGit开源社区