过滤器、维护与生产验证：练习答案

练习解答

保存answer_filter.py，在正文同一环境运行。业务计数明确区分进入过滤器与实际执行函数。

import asyncio
from semantic_kernel import Kernel
from semantic_kernel.functions import kernel_function,KernelArguments
from semantic_kernel.filters import FilterTypes
class Echo:
    def __init__(self):self.calls=0
    @kernel_function
    def read(self,text:str)->str:
        self.calls+=1
        return text
async def main():
    k=Kernel();impl=Echo();p=k.add_plugin(impl,plugin_name='echo');seen=[]
    @k.filter(FilterTypes.FUNCTION_INVOCATION)
    async def audit(context,next):
        seen.append('before')
        try:
            if context.arguments.get('text')=='blocked':raise PermissionError('blocked')
            await next(context)
            seen.append('after')
        finally:seen.append('exit')
    r=await k.invoke(p['read'],arguments=KernelArguments(text='ok'))
    assert r.value=='ok' and seen==['before','after','exit'] and impl.calls==1
    seen.clear()
    try:await k.invoke(p['read'],arguments=KernelArguments(text='blocked'))
    except Exception as e:
        chain=[]
        while e is not None:
            chain.append(e);e=e.__cause__
        assert any(isinstance(x,PermissionError) for x in chain)
    else:raise AssertionError('blocked call accepted')
    assert seen==['before','exit'] and impl.calls==1
    print('filter rejection preserves invocation count')
asyncio.run(main())

finally既记录成功也记录拒绝，但没有吞掉异常；after仅说明下游正常返回。业务函数计数保持1，证明拒绝没有进入函数。

旧计划维护

在0.9.6b1环境保存 answer_old_plan.py。parse 阶段绑定全部函数，所以缺失函数在任何 invoke 前拒绝；三步计划重新创建，避免复用已耗尽的游标。

import asyncio
from semantic_kernel import Kernel
from semantic_kernel.functions import kernel_function,KernelArguments
from semantic_kernel.planners.sequential_planner.sequential_planner_parser import SequentialPlanParser
from semantic_kernel.exceptions import PlannerInvalidPlanError
class Text:
    @kernel_function
    def upper(self,input:str)->str:return input.upper()
    @kernel_function
    def tag(self,input:str,prefix:str)->str:return prefix+input
XML='<plan><function.Text-upper/><function.Text-tag prefix="id:"/></plan>'
async def main():
    kernel=Kernel();kernel.add_plugin(Text(),plugin_name='Text')
    def parse(xml=XML):return SequentialPlanParser.to_plan_from_xml(xml,'label input',kernel)
    plan=parse()
    assert [s.name for s in plan.steps]==['upper','tag']
    assert plan.steps[1].parameters['prefix']=='id:'
    result=await plan.invoke(kernel,KernelArguments(input='abc'))
    assert str(result)=='id:ABC'
    assert plan.next_step_index==2 and not plan.has_next_step
    changed=parse();changed.steps[1].parameters['prefix']='sku:'
    assert str(await changed.invoke(kernel,KernelArguments(input='abc')))=='sku:ABC'
    reversed_plan=parse();reversed_plan.steps.reverse()
    assert str(await reversed_plan.invoke(kernel,KernelArguments(input='abc')))=='ID:ABC'
    try:parse(XML.replace('Text-upper','Text-missing'))
    except PlannerInvalidPlanError:pass
    else:raise AssertionError('unknown function accepted')
    three=parse(XML.replace('</plan>','<function.Text-tag prefix="batch:"/></plan>'))
    assert len(three.steps)==3
    assert str(await three.invoke(kernel,KernelArguments(input='abc')))=='batch:id:ABC'
    assert three.next_step_index==3 and not three.has_next_step
    print('old plan: id:ABC; edit: sku:ABC; reverse: ID:ABC; missing rejected')
asyncio.run(main())

第二个 tag 接受第一个 tag 的结果，得到 batch🆔ABC；顺序和输入传播由断言共同固定。