查询组合、后处理与可观测性：练习答案

参考实现

复用正文函数与hits：

assert postprocess(hits,'A',0.6,4)==['a']
assert postprocess([h for h in hits if h['id']!='a'],'A',0.6,4)==['b']

先前候选被删除不应缓存旧结果；缓存需要带文档与权限版本。

返回题目。

链路变体

先运行正文 query_lab.py 的首个完整代码块（无需启动 HTTP 服务），再追加以下完整变体。viewer 是服务端确定的访问者，不改写全局默认值，因此 A 基线仍能成功。

def checked_query(eng, question):
    result = eng.query(question)
    if not result.source_nodes:
        return {'status': 'refused', 'text': 'No authorized evidence', 'sources': []}
    return {'status': 'ok', 'text': str(result),
            'sources': [n.node.metadata['source'] for n in result.source_nodes]}
for text, question in [('Refund within 7 days.', 'refund deadline?'),
                       ('Shipping takes 2 days.', 'shipping time?')]:
    allowed = checked_query(engine(text, viewer='A'), question)
    denied = checked_query(engine(text, viewer='C'), question)
    assert allowed['status'] == 'ok' and text in allowed['text']
    assert allowed['sources'] == ['policy-v1']
    assert denied == {'status': 'refused', 'text': 'No authorized evidence', 'sources': []}
    print(question, denied)

class StrictSelector(PolicySelector):
    def _select(self, choices, query):
        text = query.query_str.lower()
        if 'shipping' in text:
            index = 1
        elif 'refund' in text:
            index = 0
        else:
            raise ValueError('unknown domain')
        return SelectorResult(selections=[SingleSelection(index=index, reason='known domain')])
strict_router = RouterQueryEngine(selector=StrictSelector(), 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)
assert strict_router.query('shipping time?').source_nodes
try:
    strict_router.query('weather?')
except ValueError as exc:
    assert 'unknown domain' in str(exc)
else:
    raise AssertionError('unknown domain must be refused')

拒答路径不检查成功回答文本，也不索引空来源。未知领域属于选路拒绝，与有路由但没有授权证据不同，调用方应保留这两个原因。

窗口变体

先运行独立 window_lab.py，再追加：

from copy import deepcopy
own = make_nodes()[1]
other = make_nodes(tenant='B')[1]
before = deepcopy([own.model_dump(), other.model_dump()])
expanded = expand([own, other], viewer='A', budget=100)
assert len(expanded) == 1
assert expanded[0].node.text == ' '.join(sentences[:3])
assert expanded[0].node.metadata['tenant'] == 'A'
assert expanded[0].node.ref_doc_id == 'policy-v1'
assert expand([own, other], viewer='A', budget=20) == []
assert before == [own.model_dump(), other.model_dump()]
print('window permissions and budget verified')

预算不足在合成前明确留下空证据；应用复用 checked_query 的 refused 思路，不把空上下文交给模型补猜。当前样本 A、B 用相同文档标签只是为了测试权限，生产来源标识应包含租户命名空间与版本。