大模型应用开发-Langchain(V1-最新版)-上

五 检索器

5.1 嵌入和向量存储

对于文档加载器加载完毕的文档，可以利用文档分割器对整个文档进行分割，便于后续的处理，分割的原因一方面是目前的大模型一般都有最大上下文窗口的限制，也就是大模型一次能处理的输入长度是有限的，如果文档太大则无法处理，另外一个原因是一次性处理所有文档也没有意义，我们关心的往往不是整个文档，而只是文档中的一小部分。

如果文档块过多，我们可以使用嵌入模型对每个文档块进行嵌入，对于不同长度的文档块，嵌入模型会生成长度相同的嵌入向量，这个嵌入向量保留了文档块的语义，然后可以将这个嵌入向量存储到向量存储中供后续流程使用。

嵌入模型对文档块嵌入过程如下：

5.2 向量数据库

嵌入模型得到的嵌入向量一般存放到向量数据库中，常用的向量数据库FAISS、Chroma，后面会单独写一期常用的向量数据库的讲解。

5.3 检索方式

现在文档块被表示为向量，存放到了向量数据库中，对于用户关于文档的问题，我们也可以将这个问题转换为向量，然后从向量数据库中查询和问题最相关的文档块向量，继而能找到最相关的文档块，对问题进行回答，这个过程称为检索。

5.3.1 普通检索

文档块转换为嵌入向量存储到向量数据库之后，就可以利用向量数据库查找最相似的向量，这样对于文档的语义理解、匹配的问题，转换为了向量相似性查询的问题，从而可以计算。

import os
from dotenv import load_dotenv
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
 
# ----------------------------
# 1. 加载 .env 文件中的环境变量
# ----------------------------
load_dotenv()  # 自动从 .env 读取并设置环境变量
# 读取在阿里百炼平台申请的 key
api_key = os.getenv("DASHSCOPE_API_KEY")
if not api_key:
    raise ValueError("❌ DASHSCOPE_API_KEY 未在 .env 中设置！")
print("✅ DashScope API Key 已加载")
 
# ----------------------------
# 2. 初始化嵌入模型（自动从环境变量读取 key）
# ----------------------------
embeddings = DashScopeEmbeddings(
    model="text-embedding-v4",  # v4嵌入模型
    # dashscope_api_key 参数可省略！因为 LangChain 会自动读取环境变量
)
 
# ----------------------------
# 3. 假设的示例文档
# ----------------------------
documents = [
    "LangChain is a framework for developing applications powered by language models.",
    "RAG (Retrieval-Augmented Generation) combines retrieval and generation to improve LLM responses.",
    "Qwen is a large language model developed by Alibaba Cloud."
]
#构建为多个Document文档块对象
langchain_docs = [Document(page_content=doc) for doc in documents]
#构建FAISS向量库
print("🚀 调用 DashScope API 进行嵌入...")
vectorstore = FAISS.from_documents(documents=langchain_docs, embedding=embeddings)
print("✅ 向量库构建完成！")
 
# ----------------------------
# 4. 在向量存储上构建向量检索器
# ----------------------------
retriever = vectorstore.as_retriever()
#要检索的文本
query = "What is RAG?"
#利用检索器从向量存储中返回最相关的文档块
results = retriever.invoke(query)
print("\n🔍 检索结果:")
#enumerate(results,1) 将检索结果从1开始编号 输出
for i, doc in enumerate(results, 1):
    print(f"{i}. {doc.page_content}")

5.3.2 检索器融合

检索器融合策略（Retriever Fusion / Ensemble Retrieval）是指将多个不同来源或不同算法的检索器（Retrievers）的结果进行融合，以提升最终检索结果的相关性、鲁棒性和覆盖度。

import os
from typing import List
from dotenv import load_dotenv
from langchain_core.documents import Document
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_community.retrievers import BM25Retriever
# ----------------------------
# 1. 加载环境变量
# ----------------------------
load_dotenv()
# 读取在阿里百炼平台申请的 key
if not os.getenv("DASHSCOPE_API_KEY"):
    raise ValueError("请在 .env 中设置 DASHSCOPE_API_KEY")
# ----------------------------
# 2. 准备文档
# ----------------------------
documents = [
    "LangChain is a framework for developing applications powered by language models.",
    "RAG combines retrieval and generation to improve LLM responses with external knowledge.",
    "FAISS is a library for efficient similarity search on dense vectors.",
    "BM25 is a ranking function used in search engines for keyword-based retrieval.",
    "Qwen is a large language model developed by Alibaba Cloud.",
    "Embedding models map text to fixed-dimensional vectors for semantic search."
]
# 构建为多个Document文档块对象
langchain_docs = [Document(page_content=doc) for doc in documents]
# ----------------------------
# 3. 创建两个异构检索器进行融合
# ----------------------------
# 向量检索器（Qwen Embedding + FAISS）
embeddings = DashScopeEmbeddings(model="text-embedding-v2")
vectorstore = FAISS.from_documents(langchain_docs, embeddings)
# k限制每次检索返回的最相似文档数量 不指定默认k=4
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
# BM25 检索器（关键词）
bm25_retriever = BM25Retriever.from_documents(langchain_docs, k=3)
# ----------------------------
# 4. 手动实现 RRF 检索器融合函数
# ----------------------------
def reciprocal_rank_fusion(
        retriever_results: List[List[Document]],
        weights: List[float] = None,
        k: int = 60
) -> List[Document]:
    """
    对多个检索结果列表执行 Reciprocal Rank Fusion (RRF)
    Args:
        retriever_results: 多个检索器的结果列表，每个元素是 [Document]
        weights: 每个检索器的权重（可选，RRF 通常不用）
        k: RRF 常数，默认 60
    Returns:
        融合后的去重文档列表（按分数降序）
    """
    if weights is None:
        weights = [1.0] * len(retriever_results)
    elif len(weights) != len(retriever_results):
        raise ValueError("weights 长度必须与 retriever_results 一致")
    fused_scores = {}
    doc_map = {}
    for weight, results in zip(weights, retriever_results):
        for rank, doc in enumerate(results):
            # 使用内容哈希作为唯一 ID（实际项目建议用唯一 ID 字段）
            doc_id = hash(doc.page_content + str(doc.metadata))
            if doc_id not in fused_scores:
                fused_scores[doc_id] = 0.0
                doc_map[doc_id] = doc
            fused_scores[doc_id] += weight * (1 / (rank + k))
    # 按分数排序
    sorted_items = sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)
    return [doc_map[doc_id] for doc_id, _ in sorted_items]
# ----------------------------
# 5. 构建融合检索器类（可选，封装成 retriever 风格）
# ----------------------------
class RRFRetriever:
    def __init__(self, retrievers: List, weights: List[float] = None, k: int = 60, top_k: int = 4):
        self.retrievers = retrievers
        self.weights = weights
        self.k = k
        self.top_k = top_k
    def invoke(self, query: str) -> List[Document]:
        # 并行或串行获取各检索器结果
        all_results = [retriever.invoke(query) for retriever in self.retrievers]
        # 融合不同检索器的结果
        fused = reciprocal_rank_fusion(all_results, self.weights, self.k)
        # 返回前 top_k 个结果
        return fused[:self.top_k]
# ----------------------------
# 6. 使用融合检索器
# ----------------------------
fusion_retriever = RRFRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[1.0, 1.0],  # RRF 对权重不敏感，但可调节
    k=60,
    top_k=2
)
# 测试检索查询
query = "How does RAG use retrieval?"
results = fusion_retriever.invoke(query)
print("🔍 查询:", query)
print("\n✅ RRF 融合检索结果:")
for i, doc in enumerate(results, 1):
print(f"{i}. {doc.page_content}")

RRF（Reciprocal Rank Fusion，倒数排名融合） 是一种简单而强大的多路检索结果融合算法，由 Cormack 等人在 2009 年提出。它无需知道原始相似分数，仅依赖每个结果在各检索器中的排名（rank），就能有效融合多个异构检索系统的结果。

5.3.3 上下文压缩检索

import os
from typing import List
from dotenv import load_dotenv
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.chat_models import ChatTongyi
from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
# ----------------------------
# 1. 自定义 LLM 上下文压缩器
# ----------------------------
class LLMContextualCompressor:
    def __init__(self, llm):
        self.llm = llm
        # 从文档块中提取与问题直接相关的句子(不添加、不改写)
        self.prompt = ChatPromptTemplate.from_template(
            "Given the question below, extract only the sentences from the context that are "
            "directly relevant to answering it. Do not add explanations, do not rephrase, "
            "and do not include irrelevant information. If nothing is relevant, return 'None'.\n\n"
            "question: {question}\n\n"
            "context: {context}\n\n"
            "Relevant sentences:"
        )
        #构建文档压缩提取链
        self.chain = self.prompt | llm | StrOutputParser()
    def compress_documents(self, documents: List[Document], query: str) -> List[Document]:
        #文档块压缩结果
        compressed_docs = []
        for doc in documents:
            #调用链对文档压缩
            response = self.chain.invoke({
                "question": query,
                "context": doc.page_content
            })
            extracted_text = response.strip()
            # 仅当提取到非空内容时才保留
            if extracted_text and extracted_text not in {"", "None", "N/A"," "}:
                #构建结果Documents对象
                compressed_docs.append(
                    Document(
                        page_content=extracted_text,
                        metadata=doc.metadata
                    )
                )
        return compressed_docs
# ----------------------------
# 2. 主流程：上下文压缩检索
# ----------------------------
def main():
    # 加载环境变量
    load_dotenv()
    #从阿里百炼平台申请到的api_key
    api_key = os.getenv("DASHSCOPE_API_KEY")
    if not api_key:
        raise ValueError("❌ 请在 .env 文件中设置 DASHSCOPE_API_KEY")
    # 模拟含噪声的长文档
    raw_docs = [
        "LangChain is a framework for building LLM-powered applications. "
        "It supports chains, agents, and memory. "
        "Bananas are yellow fruits that grow in tropical climates. "
        "RAG (Retrieval-Augmented Generation) enhances LLM responses by retrieving external knowledge.",
        "FAISS is a library for efficient similarity search on dense vectors, developed by Meta. "
        "It supports both CPU and GPU. "
        "Pineapples are sweet and tangy tropical fruits. "
        "Qwen is a large language model developed by Alibaba Cloud."
    ]
    documents = [Document(page_content=doc) for doc in raw_docs]
    # 初始化嵌入模型和向量库
    print("🚀 正在初始化嵌入模型并构建向量库...")
    embeddings = DashScopeEmbeddings(model="text-embedding-v2")
    vectorstore = FAISS.from_documents(documents, embeddings)
    retriever = vectorstore.as_retriever(search_kwargs={"k": 2})
    # 连接千问大模型 绑定到上下文压缩器中
    llm = ChatTongyi(model="qwen-turbo")
    compressor = LLMContextualCompressor(llm)
    # 测试查询
    query = "What is RAG?"
    print("\n" + "=" * 60)
    print(f"🔍 用户查询: '{query}'")
    print("=" * 60)
    # Step 1: 检索原始文档
    retrieved_docs = retriever.invoke(query)
    print(f"\n📄 检索到 {len(retrieved_docs)} 个原始文档:")
    for i, doc in enumerate(retrieved_docs, 1):
        preview = doc.page_content.replace('\n', ' ')
        print(f"  [{i}] {preview}...")
    # Step 2: 压缩上下文
    print("\n⏳ 正在调用 Qwen 压缩上下文（提取相关片段）...")
    compressed_docs = compressor.compress_documents(retrieved_docs, query)
    # 输出结果
    print("\n✅ 最终压缩后的上下文（送入 LLM 生成答案）:")
    if compressed_docs:
        for i, doc in enumerate(compressed_docs, 1):
            print(f"  [{i}] {doc.page_content}")
    else:
        print("  ⚠️ 未找到与查询相关的内容。")
# ----------------------------
# 3. 运行入口
# ----------------------------
if __name__ == "__main__":
    main()

5.3.4 元信息查询检索

import os
import uuid
import tempfile
from datetime import datetime
from typing import Dict, Any, List
from dotenv import load_dotenv
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.chat_models import ChatTongyi
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue, Range
# ----------------------------
# 1. 初始化环境与模型
# ----------------------------
load_dotenv()
api_key = os.getenv("DASHSCOPE_API_KEY")
if not api_key:
    raise ValueError("❌ 请在 .env 中设置 DASHSCOPE_API_KEY")
#嵌入模型
embeddings = DashScopeEmbeddings(model="text-embedding-v2")
#d大模型
llm = ChatTongyi(model="qwen-turbo", dashscope_api_key=api_key)
# ----------------------------
# 2. 构造带中文元信息的示例文档
# ----------------------------
sample_docs = [
    Document(
        page_content="LangChain 是一个用于开发大语言模型应用的框架。",
        metadata={
            "作者": "张三",
            "发布日期": 20230515,
            "文档类型": "教程"
        }
    ),
    Document(
        page_content="RAG（检索增强生成）通过外部知识增强大模型的回答能力。",
        metadata={
            "作者": "李四",
            "发布日期": 20240822,
            "文档类型": "研究论文"
        }
    ),
    Document(
        page_content="通义千问是由阿里云研发的大语言模型。",
        metadata={
            "作者": "张三",
            "发布日期": 20241110,
            "文档类型": "白皮书"
        }
    ),
    Document(
        page_content="FAISS 支持对数十亿向量进行高效相似性搜索。",
        metadata={
            "作者": "王五",
            "发布日期": 20231201,
            "文档类型": "工具指南"
        }
    ),
    Document(
        page_content="上下文压缩技术可减少检索文档中的噪声信息。",
        metadata={
            "作者": "张三",
            "发布日期": 20250115,
            "文档类型": "博客"
        }
    )
]
# ----------------------------
# 3. 初始化 Qdrant 向量数据库（本地临时目录）
# ----------------------------
print("🚀 正在初始化 Qdrant 向量数据库（中文元数据）...")
temp_dir = tempfile.mkdtemp()
client = QdrantClient(path=temp_dir)
# 创建集合（注意：DashScope text-embedding-v2 输出 1536 维向量）
client.create_collection(
    collection_name="docs_zh",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE)
)
# 批量插入文档
points = []
for doc in sample_docs:
    vector = embeddings.embed_query(doc.page_content)
    payload = {
        "内容": doc.page_content,
        "作者": doc.metadata["作者"],
        "发布日期": doc.metadata["发布日期"],
        "文档类型": doc.metadata["文档类型"]
    }
    points.append(PointStruct(
        id=str(uuid.uuid4()),
        vector=vector,
        payload=payload
    ))
#更新数据库
client.upsert(collection_name="docs_zh", points=points)
# ----------------------------
# 4. 构建元数据解析器（全中文）
# ----------------------------
#元信息解析提示
metadata_prompt = ChatPromptTemplate.from_template(
    """你是一个查询理解助手，请从用户查询中提取结构化的元数据过滤条件。
可用的元数据字段及说明：
- 作者：人名，如“张三”
- 文档类型：必须是以下之一：["教程", "研究论文", "白皮书", "工具指南", "博客"]
- 发布日期：整数格式 YYYYMMDD（例如 20240822）
规则：
1. 只提取明确提到的信息。
2. “2024年之后” → 设置 start_date = 20240101
3. “2024年之前” → 设置 end_date = 20231231
4. “2023到2024年之间” → start_date=20230101, end_date=20241231
5. 如果未提及某字段，不要包含该字段。
6. 返回 JSON，若无元数据则返回 {{}}。
示例：
- 输入：“张三写的白皮书” → {{"作者": "张三", "文档类型": "白皮书"}}
- 输入：“2024年之后的教程” → {{"文档类型": "教程", "start_date": 20240101}}
- 输入：“RAG 是什么？” → {{}}
用户查询：{query}
"""
)
#元信息解析链
metadata_parser = (
        {"query": RunnablePassthrough()}
        | metadata_prompt
        | llm
        | JsonOutputParser()
)
ALLOWED_DOC_TYPES = {"教程", "研究论文", "白皮书", "工具指南", "博客"}
def parse_metadata_filters(query: str) -> Dict[str, Any]:
    """将中文查询解析为 Qdrant 兼容的中文元数据过滤条件"""
    try:
        parsed = metadata_parser.invoke(query)
        print(f"🔍 解析结果: {parsed}")
    except Exception as e:
        print(f"⚠️ 解析失败: {e}")
        return {}
    qdrant_filter = {}
    # 处理作者
    if "作者" in parsed and isinstance(parsed["作者"], str):
        qdrant_filter["作者"] = parsed["作者"]
    # 处理文档类型（严格校验）
    if "文档类型" in parsed:
        dt = parsed["文档类型"]
        if dt in ALLOWED_DOC_TYPES:
            qdrant_filter["文档类型"] = dt
        else:
            print(f"⚠️ 忽略非法文档类型: {dt}")
    # 处理日期范围
    time_conditions = {}
    if "start_date" in parsed:
        try:
            time_conditions["$gte"] = int(parsed["start_date"])
        except (ValueError, TypeError):
            pass
    if "end_date" in parsed:
        try:
            time_conditions["$lte"] = int(parsed["end_date"])
        except (ValueError, TypeError):
            pass
    if time_conditions:
        qdrant_filter["发布日期"] = time_conditions
    return qdrant_filter
# ----------------------------
# 5. 元信息感知检索函数
# ----------------------------
def metadata_aware_retrieve(query: str, k: int = 3) -> List[Document]:
    vector = embeddings.embed_query(query)
    filter_dict = parse_metadata_filters(query)
    # 构建 Qdrant Filter
    must_conditions = []
    for key, value in filter_dict.items():
        if key != "发布日期":
            must_conditions.append(FieldCondition(key=key, match=MatchValue(value=value)))
    if "发布日期" in filter_dict:
        range_cond = filter_dict["发布日期"]
        range_obj = {}
        if "$gte" in range_cond:
            range_obj["gte"] = range_cond["$gte"]
        if "$lte" in range_cond:
            range_obj["lte"] = range_cond["$lte"]
        must_conditions.append(FieldCondition(key="发布日期", range=range_obj))
    search_filter = Filter(must=must_conditions) if must_conditions else None
    # 执行搜索
    results = client.query_points(
        collection_name="docs_zh",
        query=vector,
        limit=k,
        query_filter=search_filter
    )
    # 转换为 LangChain Document
    docs = []
    for point in results.points:
        payload = point.payload
        doc = Document(
            page_content=payload["内容"],
            metadata={
                "作者": payload["作者"],
                "发布日期": payload["发布日期"],
                "文档类型": payload["文档类型"]
            }
        )
        docs.append(doc)
    return docs
# ----------------------------
# 6. 测试用例（纯中文查询）
# ----------------------------
test_queries = [
    "RAG 是什么？",
    "张三写的白皮书",
    "2024年之后发布的教程",
    "2023年到2024年之间王五写的工具指南",
    "最近关于上下文压缩的博客"
]
def run_demo():
    for query in test_queries:
        print("\n" + "=" * 70)
        print(f"💬 用户查询: '{query}'")
        print("=" * 70)
        docs = metadata_aware_retrieve(query, k=2)
        if docs:
            print("✅ 检索结果:")
            for i, doc in enumerate(docs, 1):
                print(f"  [{i}] {doc.page_content}")
                print(f"      📎 元信息: {doc.metadata}")
        else:
            print("❌ 未找到匹配文档")
if __name__ == "__main__":
    run_demo()
    client.close()  # ← 显式关闭

5.3.5 父文档回溯检索

父文档回溯检索在保持上下文完整性和提高语义检索精度之间取得平衡。

import os
import tempfile
from typing import List
from uuid import uuid4
from dotenv import load_dotenv
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_chroma import Chroma
 
# ----------------------------
# 1. 初始化嵌入模型
# ----------------------------
#加载api_key
load_dotenv()
api_key = os.getenv("DASHSCOPE_API_KEY")
if not api_key:
    raise ValueError("请设置 DASHSCOPE_API_KEY")
embeddings = DashScopeEmbeddings(model="text-embedding-v2")
 
# ----------------------------
# 2. 示例长文档
# ----------------------------
text = """
通义千问（Qwen）是阿里云研发的大语言模型，支持多轮对话和多种语言。
它在代码写作、逻辑推理、多模态等方面表现卓越。
 
LangChain 是一个用于开发 LLM 应用的框架，支持 RAG、Agent、Chain 等模式。
它兼容 Chroma、FAISS、Pinecone、Qdrant 等向量数据库。
 
RAG 技术通过检索外部知识来增强生成结果的准确性与可解释性。
企业常将其用于智能客服、知识库问答和文档分析系统。
"""
original_docs = [Document(page_content=text)]
 
# ----------------------------
# 3. 分割策略
# ----------------------------
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=30)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=10)
 
# ----------------------------
# 4. 手动构建父子映射（核心！）
# ----------------------------
# 用普通字典模拟 docstore（无需 langchain.storage）
parent_doc_store = {}  # {parent_id: Document}
# 初始化 Chroma（只存子文档）
vectorstore = Chroma(
    collection_name="sub_docs",
    embedding_function=embeddings,
    persist_directory=tempfile.mkdtemp()
)
print("📚 正在处理文档并建立父子索引...")
for doc in original_docs:
    parent_chunks = parent_splitter.split_documents([doc])
    for parent_chunk in parent_chunks:
        parent_id = str(uuid4())
        # 保存父文档到内存字典
        parent_doc_store[parent_id] = parent_chunk
        # 切分子文档
        sub_docs = child_splitter.split_documents([parent_chunk])
        # 为每个子文档添加 parent_id 元数据
        for sub_doc in sub_docs:
            sub_doc.metadata["parent_id"] = parent_id
        # 子文档加入向量库
        vectorstore.add_documents(sub_docs)
 
 
# ----------------------------
# 5. 自定义 retriever 函数（无任何废弃包依赖）
# ----------------------------
def retrieve_with_parent_context(query: str, k: int = 3) -> List[Document]:
    """执行子文档检索，并返回对应的完整父文档（去重）"""
    # 1. 检索子文档
    sub_docs = vectorstore.similarity_search(query, k=k)
 
    # 2. 提取 parent_id 并去重
    parent_ids = list({
        d.metadata.get("parent_id") for d in sub_docs
        if d.metadata and "parent_id" in d.metadata
    })
 
    # 3. 回溯父文档
    parent_docs = []
    for pid in parent_ids:
        if pid in parent_doc_store:
            parent_docs.append(parent_doc_store[pid])
 
    return parent_docs
 
 
# ----------------------------
# 6. 测试
# ----------------------------
query = "LangChain 支持哪些向量数据库？"
print(f"\n🔍 查询: {query}\n")
results = retrieve_with_parent_context(query)
if results:
    for i, doc in enumerate(results, 1):
        print(f"✅ 结果 [{i}]:\n{doc.page_content.strip()}\n{'-' * 60}")
else:
    print("❌ 未找到相关文档")

5.3.6 多维度回溯检索

import os
import tempfile
from typing import List, Dict, Any
from uuid import uuid4
from dotenv import load_dotenv
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_chroma import Chroma
# ----------------------------
# 1. 初始化嵌入模型
# ----------------------------
load_dotenv()
api_key = os.getenv("DASHSCOPE_API_KEY")
if not api_key:
    raise ValueError("请设置 DASHSCOPE_API_KEY")
embeddings = DashScopeEmbeddings(model="text-embedding-v2")
# ----------------------------
# 2. 模拟“多维度”表示生成器（实际可用 LLM 生成摘要/关键词）
# ----------------------------
def generate_multi_views(doc: Document) -> List[Document]:
    """
    为一个原始文档生成多个“视图”（维度）
    实际项目中可用 LLM 生成摘要、关键词、问答对等
    """
    content = doc.page_content.strip()
    # 视图1: 原文片段（小块）
    splitter = RecursiveCharacterTextSplitter(chunk_size=120, chunk_overlap=20)
    chunks = splitter.split_documents([doc])
    # 视图2: 人工摘要（模拟 LLM 生成）
    summary = Document(
        page_content="该文档介绍了通义千问、LangChain 框架和 RAG 技术的核心概念与应用场景。",
        metadata={"view_type": "summary"}
    )
    # 视图3: 关键句（模拟提取）
    key_sentences = [
        Document(page_content="LangChain 支持 Chroma、FAISS、Pinecone 等向量数据库。"),
        Document(page_content="RAG 通过检索外部知识增强生成结果的准确性。")
    ]
    # 合并所有视图
    all_views = chunks + [summary] + key_sentences
    return all_views
# ----------------------------
# 3. 原始文档库
# ----------------------------
raw_docs = [
    Document(page_content="""
通义千问（Qwen）是阿里云研发的大语言模型，支持多轮对话和多种语言。
它在代码写作、逻辑推理、多模态等方面表现卓越。
LangChain 是一个用于开发 LLM 应用的框架，支持 RAG、Agent、Chain 等模式。
它兼容 Chroma、FAISS、Pinecone、Qdrant 等向量数据库。
RAG 技术通过检索外部知识来增强生成结果的准确性与可解释性。
企业常将其用于智能客服、知识库问答和文档分析系统。
""")
]
# ----------------------------
# 4. 构建多维度索引
# ----------------------------
vectorstore = Chroma(
    collection_name="multi_views",
    embedding_function=embeddings,
    persist_directory=tempfile.mkdtemp()
)
# 存储原始文档的字典：{doc_id: original_document}
original_doc_store: Dict[str, Document] = {}
print("📚 正在生成多维度表示并建立索引...")
for raw_doc in raw_docs:
    doc_id = str(uuid4())
    # 保存原始文档
    original_doc_store[doc_id] = raw_doc
    # 生成多个视图
    views = generate_multi_views(raw_doc)
    # 为每个视图添加元数据：指向原始文档 ID
    indexed_views = []
    for view in views:
        view.metadata["original_doc_id"] = doc_id
        indexed_views.append(view)
    # 将所有视图加入向量库
    vectorstore.add_documents(indexed_views)
# ----------------------------
# 5. 多维度回溯检索函数
# ----------------------------
def multi_dimension_retrieve(query: str, k: int = 5) -> List[Document]:
    """
    1. 用 query 检索多维度视图（摘要/关键词/片段）
    2. 回溯到对应的原始完整文档（去重）
    """
    # 检索匹配的视图
    matched_views = vectorstore.similarity_search(query, k=k)
    # 提取原始文档 ID（去重）
    original_ids = list({
        v.metadata.get("original_doc_id") for v in matched_views
        if v.metadata and "original_doc_id" in v.metadata
    })
    # 回溯原始文档
    results = []
    for oid in original_ids:
        if oid in original_doc_store:
            results.append(original_doc_store[oid])
    return results
# ----------------------------
# 6. 测试不同查询
# ----------------------------
test_queries = [
    "总结一下这个文档",
    "LangChain 支持哪些向量数据库？",
    "RAG 有什么用？"
]
for query in test_queries:
    print("\n" + "=" * 70)
    print(f"🔍 查询: {query}")
    print("=" * 70)
    results = multi_dimension_retrieve(query)
    if results:
        for i, doc in enumerate(results, 1):
            print(f"✅ 结果 [{i}]:\n{doc.page_content.strip()}\n{'-' * 60}")
    else:
        print("❌ 未找到相关文档")

5.3.7 多角度查询检索

import os
from dotenv import load_dotenv
 
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.chat_models import ChatTongyi  # 👈 官方千问模型
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_chroma import Chroma
from langchain_core.documents import Document
 
# ----------------------------
# 1. 初始化（自动读取 DASHSCOPE_API_KEY）
# ----------------------------
load_dotenv()  # 确保 .env 中有 DASHSCOPE_API_KEY
# 使用官方 ChatTongyi（底层调用 DashScope Qwen）
llm = ChatTongyi(
    model="qwen-max",
    temperature=0.7,
)
# 嵌入模型
embeddings = DashScopeEmbeddings(model="text-embedding-v2")
 
# ----------------------------
# 2. 构建知识库
# ----------------------------
text = """
通义千问（Qwen）是阿里云研发的大语言模型，支持多轮对话和多种语言。
LangChain 是一个用于开发 LLM 应用的框架，支持 RAG、Agent、Chain 等模式。
它兼容 Chroma、FAISS、Pinecone、Qdrant 等向量数据库。
RAG 技术通过检索外部知识来增强生成结果的准确性。
"""
splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=20)
docs = splitter.split_documents([Document(page_content=text)])
vectorstore = Chroma.from_documents(docs, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})
 
# ----------------------------
# 3. 多角度查询生成链
# ----------------------------
multi_query_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个查询扩展助手。"),
    ("human", """基于原始问题，生成 3 个不同表述但语义相同的查询。
每行一个，不要编号，不要解释。
原始问题: {question}""")
])
# 生成多查询的 chain
query_generator = (
        {"question": RunnablePassthrough()}
        | multi_query_prompt
        | llm
        | StrOutputParser()
)
 
 
def get_unique_docs(question: str):
    # 1. 生成多查询
    response = query_generator.invoke(question)
    queries = [q.strip() for q in response.split("\n") if q.strip()]
    all_queries = [question] + queries  # 包含原始问题
    print(f"🔍 原始: {question}")
    print(f"🔄 扩展: {queries}\n")
 
    # 2. 检索所有查询结果并去重
    seen = set()
    unique_docs = []
    for q in all_queries:
        for doc in retriever.invoke(q):
            content = doc.page_content.strip()
            if content not in seen:
                seen.add(content)
                unique_docs.append(doc)
    return unique_docs
 
 
# ----------------------------
# 4. 测试
# ----------------------------
results = get_unique_docs("LangChain 支持哪些向量数据库？")
for i, doc in enumerate(results, 1):
print(f"[{i}] {doc.page_content.strip()}\n{'-' * 50}")

5.4 文档嵌入管理

文档块嵌入存储到向量数据库中，需要对文档嵌入进行管理，如果是相同文档块，可以不用再次嵌入存储向量数据库，如果是文档块更新则需要重新嵌入

import hashlib
import os
from typing import List, Optional
from dotenv import load_dotenv
from langchain_core.documents import Document
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_chroma import Chroma
load_dotenv()
# 初始化
embeddings = DashScopeEmbeddings(model="text-embedding-v2")
vectorstore = Chroma(
    collection_name="docs_with_hash",
    embedding_function=embeddings,
    persist_directory="./chroma_db"  # 持久化，方便后续更新
)
def compute_doc_hash(doc: Document) -> str:
    """基于内容和元数据计算唯一哈希（可扩展）"""
    content = doc.page_content.strip()
    # 可选：加入关键元数据（如 source、version）
    hash_input = content.encode("utf-8")
    return hashlib.sha256(hash_input).hexdigest()[:16]  # 缩短便于阅读
def get_existing_doc_ids_by_hash(doc_hash: str) -> List[str]:
    """查询向量库中是否存在相同哈希的文档"""
    results = vectorstore.get(
        where={"doc_hash": doc_hash},
        include=["metadatas"]
    )
    return results["ids"]
def add_or_update_documents(docs: List[Document]):
    """智能添加或更新文档（避免重复，支持更新）"""
    for doc in docs:
        doc_hash = compute_doc_hash(doc)
        existing_ids = get_existing_doc_ids_by_hash(doc_hash)
        if existing_ids:
            print(f"✅ 文档已存在（hash={doc_hash}），跳过嵌入")
            continue
        # 检查是否有**内容不同但来源相同**的旧版本（需更新）
        source = doc.metadata.get("source", "unknown")
        old_docs = vectorstore.get(
            where={"source": source},
            include=["metadatas", "documents"]
        )
        # 如果内容变了（哈希不同），但 source 相同 → 视为更新
        if old_docs["ids"]:
            old_content = old_docs["documents"][0]
            old_hash = hashlib.sha256(old_content.encode()).hexdigest()[:16]
            if old_hash != doc_hash:
                # 删除旧版本
                vectorstore.delete(ids=old_docs["ids"])
                print(f"🔄 更新文档（source={source}）：删除旧版，插入新版")
            else:
                print(f"✅ 文档未变（source={source}），跳过")
                continue
        # 添加新文档（带哈希）
        doc.metadata["doc_hash"] = doc_hash
        vectorstore.add_documents([doc])
        print(f"🆕 新增文档（hash={doc_hash}）")
# ----------------------------
# 示例使用
# ----------------------------
if __name__ == "__main__":
    # 初始文档
    doc1 = Document(
        page_content="LangChain 支持 Chroma、FAISS 等向量数据库。",
        metadata={"source": "guide.md"}
    )
    print("第一次添加：")
    add_or_update_documents([doc1])
    print("\n第二次添加（相同内容）：")
    add_or_update_documents([doc1])  # 应跳过
    print("\n更新文档内容：")
    doc1_updated = Document(
        page_content="LangChain 支持 Chroma、FAISS、Pinecone、Qdrant 等向量数据库。",
        metadata={"source": "guide.md"}  # 相同 source
    )
    add_or_update_documents([doc1_updated])  # 应更新

六 记忆

6.1 基础会话记忆

将会话信息保存到指定存储对象中(如文件、Redis),

import os
from pathlib import Path
from dotenv import load_dotenv
from uuid import uuid4
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_models import ChatTongyi
from langchain_community.chat_message_histories import FileChatMessageHistory
#加载阿里百炼平台申请的api_key
load_dotenv()
os.environ["DASHSCOPE_API_KEY"] = os.getenv("DASHSCOPE_API_KEY")
 
# 创建记忆存放目录
Path("histories").mkdir(exist_ok=True)
# 创建模型
llm = ChatTongyi(model="qwen-max", temperature=0.7)
# 创建带记忆的 prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个 helpful AI 助手，请用中文回答。"),
    #自动添加历史消息 MessagesPlaceholder占位
    MessagesPlaceholder("history"),
    ("human", "{input}")
])
#创建指定session_id的历史消息存储对象
def get_history(session_id: str):
    return FileChatMessageHistory(f"histories/{session_id}.json")
#创建带记忆的 chain
chain_with_mem = RunnableWithMessageHistory(
    prompt | llm,
    # 指定历史消息的存储对象
    get_history,
    # 指定输入消息的 key名称
    input_messages_key="input",
    # 添加历史消息的 key名称
    history_messages_key="history",
)
 
if __name__ == "__main__":
    #获取session_id
    session_id = str(uuid4())
 
    print("💬 基础会话记忆（输入 'quit' 退出）")
    while True:
        user_input = input("\n👤 你: ")
        if user_input.lower() == "quit":
            break
        response = chain_with_mem.invoke(
            {"input": user_input},
            config={"configurable": {"session_id": session_id}}
        )
        print(f"🤖 AI: {response.content}")
 
print(f"\n✅ 历史已保存至: histories/{session_id}.json")

6.2 带上下文裁剪记忆

当保存的上下文过长时，对上下文裁剪进行保存

import os
from pathlib import Path
from dotenv import load_dotenv
from uuid import uuid4
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.messages.utils import trim_messages
from langchain_community.chat_models import ChatTongyi
from langchain_community.chat_message_histories import FileChatMessageHistory
 
load_dotenv()
os.environ["DASHSCOPE_API_KEY"] = os.getenv("DASHSCOPE_API_KEY")
Path("histories").mkdir(exist_ok=True)
llm = ChatTongyi(model="qwen-max", temperature=0.7)
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个 helpful AI 助手，请用中文回答。"),
    MessagesPlaceholder("history"),
    ("human", "{input}")
])
 
 
def get_history(session_id: str):
    return FileChatMessageHistory(f"histories/{session_id}_trimmed.json")
 
 
# 构建带裁剪的 chain
chain = (
        RunnablePassthrough.assign(
            history=lambda x: trim_messages(
                x["history"],
                max_tokens=1500,
                strategy="last",
                token_counter=llm,
            )
        )
        | prompt
        | llm
)
 
chain_with_mem = RunnableWithMessageHistory(
    chain,
    get_history,
    input_messages_key="input",
    history_messages_key="history",
)
 
if __name__ == "__main__":
    session_id = str(uuid4())
    print("💬 带裁剪的记忆（最多保留 1500 tokens 历史）")
    while True:
        user_input = input("\n👤 你: ")
        if user_input.lower() == "quit":
            break
        response = chain_with_mem.invoke(
            {"input": user_input},
            config={"configurable": {"session_id": session_id}}
        )
        print(f"🤖 AI: {response.content}")

其中上下文裁剪代码解释如下：

6.3 摘要式记忆

对记忆内容进行摘要再记录，避免浪费存储空间、增大模型上下文

import os
from pathlib import Path
from dotenv import load_dotenv
from uuid import uuid4
from langchain_core.messages import SystemMessage, BaseMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_models import ChatTongyi
from langchain_community.chat_message_histories import FileChatMessageHistory
load_dotenv()
os.environ["DASHSCOPE_API_KEY"] = os.getenv("DASHSCOPE_API_KEY")
Path("histories").mkdir(exist_ok=True)
# 对话模型
llm = ChatTongyi(model="qwen-max", temperature=0.7)
# 摘要模型
summarizer_llm = ChatTongyi(model="qwen-turbo", temperature=0.3)
def summarize_history(messages):
    """
    对对话历史进行摘要处理
    
    Args:
        messages: 消息列表
        
    Returns:
        处理后的消息列表（包含摘要和最近消息）
    """
    if len(messages) < 4:
        return messages
    summarizer_prompt = ChatPromptTemplate.from_messages([
        ("system", "请将以下对话总结为一段不超过 80 字的连贯摘要，保留关键信息："),
        ("human", "{dialogue}")
    ])
    # 保留最近两条对话内容，前面的进行总结
    dialogue = "\n".join([f"{m.type}: {m.content}" for m in messages[:-2]])
    summary = (summarizer_prompt | summarizer_llm).invoke({"dialogue": dialogue})
    return [SystemMessage(content=f"之前的对话摘要：{summary.content}")] + messages[-2:]
class SummaryFileHistory(FileChatMessageHistory):
    """带摘要功能的文件历史记录类"""
    def add_message(self, message):
        """
        添加消息并根据需要进行摘要处理
        
        Args:
            message: 要添加的消息
        """
        # 先添加新消息
        super().add_message(message)
        # 获取当前所有消息
        msgs = super().messages
        # 如果消息数量达到阈值，进行摘要并保存
        if len(msgs) >= 4:
            summarized_msgs = summarize_history(msgs)
            # 清空原历史并保存摘要后的内容
            self.clear()
            for msg in summarized_msgs:
                super().add_message(msg)
    @property
    def messages(self):
        """
        获取消息列表，如果需要则进行摘要处理
        
        Returns:
            消息列表
        """
        msgs = super().messages
        return summarize_history(msgs)
def get_history(session_id: str):
    """
    获取指定会话ID的历史记录对象
    
    Args:
        session_id: 会话ID
        
    Returns:
        SummaryFileHistory对象
    """
    return SummaryFileHistory(f"histories/{session_id}_summary.json")
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个 helpful AI 助手，请用中文回答。"),
    MessagesPlaceholder("history"),
    ("human", "{input}")
])
chain_with_mem = RunnableWithMessageHistory(
    prompt | llm,
    get_history,
    input_messages_key="input",
    history_messages_key="history",
)
if __name__ == "__main__":
    session_id = str(uuid4())
    print("💬 摘要式记忆（超过 4 轮后自动摘要）")
    for i in range(6):  # 模拟多轮对话
        now_input = input("\n👤 你: ")
        user_input = f"这是第 {i + 1} 轮对话" + now_input
        response = chain_with_mem.invoke(
            {"input": user_input},
            config={"configurable": {"session_id": session_id}}
        )
        print(f"[{i + 1}] 🤖 {response.content}")

6.4 RAG增强记忆

将过往对话信息存储到向量存储，然后对话id作为元信息，查询相关记忆

import os
from pathlib import Path
from dotenv import load_dotenv
from uuid import uuid4
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_community.chat_models import ChatTongyi
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_chroma import Chroma
 
load_dotenv()
os.environ["DASHSCOPE_API_KEY"] = os.getenv("DASHSCOPE_API_KEY")
Path("chroma_rag_mem").mkdir(exist_ok=True)
llm = ChatTongyi(model="qwen-max", temperature=0.7)
embeddings = DashScopeEmbeddings(model="text-embedding-v2")
# 初始化向量库（持久化）
vectorstore = Chroma(
    collection_name="user_conversations",
    embedding_function=embeddings,
    persist_directory="chroma_rag_mem"
)
 
# 保存对话
def save_interaction(session_id: str, user_msg: str, ai_msg: str):
    doc = Document(
        page_content=f"User: {user_msg}\nAI: {ai_msg}",
        metadata={"session_id": session_id}
    )
    vectorstore.add_documents([doc])
 
# 查询指定会话的关联历史
def retrieve_relevant(session_id: str, query: str, k=2):
    docs = vectorstore.similarity_search(
        query,
        k=k,
        filter={"session_id": session_id}
    )
    return "\n".join([d.page_content for d in docs])
 
#基于RAG记忆的对话链
prompt = ChatPromptTemplate.from_messages([
    ("system", "相关历史对话：\n{relevant_history}\n\n请基于以上信息回答当前问题。"),
    ("human", "{input}")
])
chain = (
        {
            "input": lambda x: x["input"],
            "relevant_history": lambda x: retrieve_relevant(x["session_id"], x["input"])
        }
        | prompt
        | llm
)
 
if __name__ == "__main__":
    session_id = str(uuid4())
 
    print("💬 RAG 增强记忆（历史存入向量库）")
    interactions = [
        ("我喜欢吃苹果", "苹果富含维生素C，对健康有益。"),
        ("我不喜欢香蕉", "明白了，你不喜欢香蕉。"),
        ("我昨天吃了苹果", "很好！苹果是健康的水果选择。")
    ]
    # 先存入历史
    for user_msg, ai_msg in interactions:
        save_interaction(session_id, user_msg, ai_msg)
 
    # 测试RAG记忆查询
    test_query = "我喜欢吃什么水果？"
    response = chain.invoke({"input": test_query, "session_id": session_id})
    print(f"❓ 问题: {test_query}")
    print(f"🤖 回答: {response.content}")

6.5 知识图谱记忆

根据对话历史构建知识图谱进行记忆

import os
import json
import re
from pathlib import Path
from typing import List, Tuple
from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.chat_models import ChatTongyi
# 加载环境变量
load_dotenv()
os.environ["DASHSCOPE_API_KEY"] = os.getenv("DASHSCOPE_API_KEY")
# 创建存储目录
Path("kg_memory").mkdir(exist_ok=True)
# 初始化模型（低 temperature 提高确定性）
llm = ChatTongyi(model="qwen-max", temperature=0.1)
# 知识图谱存储路径
KG_FILE = "kg_memory/user_kg.json"
# 知识图谱加载
def load_kg() -> List[Tuple[str, str, str]]:
    """从 JSON 文件加载知识图谱"""
    if os.path.exists(KG_FILE):
        with open(KG_FILE, "r", encoding="utf-8") as f:
            return [tuple(triple) for triple in json.load(f)]
        
    return []
# 知识图谱保存
def save_kg(triples: List[Tuple[str, str, str]]):
    """将知识图谱保存到 JSON 文件"""
    with open(KG_FILE, "w", encoding="utf-8") as f:
        json.dump(triples, f, ensure_ascii=False, indent=2)
# 知识图谱抽取
def extract_triples(text: str) -> List[Tuple[str, str, str]]:
    """
    从对话中抽取以'用户'为主语的三元组
    格式：[用户, 关系, 客体]
    """
    #构建知识谱抽取模版
    prompt = ChatPromptTemplate.from_messages([
        ("system",
         "你是一个用户画像构建助手。请严格从以下对话中提取关于**用户本人**的事实。\n"
         "要求：\n"
         "1. 所有三元组的主体必须是'用户'\n"
         "2. 关系应描述用户的属性、状态、喜好、行为或身份等\n"
         "3. 客体必须是具体值（如名字、城市、爱好等）\n"
         "4. 格式必须为：[用户, 关系, 客体]\n"
         "5. 每行只输出一个三元组，不要任何解释、序号或其他文字\n"
         "6. 如果没有可提取的用户事实，请输出空"),
        ("human", text)
    ])
    response = (prompt | llm).invoke({})
    content = response.content.strip()
    # 调试用（可选）：打印原始输出
    # print("🔍 模型原始输出:", repr(content))
    
    triples = []
    # 正则匹配：[用户, 关系, 客体]
    pattern = r'\[\s*用户\s*,\s*([^,\]]+)\s*,\s*([^,\]]+)\s*\]'
    matches = re.findall(pattern, content)
    for rel, obj in matches:
        rel = rel.strip()
        obj = obj.strip().strip('"“”')  # 去除可能的引号
        if rel and obj:
            triples.append(("用户", rel, obj))
    return triples
def update_kg(user_input: str, ai_response: str):
    """更新知识图谱：抽取新事实并去重合并"""
    dialogue = f"User: {user_input}\nAI: {ai_response}"
    new_triples = extract_triples(dialogue)
    existing = load_kg()
    kg_set = set(existing)
    for t in new_triples:
        kg_set.add(t)
    save_kg(list(kg_set))
    
    return list(kg_set)
def format_kg_for_prompt(triples: List[Tuple[str, str, str]]) -> str:
    """将三元组格式化为自然语言提示"""
    if not triples:
        return "暂无关于用户的已知事实。"
    
    lines = []
    for s, p, o in triples:
        # 将关系转为自然表达（可扩展）
        if p == "姓名":
            lines.append(f"- 用户的姓名是 {o}。")
        elif p == "喜好":
            lines.append(f"- 用户喜欢 {o}。")
        elif p == "居住地":
            lines.append(f"- 用户住在 {o}。")
        else:
            lines.append(f"- 用户 {p} {o}。")
            
    return "\n".join(lines)
# 构建带知识图谱的对话链
prompt = ChatPromptTemplate.from_messages([
    ("system",
     "你是一个 AI 助手。以下是关于当前用户的已知事实：\n{kg_facts}\n\n"
     "请基于这些事实回答用户的问题。"
     ),
    ("human", "{input}")
])
chain = prompt | llm
if __name__ == "__main__":
    print("💬 知识图谱记忆（V2：以用户为中心，支持长期记忆）")
    # 模拟多轮对话
    dialogues = [
        ("我叫张三", "你好，张三！"),
        ("我喜欢Python编程", "Python 是一门强大的编程语言！"),
        ("我住在杭州", "杭州是个美丽的城市！")
    ]
    for user_msg, ai_msg in dialogues:
        kg_before = set(load_kg())
        kg = update_kg(user_msg, ai_msg)
        kg_after = set(kg)
        new_facts = kg_after - kg_before
        print(f"👤 {user_msg}")
        print(f"🤖 {ai_msg}")
        if new_facts:
            for fact in new_facts:
                print(f"💾 新增知识: {fact}")
        else:
            print("💾 未提取到新知识")
        print()
    # 显示当前知识库
    current_kg = load_kg()
    kg_facts_str = format_kg_for_prompt(current_kg)
    print("📚 当前用户知识库:")
    print(kg_facts_str)
    print("\n" + "=" * 50)
    # 测试推理能力
    test_questions = ["我是谁？", "我喜欢什么？", "我住在哪里？", "我会 Java 吗？"]
    for q in test_questions:
        response = chain.invoke({"input": q, "kg_facts": kg_facts_str})
        print(f"❓ {q}")
        print(f"🤖 {response.content}\n")
 
    print(f"\n📊 完整知识图谱已保存至: {KG_FILE}")