AI–LangChain的向量

1.向量的概念

  从前面的流程图中，涉及到了向量的相关概念，向量库是RAG中的一个重要节点

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

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

  向量的计算，可以接祖文本嵌入模型实现，如text-embedding-v1，其匹配通过算法实现，如余弦相似度

  

2.余弦相似度

  在前面我们介绍了向量的概念，而在算法匹配中提到了余弦相似度，这里就简单的学习下。首先我们知道向量的数字序列，共同决定了向量在高维空间中的方向和成都，而余弦相似度主要就是在撇除长度的影响，得到方向的夹角，夹角越小则约相似，也就代表了两段文本语义更加相似。

  这里只需要简单了解下即可，后续肯定是有python脚本帮助我们计算

#计算余弦相似度
import numpy as np

def cosine_similarity(vec1, vec2):
    dot_product = np.dot(vec1, vec2)
    norm_vec1 = np.linalg.norm(vec1)
    norm_vec2 = np.linalg.norm(vec2)
    return dot_product / (norm_vec1 * norm_vec2)

# 示例用法
vec1 = np.array([1, 2, 3, 4, 5])
vec2 = np.array([4, 5, 6, 7, 8])
vec3 = np.array([1, -2, -3, -4, -5])

similarity = cosine_similarity(vec1, vec2)
print(f"余弦相似度: {similarity}")
similarity = cosine_similarity(vec1, vec3)
print(f"余弦相似度: {similarity}")

  

3.向量的相关操作

  在前文中有提到过文本转向量的概念，而本章节则学习向量相关的内容，典型的向量存储应用如下：

  对于向量的处理主要就是存储、删除以及检索，而在LangChain总也为向量存储提供了统一的接口

• 存入向量：add_documents
• 删除向量：delete
• 向量检索：similarity_search

  

4.内部向量存储

  示例文本：

  示例代码：

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

vectorstore = InMemoryVectorStore(
    embedding=DashScopeEmbeddings()
)

loader = CSVLoader(
    file_path="./AI_Python/data/xl.csv",
    encoding="utf-8",
    source_column="source", #指定本条数据的来源字段
)   

documents = loader.load()

#向量存储的新增
vectorstore.add_documents(
    documents=documents, #被添加的文档，类型：list[Document]
    ids = ["id" + str(i) for i in range(1,len(documents)+1)]    # 每个文档的id，必须唯一
)

#删除
vectorstore.delete(
    ids = ["id1", "id2"]    # 要删除的文档id列表
)

#检索 返回类型list[Document]
retriever = vectorstore.similarity_search(
    query="科学健身", # 要查询的文本
    k=2    # 要返回的文档数
)
print(retriever)

  

5.外部向量存储

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

# Chroma 向量数据库（轻量级）
# 确保安装了chromadb和langchain_chroma库：pip install chromadb langchain_chroma

vectorstore = Chroma(
    collection_name="my_collection",  # 集合名称,类似于数据库中的表名
    embedding_function=DashScopeEmbeddings(),  # 嵌入模型
    persist_directory="./chroma_db"  # 数据库文件存储路径
)

loader = CSVLoader(
    file_path="./AI_Python/data/xl.csv",
    encoding="utf-8",
    source_column="source", #指定本条数据的来源字段
)   

documents = loader.load()

#向量存储的新增
vectorstore.add_documents(
    documents=documents, #被添加的文档，类型：list[Document]
    ids = ["id" + str(i) for i in range(1,len(documents)+1)]    # 每个文档的id，必须唯一
)

#删除
vectorstore.delete(
    ids = ["id1", "id2"]    # 要删除的文档id列表
)

#检索 返回类型list[Document]
retriever = vectorstore.similarity_search(
    query="科学健身", # 要查询的文本
    k=2    # 要返回的文档数
)
print(retriever)

  

6.基于向量检索构建提示词

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

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

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

# 准备一下资料（向量库的数据）
# add_texts 传入一个 list[str]
vector_store.add_texts(["减肥就是要少吃多练", "在减脂期间吃东西很重要,清淡少油控制卡路里摄入并运动起来", "跑步是很好的运动哦"])

input_text = "怎么减肥？"

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


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

# chain
chain = prompt | print_prompt | model | StrOutputParser()

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

  

7.向量检索入链的优化

  使用RunnablePassthrough

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

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

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

# 准备一下资料（向量库的数据）
# add_texts 传入一个 list[str]
vector_store.add_texts(["减肥就是要少吃多练", "在减脂期间吃东西很重要,清淡少油控制卡路里摄入并运动起来", "跑步是很好的运动哦"])

input_text = "怎么减肥？"

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

# chain = retriever | prompt | model | StrOutputParser()  因为retriever的输出是list[Document]，而prompt的输入是dict，所以不能直接用chain.invoke()方法

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

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

    return formatted_str


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

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

model：
     - 输入：完整的提示词 PromptValue
     - 输出：模型的回复   str
"""