真实查询、来源节点与路由追踪

前置为本章第一篇和前两章真实索引实验。检索器返回 NodeWithScore；后处理器处理节点集合；合成器把问题与保留的节点组织成提示；Query Engine 负责串联。若先合成再过滤，秘密已经进入模型，即便最后隐藏引用也无济于事。

下面用真正的 RetrieverQueryEngine 与 RouterQueryEngine，但确定性 MockEmbedding 不提供语义区分，MockLLM 只回显合成提示。这个选择让我们能断言“哪个来源实际进入合成”，不能据此报告回答质量。生产权限必须在存储查询层先过滤，本例后处理是第二道检查；它不是大库跨租户检索的主授权方案。

首次完整实验

使用前两章 Python 3.12 / llama-index-core==0.14.6 环境，将完整代码保存为 query_lab.py，运行 python query_lab.py。无需模型下载与 API 密钥。

from llama_index.core import VectorStoreIndex, Settings, get_response_synthesizer
from llama_index.core.schema import TextNode
from llama_index.core.embeddings import MockEmbedding
from llama_index.core.llms import MockLLM
from llama_index.core.postprocessor.types import BaseNodePostprocessor
from llama_index.core.query_engine import RetrieverQueryEngine, RouterQueryEngine
from llama_index.core.tools import QueryEngineTool
from llama_index.core.base.base_selector import BaseSelector,SelectorResult,SingleSelection
from llama_index.core.callbacks import CallbackManager, LlamaDebugHandler

trace=LlamaDebugHandler(print_trace_on_end=False)
Settings.callback_manager=CallbackManager([trace])
Settings.embed_model=MockEmbedding(embed_dim=8)
Settings.llm=MockLLM() # echoes the complete synthesis prompt; not a quality evaluator
class TenantFilter(BaseNodePostprocessor):
    tenant: str='A'
    def _postprocess_nodes(self,nodes,query_bundle=None):
        return [x for x in nodes if x.node.metadata.get('tenant')==self.tenant]
class PolicySelector(BaseSelector):
    def _get_prompts(self):return {}
    def _update_prompts(self,prompts):pass
    def _select(self,choices,query):
        index=1 if 'shipping' in query.query_str else 0
        return SelectorResult(selections=[SingleSelection(index=index,reason='explicit domain route')])
    async def _aselect(self,choices,query):return self._select(choices,query)

def engine(text, tenant='A', viewer='A'):
    node=TextNode(id_=text[:8],text=text,metadata={'tenant':tenant,'source':'policy-v1'})
    index=VectorStoreIndex([node])
    return RetrieverQueryEngine(retriever=index.as_retriever(similarity_top_k=2),
        response_synthesizer=get_response_synthesizer(llm=Settings.llm,response_mode='compact'),
        node_postprocessors=[TenantFilter(tenant=viewer)])
refund=engine('Refund within 7 days.');shipping=engine('Shipping takes 2 days.')
router=RouterQueryEngine(selector=PolicySelector(),query_engine_tools=[
    QueryEngineTool.from_defaults(query_engine=refund,name='refund',description='Refund policy'),
    QueryEngineTool.from_defaults(query_engine=shipping,name='shipping',description='Shipping policy')],llm=Settings.llm)
response=router.query('shipping time?')
assert len(response.source_nodes)==1
assert response.source_nodes[0].node.text=='Shipping takes 2 days.'
assert 'Shipping takes 2 days.' in str(response)
assert len(engine('Private policy',tenant='B').query('refund?').source_nodes)==0
assert trace.get_events()
print('route',response.metadata,'source',response.source_nodes[0].node.metadata,'events',len(trace.get_events()))
# Independent subquestions can fail independently; never hide failed domain results.
parts={}
for name,eng,question in [('refund',refund,'refund deadline?'),('shipping',shipping,'shipping time?')]:
    try:parts[name]={'response':str(eng.query(question)),'status':'ok'}
    except Exception as exc:parts[name]={'status':'failed','error':type(exc).__name__}
assert set(parts)=={'refund','shipping'}

输出显示路由 index=1、source=policy-v1 与非空事件记录。shipping 输入只进入 shipping 引擎；B 租户节点被移除后 source_nodes 为空。空节点不能被当作已回答：应用应将空 source_nodes 转为拒答状态。CallbackManager 观察的是框架事件，日志应只保存 ID/耗时，避免把整个提示和私人原文上传。

Router 是选路，SubQuestion 是拆题再组合。例末对退款和物流独立提问，各保留 ok/failed 状态；如果第二题需要第一题查出的订单号，就必须顺序绑定那个值，不能并行。合并响应时保留各自 source_nodes，不把一条成功结果伪装成全部问题成功。

接真实模型时在同一个环境安装匹配的 llama-index-llms-openai 集成包，把 MockLLM 换成 OpenAI(model=实际有权限的模型名)，其余节点和后处理不变；密钥从 OPENAI_API_KEY 读取。先保留这里的来源断言，再单独测正确性。此处没有执行付费模型，不能把回显当作生成答案。

把合成器接到真实本地模型

保持前面代码定义，在同一文件末尾追加以下适配器；先按本地CPU服务启动8087。它执行真实模型前向，响应仍可能是重复词；source_nodes证明来源传递，不能证明回答正确。context_window与num_output必须对应服务预算，框架才知道为证据预留多少空间。

from llama_index.core.llms import CustomLLM,CompletionResponse,LLMMetadata
from llama_index.core.llms.callbacks import llm_completion_callback
from urllib.request import Request,urlopen
import json

class LocalHTTP(CustomLLM):
    @property
    def metadata(self):
        return LLMMetadata(context_window=128,num_output=16,model_name='tiny-gpt2-local')
    @llm_completion_callback()
    def complete(self,prompt,**kwargs):
        req=Request('http://127.0.0.1:8087/generate',data=json.dumps({'prompt':prompt,'max_new_tokens':16}).encode(),headers={'Content-Type':'application/json'})
        data=json.load(urlopen(req,timeout=60))
        return CompletionResponse(text=data['text'])
    @llm_completion_callback()
    def stream_complete(self,prompt,**kwargs):
        # The HTTP fixture is non-streaming; this emits a single final chunk.
        result=self.complete(prompt,**kwargs)
        yield CompletionResponse(text=result.text,delta=result.text)

real=RetrieverQueryEngine(retriever=VectorStoreIndex([TextNode(text='Refund within 7 days.',metadata={'tenant':'A'})]).as_retriever(),
    response_synthesizer=get_response_synthesizer(llm=LocalHTTP(),response_mode='compact'),node_postprocessors=[TenantFilter()])
result=real.query('Refund deadline?')
assert result.source_nodes
print('actual model synthesis',str(result))

stream_complete在这里只发一个最终块，没有伪装成逐token流。要接流式服务，需把实际delta转为累计text，并传播取消与错误；不要按字符切割完整结果假造模型时延。

命中短句以后恢复限制条件

“可以退货”命中查询，却可能把下一句“仅限未拆封”遗漏。句子窗口把检索单位和生成上下文分开：每个节点的 text 是一个句子；window 元数据保存左右邻句；original 保存原句。解析器把窗口和原句字段排除在嵌入元数据之外，避免短节点嵌入又被整段上下文淹没。检索之后，MetadataReplacementPostProcessor 才用 window 替换待合成节点的正文。

在同一 0.14.6 环境独立保存 window_lab.py。这里用竖线指定句界，避免语言分句和外部词典下载影响机制观察。确定性地选中第一句代表检索器命中结果，未声称测试语义检索准确率；索引/检索的真实调用已在前一个实验运行。

from llama_index.core import Document
from llama_index.core.node_parser import SentenceWindowNodeParser
from llama_index.core.postprocessor import MetadataReplacementPostProcessor
from llama_index.core.schema import NodeWithScore, MetadataMode

sentences = ['Returns are accepted.', 'Only unopened items qualify.',
             'Keep the receipt.', 'Store hours are unrelated.']
def make_nodes(width=1, tenant='A'):
    parser = SentenceWindowNodeParser.from_defaults(
        window_size=width, window_metadata_key='window',
        original_text_metadata_key='original',
        sentence_splitter=lambda text: text.split('|'))
    return parser.get_nodes_from_documents([Document(
        text='|'.join(sentences), id_='policy-v1',
        metadata={'tenant': tenant, 'source': 'policy-v1'})])
def expand(nodes, viewer='A', budget=80):
    # Deep copies prevent one query from mutating shared index nodes.
    allowed = [NodeWithScore(node=n.model_copy(deep=True), score=1.0)
               for n in nodes if n.metadata['tenant'] == viewer]
    expanded = MetadataReplacementPostProcessor(
        target_metadata_key='window').postprocess_nodes(allowed)
    # Character budget for this deterministic fixture; production uses model tokens.
    return [n for n in expanded if len(n.node.text) <= budget]
nodes = make_nodes()
assert nodes[0].text == 'Returns are accepted.'
assert 'Only unopened' not in nodes[0].get_content(metadata_mode=MetadataMode.EMBED)
result = expand([nodes[0]])
assert 'Only unopened items qualify.' in result[0].node.text
assert result[0].node.metadata['source'] == 'policy-v1'
assert result[0].node.ref_doc_id == 'policy-v1'
assert nodes[0].text == 'Returns are accepted.'
assert expand([nodes[0]], viewer='C') == []
assert expand([make_nodes(width=3)[0]]) == []
print(nodes[0].text, '=>', result[0].node.text)

输出从一句扩为包含“Only unopened items qualify.”的两句。源文档 ID 与租户元数据没有改变。替换器会修改节点对象，因此复制命中节点再展开，避免第二次查询把上次的长正文当成短句。权限过滤必须在窗口内容进入合成器之前完成；这里的一份 Document 所有句子共享同一权限，若原文有段落级权限，应先按权限拆文档，否则有权限句子的窗口仍可能夹入别人的内容。

width=3 时第一句窗口含四句，超过 80 字符预算，被整体拒绝而不是截掉末尾。窗口越宽并不一定越好：它增加 token、混入无关话题；直接硬截断又可能丢失最后的约束。本例按字符测预算只为稳定演示，接模型时应按其 tokenizer 计数并记录因预算遗漏的来源。句子窗口适合局部限制条件；章节级父子展开还需要父节点关系及父文档权限，不应把邻句元数据误当成父子树。

练习

保留 A 的成功基线，使用 engine 的 viewer 参数分别为 C 创建退款与物流引擎，证明都没有来源节点并返回 refused 状态；再给路由新增“unknown”输入，返回明确无法选路，而不是默认为退款。

完成后查看本章答案。

窗口练习：命中第二句，证明左右邻句都能恢复；将预算缩至 20，验证拒绝而不丢失限制条件；混入 B 租户同句，证明 A 只保留自己的来源。不要修改输入节点。完成后查看窗口答案。

返回完整学习路线。