from pathlib import Path
from typing import Dict

def make_meta_func(department: str, upload_user: str):
    """工厂：返回一个符合签名的 metadata 函数"""
    def rich_meta(fp: Path) -> Dict[str, str]:
        stat = fp.stat()
        return {
            "file_name": fp.name,
            "department": department,   # 外部绑进来的
            "upload_user": upload_user, # 外部绑进来的
            "create_time": datetime.fromtimestamp(stat.st_ctime).isoformat(),
            "confidential": "internal",
        }
    return rich_meta

# 使用
documents = SimpleDirectoryReader(
    "data",
    file_metadata=make_meta_func(department="ABC", upload_user="sam")
).load_data()

1. 基础 RAG 流程（VectorStoreIndex → Retriever → Synthesizer）
2. 过滤器体系（metadata、相似度、关键词、时间、LLM 重排）
3. 多路召回与融合（QueryFusionRetriever）
4. 混合检索（向量 + 关键词）
5. 节点后处理链（Postprocessor 顺序、扩上下文、引用来源）
6. 模板与占位符、输出解析、结构化引用
7. 模型角色划分（embedding vs LLM）与 Ollama 本地部署
8. 元数据治理（文件名、时间、部门、密级）

→ 这 8 大块基本覆盖了 90 % 的 LlamaIndex 面试题：

• “召回阶段有哪些Retriever？”
• “过滤器执行顺序？”
• “LLM 重排原理？”
• “怎么给答案加引用？”
• “向量与关键词如何混合？”
• “metadata 缺失怎么办？”

所以确实够应付一轮中级/高级面试；真要被问到更深，一般就两条延伸：

1. 源码级：IndexStore、VectorStore 自定义实现，NodeParser 嵌套，异步 aretrieve 流程；
2. 生产级：分布式索引、增量更新、多租户权限、量化部署、重排模型微调。

from llama_index.core import SimpleKeywordTableIndex
#创建关键词引索
keyword_index = SimpleKeywordTableIndex.from_documents(documents)

#检索器导入 1向量 2关键词 3混合
from llama_index.core.retrievers import (
    VectorIndexRetriever,
    KeywordTableSimpleRetriever,
    QueryFusionRetriever,
)

# 1. 向量召回 10 条
vector_ret = VectorIndexRetriever(
    index=vector_index,
    similarity_top_k=10,
    filters=filtersABC,  # 元数据过滤仍可用
)
# 2. 关键词召回 10 条
keyword_ret = KeywordTableSimpleRetriever(
    index=keyword_index,
    keywords=["author", "childhood"],  # 也可放空，让框架自动抽关键词，这里添加可以让后续抽中更多关键词
)

# 3. 融合：去重 + 向量相似度重排（默认 fusion_mode="reciprocal_rerank"）
fusion_retriever = QueryFusionRetriever(
    [vector_ret, keyword_ret],  # 并行跑
    similarity_top_k=10,        # 融合后最终保留 10 条
)

#后面简写
query_engine = RetrieverQueryEngine(
    retriever=fusion_retriever,  # 只用换这一行
    response_synthesizer=get_response_synthesizer(),
    node_postprocessors=[
        SimilarityPostprocessor(similarity_cutoff=0.7),
        KeywordNodePostprocessor(...),  # 仍可再卡关键词
        FixedRecencyPostprocessor(top_k=7, date_key="date"),
        LLMRerank(top_n=3, choice_batch_size=3),
        PrevNextNodePostprocessor(num_prev=1, num_next=1),
    ],
)

关键词后处理KeywordNodePostprocessor 是后处理阶段的字面关键词硬过滤器——
向量或关键词召回的节点已经拿到手以后，再用字符串包含/排除的方式筛一遍，不走路径、不走模型，纯文本匹配。

不管它是向量检索召回的、关键词检索召回的、还是两者融合后的，KeywordNodePostprocessor 只认节点文本本身：

• 文本里同时包含required_keywords 所有词 → 放行
• 文本里一旦出现exclude_keywords 任意词 → 整段扔掉

它不关心节点最初来自哪个索引、哪个算法，只要最终进了 node_postprocessors 列表，就会被逐个字面扫描一遍。

from llama_index.core.postprocessor import KeywordNodePostprocessor

key_filter = KeywordNodePostprocessor(
    required_keywords=["author", "childhood"],  # 同时出现才保留
    exclude_keywords=["fiction", "novel"]       # 出现一个就扔掉
)

响应器（response_synthesizer）负责把最终节点列表 + 用户问题拼成提示词，调用大模型生成自然语言答案。它不参与检索、不过滤、不重排，只在链路最末端做“阅读-理解-回答”。

from llama_index.core import VectorStoreIndex, get_response_synthesizer


response_synthesizer = get_response_synthesizer()
query_engine = RetrieverQueryEngine(
    retriever=retriever,
    response_synthesizer=response_synthesizer, #响应器默认
    node_postprocessors=[...]   # 各种过滤器/重排/扩上下文
)

node_postprocessors处理完的节点 → response_synthesizer→ LLM→ 返回 Response

A 把节点文本切成 text_qa_template 能容纳的 token 窗口
B 响应器的两种工作状态

1.全部打包（compact 默认）

2.分块迭代（refine 先拿第一段生成初版答案，再依次把后续段落 refine 进去）

C 拼好 prompt → 调 Settings.llm.complete() → 解析返回内容 → 包成 Response 对象

更换策略 从默认的compact 变为refine

response_synthesizer = get_response_synthesizer(
    response_mode="refine",      # 或 "compact", "tree_summarize", "simple_summarize", "no_text"
    llm=Ollama(model="qwen3-14b")  # 用更大的模型专门负责生成答案
)

还可以自定义prompt模板 ，精华部分！默认compact使用text_qa_template参数；refine模式使用refine_template模板设置。

from llama_index.core import PromptTemplate

qa_tmpl = PromptTemplate(
    "你是小助手，请依据下列资料回答，若信息不足请说“不知道”。\n"
    "资料：{context_str}\n问题：{query_str}\n答案："
)

response_synthesizer = get_response_synthesizer(
    text_qa_template=qa_tmpl,
    refine_template=...          # 如果选 refine 模式再配 refine 模板
)

这两个占位符：

• {context_str} – 框架会把“经过所有 postprocessor 处理后的节点文本”自动拼成一段塞进来
• {query_str} – 用户当前的问题

下面的模板

资料：{context_str}
问题：{query_str}
答案：

添加文章来源

在 LlamaIndex 里想“给答案加引用”有两种常用做法，都只需要改响应合成器的模板，让它把节点自带的 metadata 打印出来即可；

因此metadata数据的完善性尤为重要：
文件名称，创建时间，修改时间，上传时间，上传人，上传部门，文件类型，机密类型，hash值，上传的IP

from llama_index.core import PromptTemplate

# {file_name} 就是 metadata 里的字段，可随意加
cite_tmpl = PromptTemplate(
    "资料：{context_str}\n\n"
    "问题：{query_str}\n\n"
    "答案：\n"
    "----------\n"
    "出处：{file_name}"  # 这里会被替换成真实文件名
)

response_synthesizer = get_response_synthesizer(
    text_qa_template=cite_tmpl,
    response_mode="compact"
)

检索器作用：

• 1向量模型的语义过滤（更具问题的向量和向量存储数据进行向量匹配）
• 2 使用元数据过滤

初步作用

1 创建是加入metadata属性

2创建 过滤器

3 使用检索器(index.as_retriever())过滤器（检索器内使用 或者如下简单使用）

• 基础元数据过滤器(MetadataFilter)
• 精确匹配过滤器(ExactMatchFilter)
• 嵌套的MetadataFilters实例

from llama_index.core.vector_store.types import MetadataFilter,ExactMatchFilter
documents = SimpleDirectoryReader("data").load_data()
for document in documents:
    document.metadata['department']="ABC"
#添加元数据 department为ABC 单文件也是同样返回list类型
#之后就是节点化
#过滤模块 ExactMatchFilter 和MetadataFilters
#filters过滤,下面创建了ABC部门的过滤器
filtersABC=MetadataFilters(
    filters=[ExactMatchFilter(key="department",value="ABC")]
)
实际使用最简单的查询 index对象
xxx=index.as_query_engine(filters=filtersABC)

正式查询 RetrieverQueryEngine

from llama_index.core.query_engine import RetrieverQueryEngine

下面正式简写的逻辑代码

from llama_index.core import VectorStoreIndex, get_response_synthesizer
from llama_index.core.retrievers import VectorIndexRetriever
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.postprocessor import SimilarityPostprocessor
# 1. 新增过滤器相关导入
from llama_index.core.vector_stores import ExactMatchFilter, MetadataFilters
# 2. 建立过滤器
filtersABC = MetadataFilters(
    filters=[ExactMatchFilter(key="department", value="ABC")]
)
# 3. 建索引（保持不变）
index = VectorStoreIndex.from_documents(documents)
# 4. 把过滤器挂到检索器，similarity_top_k相似度最高的10个结果，使用过滤器filtersABC过滤，此处还用向量模型更具问题向量与向量库进行匹配并将score值附上，以此给下面的节点后处理的相似度过滤器使用
retriever = VectorIndexRetriever(
    index=index,
    similarity_top_k=10,
    filters=filtersABC          # <── 关键一句
)
# 5. 响应器的创建，下面使用默认配置
response_synthesizer = get_response_synthesizer()
# 6.创建查询器 检索器来的数据，响应器，节点后处理（相似度大于0.7）
query_engine = RetrieverQueryEngine(
    retriever=retriever,
    response_synthesizer=response_synthesizer,
    node_postprocessors=[SimilarityPostprocessor(similarity_cutoff=0.7)],
)
response = query_engine.query("What did the author do growing up?")
print(response)

再加上节点关键词后处理KeywordNodePostprocessor

from llama_index.core.postprocessor import KeywordNodePostprocessor
#和相似度一个地方导入
keyword_filter = KeywordNodePostprocessor(
    required_keywords=["author", "childhood"],   # 必须同时出现
    exclude_keywords=["fiction", "novel"]        # 一旦出现就整段扔掉
)
#在其中查询加入：
node_postprocessors=[
        SimilarityPostprocessor(similarity_cutoff=0.7),  # 1. 先卡相似度
        keyword_filter,                                  # 2. 再卡关键词
    ]

前后节点PrevNextNodePostprocessor一起拉回来，实现更好的上下文

from llama_index.core.postprocessor import PrevNextNodePostprocessor
#和相似度一个地方导入
prev_next = PrevNextNodePostprocessor(
    num_prev=1,      # 往前多拿 1 个
    num_next=1,      # 往后多拿 1 个
    include_original=True,   # 把原始节点也保留
)
#在其中查询加入：
node_postprocessors=[
        SimilarityPostprocessor(similarity_cutoff=0.7),  # ① 相似度
        keyword_filter,                                  # ② 关键词
        prev_next,                                       # ③ 扩上下文
    ],

加上时间顺序

from llama_index.core import SimpleDirectoryReader
documents = SimpleDirectoryReader("data").load_data()
for doc in documents:
    doc.metadata["date"] = "2025-12-12"   # 统一日期
from llama_index.core.postprocessor import FixedRecencyPostprocessor
#和相似度一个地方导入
# 1. 按 metadata 里的 "date" 字段排序，只留最新的 5 个节点
recency = FixedRecencyPostprocessor(
    top_k=5,               # 保留最新 5 个
    date_key="date"        # 对应 metadata 中的时间字段名
)
#在其中查询加入：
 node_postprocessors=[
        SimilarityPostprocessor(similarity_cutoff=0.7),  # ① 相似度
        keyword_filter,                                  # ② 关键词
        prev_next,                                       # ③ 扩上下文
        recency,                                         # ④ 只留最新 5 个
    ],

EmbeddingRecencyPostprocessor TimeWeightedPostprocessor

参数	EmbeddingRecencyPostprocessor	TimeWeightedPostprocessor	说明
alpha	✅ 0→1 可调	❌ 无	时间权重；0=只看向量，1=只看新鲜度
time_decay (λ)	✅ 可调	✅ 可调	衰减系数；越大，旧文档得分掉得越快
top_k	✅	✅	最终按综合分排序后保留多少节点
date_key	✅ 默认 “date”	✅ 默认 “date”	节点 metadata 里的日期字段名

• 想“稍微”让新文档靠前，但又不想误杀经典老文档
  → EmbeddingRecencyPostprocessor（alpha 设 0.3–0.5）
• 业务只关心“最近一年”的内容，老文档基本无用
  → TimeWeightedPostprocessor（lambda 0.99–0.995）
• 希望平滑调节“向量 vs 时间”天平
  → 只有 EmbeddingRecencyPostprocessor 能给 alpha 旋钮。

# 方案 A：线性插值
recency = EmbeddingRecencyPostprocessor(
    alpha=0.5,          # 时间占 50 %
    time_decay=0.995,   # 旧文档衰减斜率
    top_k=5,
    date_key="date"
)
# 方案 B：乘法衰减
time_weight = TimeWeightedPostprocessor(
    time_decay=0.995,
    top_k=5,
    date_key="date"
)

重排 LLMRerank 默认，需要在前后节点召回之前PrevNextNodePostprocessor省tokens

from llama_index.core.postprocessor import (
    SimilarityPostprocessor,
    KeywordNodePostprocessor,
    PrevNextNodePostprocessor,
    FixedRecencyPostprocessor,
    LLMRerank,  # ① 新增
)
llm_rerank = LLMRerank(top_n=4, choice_batch_size=3)  
# 一次让 LLM 评 3 条 一共回4条
query_engine = RetrieverQueryEngine(
    retriever=retriever,
    response_synthesizer=get_response_synthesizer(),
    node_postprocessors=[sim, key, date, llm_rerank, prev_next], 
)
###
向量召回 10 节点
→ similarity≥0.7
→ 含关键词
→ 取最新 7 个
→ LLM 重排选 3 个
→ 前后各补 1 个相邻节点
→ 送 LLM 生成答案

去重（MD5）DeduplicatePostProcessor(show_progress=True)

在重排LLMRerank之后,扩充上下文之前PrevNextNodePostprocessor

dedup=DeduplicatePostProcessor(show_progress=True)

时间硬截断 时间不在最新的几个节点，相当于最近时间的几个节点top_k。主要针对metadata中的时间元数据 date_key。reverse=False(默认False是升序，True则为最先的，自己看着办吧)

fixedt=FixedRecencyPostprocessor(top_k=8,date_key=’create_time’,reverse=False)