Skip to content

读取、修改与执行旧 SequentialPlanner 计划 ​

先维护存档计划,再决定编排迁移 ​

旧 SequentialPlanner 先让模型产生 XML,再用 SequentialPlanParser 解析为 Plan。本节维护这个真实旧计划对象:不调用模型,固定其输出 XML,随后走原 SDK 的解析和执行。固定 Python 3.12 与 0.9.6b1,另建目录;这不是第一章 1.36.0 的升级命令。

bash
python3.12 -m venv .venv-old
source .venv-old/bin/activate
python -m pip install 'semantic-kernel==0.9.6b1' 'pydantic==2.6.4' 'openai==1.30.5' 'httpx==0.27.0' 'numpy==1.26.4'
python -m pip freeze > requirements-old.lock

旧版本依赖没有足够紧的上界:Pydantic 新版移除其导入的 Url,会在 Kernel 导入时失败,所以必须连同依赖锁定。无需密钥。保存 old_plan.py,运行 python old_plan.py:

python
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')
    print('old plan: id:ABC; edit: sku:ABC; reverse: ID:ABC; missing rejected')
asyncio.run(main())

XML 的 function.Text-upper 绑定注册插件 Text 的 upper 函数;后续步骤没有指定 input,因此使用前一步结果。第一次 abc → ABC → id:ABC。修改 tag 的 prefix 参数得到 sku:ABC;反转步骤变成 abc → id:abc → ID:ABC,证明调用顺序属于行为合同,不只是显示顺序。未知函数在解析时抛 PlannerInvalidPlanError,不能靠忽略步骤来“修好”计划。

Plan 同时含步骤、输入参数和运行游标;一次运行后 next_step_index 已到2。要比较另一个输入,应重新解析 XML 得到新计划,不能重复 invoke 已消耗的 Plan 并误以为空结果是函数行为。只有输入输出均无副作用时,这种重新运行才可随意做;写订单需要独立幂等键。

与 1.36.0 对照相同业务输入 ​

回到本课程现代环境,保存 modern_sequence.py。本地业务 abc → id:ABC 可以显式编排,保留步骤顺序与参数;若采用第二章 Process,事件和持久状态属于另外一套执行合同,不能把旧 Plan 游标当成 Process 状态。

python
import asyncio
from semantic_kernel import Kernel
from semantic_kernel.functions import kernel_function,KernelArguments
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
async def main():
    k=Kernel();p=k.add_plugin(Text(),plugin_name='Text')
    upper=await k.invoke(p['upper'],arguments=KernelArguments(input='abc'))
    result=await k.invoke(p['tag'],arguments=KernelArguments(input=str(upper),prefix='id:'))
    assert str(result)=='id:ABC'
    print(result)
asyncio.run(main())

这里验证的是两种环境对同一确定性链的结果相同,不证明模型生成的任意 XML 与现代自动函数选择等价。实际维护先保存旧 XML、版本和验收输入,再修改步骤;迁移后以相同输入比较中间结果与失败处理。

练习 ​

在旧环境给 tag 增加第二次调用,prefix 为 batch:。要求结果为 batch🆔ABC,计划步数为3且终止游标为3。另构造没有注册的函数名,确认错误出现在执行任何步骤之前。

本章答案

用心记录每一步成长