RAG Advanced (RAG 进阶) - code.py

发表于   |   分类于 Python   |   暂无评论

加载文件进行分割向量化存储到 faiss

from langchain_experimental.text_splitter import SemanticChunker
from langchain_community.vectorstores import FAISS


# 读取文件内容
with open("./langchaintest.txt", encoding="utf-8") as f:
    state_of_the_union = f.read() #返回字符串

# 加载embedding模型
from langchain_huggingface import HuggingFaceEmbeddings
MODEL_PATH = "/Users/t-mac/.cache/huggingface/hub/models--Qwen--Qwen3-Embedding-0.6B/snapshots/c54f2e6e80b2d7b7de06f51cec4959f6b3e03418"

# mps mac芯片mps加速
embeddings = HuggingFaceEmbeddings(
    model_name=MODEL_PATH,
    model_kwargs={"device": "mps"},
    encode_kwargs={"normalize_embeddings": True}
)
# # 获取切割器
text_splitter = SemanticChunker(
    embeddings=embeddings,
    breakpoint_threshold_type="percentile",#断点阈值类型：字面值["百分位数", "标准差", "四分位距", "梯度"] 选其一
    breakpoint_threshold_amount=15 #断点阈值数量 (极低阈值 → 高分割敏感度)
)
# 切分文档
docs = text_splitter.create_documents(texts = [state_of_the_union])
# 从文档列表创建 FAISS 向量库
vector_store = FAISS.from_documents(documents=docs, embedding=embeddings)
# 【可选】持久化存储：将索引保存到本地磁盘
vector_store.save_local("faiss_index")

读取 现有json文件，不需要分割的数据直接构建

import json
from langchain_experimental.text_splitter import SemanticChunker
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document


# 读取文件内容
with open("./iciba_data_multi.json", encoding="utf-8") as f:
    json_data = json.load(f)

# 加载embedding模型
from langchain_huggingface import HuggingFaceEmbeddings
MODEL_PATH = "/Users/t-mac/.cache/huggingface/hub/models--Qwen--Qwen3-Embedding-0.6B/snapshots/c54f2e6e80b2d7b7de06f51cec4959f6b3e03418"

# mps mac芯片mps加速
embeddings = HuggingFaceEmbeddings(
    model_name=MODEL_PATH,
    model_kwargs={"device": "mps"},
    encode_kwargs={"normalize_embeddings": True}
)

# 2. 准备文档列表
# 这里我们不再使用 text_splitter，而是手动构建 Document 列表
documents = []

for item in json_data:
    # 拼接文本：英文+中文
    full_text = f"{item['content']} {item['note']}"

    # 构建 LangChain 的 Document 对象
    # page_content 是要向量化的内容
    # metadata 是你想保留但不参与向量化的信息（比如日期）
    doc = Document(
        page_content=full_text,
        metadata={"date": item["date"]}
    )
    documents.append(doc)

# 从文档列表创建 FAISS 向量库
vector_store = FAISS.from_documents(documents=documents, embedding=embeddings)
# 【可选】持久化存储：将索引保存到本地磁盘
vector_store.save_local("faiss_index")

向量检索（语义搜索）

fine_retriever = vector_store.as_retriever(search_kwargs={"k": 5})
final_docs = fine_retriever.invoke("光明")
print(final_docs)

22248-chrodae7ibn.png

BM25 检索（关键词搜索）

from langchain_community.retrievers import BM25Retriever

# 从分割好的文档创建BM25检索器（chunks和向量检索用的是同一个）
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 3  # 每次搜3个最相关的
# 实际查询
results = bm25_retriever.invoke("失败")
print(results)

# 你的搜索过程可能是这样的：
# 你搜 "失败"。
# 如果你的 BM25 没有配置中文分词器（比如 jieba），它可能在找 "失败" 这个整体。
# 但文档里只有 "成功不会..."。
# 文档里没有 "失败" 这个词（只有“失”和“败”散落在不同地方，或者根本没有），所以 BM25 判定为 0 匹配。
# 它返回的那三个结果（潜力、冒险、成功），其实是因为它们包含了“我”、“你”、“的”、“是”这些高频通用词，在没有精准匹配的情况下，BM25 只能把这些“沾点边”的文档凑数返回给你。

print("-"*50)
results = bm25_retriever.invoke("potential")
print(results)
# 英文： 句子是用空格隔开的。
# Success doesn't come to you -> 自动变成 ['Success', "doesn't", 'come', 'to', 'you']。BM25 很擅长处理这个。

07716-wihxm7oc9c.png

怎么解决？（让 BM25 支持中文）

import jieba
# 定义一个专门切分中文的函数
def chinese_tokenizer(text):
    return list(jieba.cut(text))

# 创建检索器
bm25_retriever = BM25Retriever.from_documents(
    documents,
    # 关键在这里：告诉它用 jieba 来切分
    preprocess_func=chinese_tokenizer
)

bm25_retriever.k = 3

# 再次搜索
results = bm25_retriever.invoke("失败")
print(results)

18055-z8nuvgkg64o.png

RAG 进阶 - EnsembleRetriever 混合检索器（核心工具）

from langchain_classic.retrievers import EnsembleRetriever
# 创建混合检索器
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, fine_retriever],  # 要组合的两个检索器
    weights=[0.4, 0.6]  # 各占50%权重（平衡配置）
)
# 实际查询
query = "接受失败"
results = ensemble_retriever.invoke(query)
print(results)

94042-pimnopmjp5.png

实战 - 在线阶段：RAG 问答

# 在线阶段：RAG 问答
# 1. 核心思路
# 创建一个 “检索工具”，AI 回答问题前，先调用这个工具找相关文档，再结合文档回答（避免瞎编）。
import json
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_ollama import ChatOllama

# ==========================================
# 1. 初始化模型
# ==========================================
model = ChatOllama(
    model="qwen3.5:2b",
    base_url="http://localhost:11434",
    temperature=0.7,
)

# ==========================================
# 2. 模拟你的知识库数据 (实际使用时请替换为你的向量库检索逻辑)
# ==========================================
# 这里为了演示，我直接把你的 JSON 数据放在这里，实际应替换为 vector_store.similarity_search(query)
@tool
def search_docs(query: str) -> str:
    """
    在文档库中搜索相关信息。
    如果用户询问关于日期、格言或特定事件的内容，请使用此工具。
    """
    # --- 实际逻辑替换结束 ---
    formatted_results = []
    results = ensemble_retriever.invoke(query)
    for doc in results:
        text = f"内容: {doc.page_content}"
        formatted_results.append(text)

    return "\n---\n".join(formatted_results)

# ==========================================
# 3. 创建 AI Agent
# ==========================================
agent = create_agent(
    model=model,
    tools=[search_docs],
    system_prompt="""你是每日一词的助手。
    1. 回答用户问题前，必须先调用 search_docs 工具查找信息。
    2. 只能根据搜索到的 [内容] 和 [备注] 进行回答。
    3. 如果搜索不到相关信息，请直接说“我在日记中没找到相关内容”，不要编造。
    4. 回答时请尽量保持优雅和文学性，就像那些格言一样。
    """
)

# ==========================================
# 4. 问答循环
# ==========================================
print("🤖 助手已就绪，请问关于这些日期的格言，你想知道什么？")

while True:
    try:
        query = input("\n请输入你的问题 (输入 q 退出): ")
        if query.lower() == 'q':
            break
        # 调用 Agent
        # 注意：这里使用 invoke 传入字典
        response = agent.invoke({
            "messages": [{"role": "user", "content": query}]
        })

        # 【关键优化】：正确提取回复内容
        # response["messages"] 是一个列表，[-1] 是最后一条消息对象
        # .content 是获取该对象中的文本内容
        final_answer = response["messages"][-1].content

        print("\n📝 助手回答：")
        print(final_answer)
        print("-" * 30)

    except Exception as e:
        print(f"发生错误: {e}")

54249-u4tolf79qxa.png

添加新评论