Semantic Kernel Python 连接器抽象层重构：`_inner_*` 内部方法、`SUPPORTS_FUNCTION_CALLING` 与自动函数调用机制解析

Semantic Kernel Python 连接器抽象层重构_inner_*内部方法、SUPPORTS_FUNCTION_CALLING与自动函数调用机制解析【免费下载链接】semantic-kernelIntegrate cutting-edge LLM technology quickly and easily into your apps项目地址: https://gitcode.com/GitHub_Trending/se/semantic-kernel本篇文章以 Semantic Kernel 仓库中的架构决策记录ADR0052-python-ai-connector-new-abstract-methods.md 为主体讲解 Python 版ChatCompletionClientBase与TextCompletionClientBase如何通过新增一组_inner_*内部方法与SUPPORTS_FUNCTION_CALLING类变量把一次模型调用与自动函数调用auto function invocation编排解耦为两层。读完本文你将理解该抽象层的设计动机、默认实现的工作流程以及如何基于这套约定为 Semantic Kernel 编写新的 AI 连接器。背景与问题自动函数调用带来的分层需求在 Semantic Kernel 中ChatCompletionClientBase是所有聊天补全chat completionAI 服务连接器connector的基类。在引入本 ADR 之前该类只暴露两个抽象方法get_chat_message_contents非流式地获取模型返回的聊天消息内容列表get_streaming_chat_message_contents以异步生成器方式获取流式聊天消息内容。这两个方法为上层Kernel、Agent、插件编排提供了与具体模型无关的标准化接口。随着众多模型开始支持 function callingSemantic Kernel 实现了auto function invocation自动函数调用特性当模型在回复中请求调用某个 Kernel 函数时框架会自动执行该函数、把结果写回ChatHistory并再次请求模型循环往复直到模型给出最终答案。开发者无需手动解析 function call、逐个调用插件再拼装消息开发体验因此大幅简化。但自动函数调用有一个重要的副作用一次对get_chat_message_contents或流式版本的调用底层可能触发对模型的多次调用一次原始请求 多轮工具调用往返。这说明原有的两个抽象方法承担了两种职责①真正与模型进行单次 HTTP 通信②围绕单次通信做函数调用编排。这正是一个引入新抽象层的绝佳机会——让单次模型调用成为独立的、可被专门追踪与监控的单元。设计目标三层收益ADR 明确了这次引入抽象层的三个收益简化连接器实现在基类中为get_chat_message_contents和get_streaming_chat_message_contents提供默认实现派生类只需实现真正发送单次请求的内部方法无需重复编写函数调用编排逻辑可观测性可以围绕单次模型调用建立公共的追踪tracing接口提升系统的监控与管理能力这一点在源码中体现为trace_chat_completion、trace_streaming_chat_completion装饰器见下文降低新连接器的接入成本新 AI 提供商接入 Semantic Kernel 时只需实现少量内部方法即可自动获得函数调用、流式、追踪等全部公共能力。核心改动一两个新的内部抽象方法在ChatCompletionClientBase中新增两个内部方法分别对应非流式与流式的单次模型调用。为了不破坏已经实现过自定义 AI 连接器的存量用户这两个方法没有使用abstractmethod装饰器而是采用内置连接器若不实现则抛出异常的约定ADR 中的 Revision 说明。ADR 给出的方法签名为async def _inner_get_chat_message_content( self, chat_history: ChatHistory, settings: PromptExecutionSettings ) - list[ChatMessageContent]: raise NotImplementedErrorasync def _inner_get_streaming_chat_message_content( self, chat_history: ChatHistory, settings: PromptExecutionSettings ) - AsyncGenerator[list[StreamingChatMessageContent], Any]: raise NotImplementedError需要说明的是ADR 中方法名写作单数_inner_get_chat_message_content而当前仓库实际落地的实现采用了复数形式。在 chat_completion_client_base.py 中可以看到最终版async def _inner_get_chat_message_contents( self, chat_history: ChatHistory, settings: PromptExecutionSettings, ) - list[ChatMessageContent]: Send a chat request to the AI service. raise NotImplementedError(The _inner_get_chat_message_contents method is not implemented.)async def _inner_get_streaming_chat_message_contents( self, chat_history: ChatHistory, settings: PromptExecutionSettings, function_invoke_attempt: int 0, ) - AsyncGenerator[list[StreamingChatMessageContent], Any]: Send a streaming chat request to the AI service. raise NotImplementedError(The _inner_get_streaming_chat_message_contents method is not implemented.)实现细节上有两点值得注意两个内部方法都位于Internal methods to be implemented by the derived classes区域明确的代码注释规定了它们的契约接收ChatHistory与PromptExecutionSettings返回list[ChatMessageContent]或以异步生成器产出list[StreamingChatMessageContent]流式内部方法额外接收一个function_invoke_attempt: int 0参数用于标记当前处于自动函数调用循环中的第几轮该信息会随流式消息内容一并传递便于下游区分这是第几次调用模型产生的内容例如 Anthropic 连接器 在实现时就把该参数透传给_send_chat_stream_request由于流式内部方法签名是异步生成器抛异常后函数体内还需要if False: yield这样的哑代码来满足 mypy 对异步迭代器返回类型的检查源码中对此有专门注释。TextCompletionClientBase采用完全对偶的结构ADR 中明确指出 TextCompletionClientBase will be having a similar structure在 text_completion_client_base.py 中新增_inner_get_text_contents与_inner_get_streaming_text_contents分别用于文本补全text completion场景下的单次调用与流式单次调用其get_text_contents/get_streaming_text_contents公共方法则直接委托给内部方法。核心改动二SUPPORTS_FUNCTION_CALLING类变量第二个核心改动是在ChatCompletionClientBase中引入一个ClassVar[bool]类型的类变量用于标记该连接器是否支持 function calling。它在基类中的默认值为False由派生类按需覆盖并被get_chat_message_contents/get_streaming_chat_message_contents的默认实现读取以决定是否走自动函数调用编排路径。ADR 中的示例代码class ChatCompletionClientBase(AIServiceClientBase, ABC): Base class for chat completion AI services. SUPPORTS_FUNCTION_CALLING: ClassVar[bool] False ...以及一个支持函数调用的模拟实现class MockChatCompletionThatSupportsFunctionCalling(ChatCompletionClientBase): SUPPORTS_FUNCTION_CALLING: ClassVar[bool] True override async def get_chat_message_contents( self, chat_history: ChatHistory, settings: PromptExecutionSettings, **kwargs: Any, ) - list[ChatMessageContent]: if not self.SUPPORTS_FUNCTION_CALLING: return ... ...注在最终落地版本中该模拟类示例中的if not self.SUPPORTS_FUNCTION_CALLING分支逻辑已被上移到基类的默认实现中派生类不再需要自己判断。使用ClassVar[bool]而非实例属性是因为该能力是类级别的一个连接器是否支持 function calling 由实现决定与该实例的配置api key、model id 等无关因此它应当作为类属性存在子类通过类级覆盖即可声明能力无需在__init__中重复设置。默认实现自动函数调用循环如何运转SUPPORTS_FUNCTION_CALLING与_inner_*方法最终在基类的公共默认实现中汇合。以 get_chat_message_contents 为例其完整流程为深拷贝并规范化 settingscopy.deepcopy(settings)避免修改调用方传入的对象若非本连接器对应的 settings 类型则通过get_prompt_execution_settings_from_settings转换快速路径若not self.SUPPORTS_FUNCTION_CALLING说明连接器不支持函数调用直接调用self._inner_get_chat_message_contents(chat_history, settings)并返回——这也是不开启函数调用时几乎所有场景走的路径校验与配置若settings.function_choice_behavior非空则要求kwargs中必须携带kernel否则抛出ServiceInvalidExecutionSettingsError并调用_verify_function_choice_settings做连接器级校验随后调用function_choice_behavior.configure(...)通过_update_function_choice_settings_callback()把可用函数列表写入请求参数如 OpenAI 的tools、Anthropic 的tools等无自动调用时退化为单次调用若function_choice_behavior为空、或auto_invoke_kernel_functions为False则同样直接走内部方法自动调用主循环在use_span(...)OpenTelemetry spanspan 名为AUTO_FUNCTION_INVOCATION_SPAN_NAME包裹下循环最多maximum_auto_invoke_attempts次调用_inner_get_chat_message_contents得到本次模型回复从completions[0].items中过滤出所有FunctionCallContent若数量为 0说明模型给出了最终答复直接返回把含工具调用的 assistant 消息追加进chat_history用asyncio.gather并行执行kernel.invoke_function_call处理全部工具调用多个函数调用同时执行若任一调用结果terminate True则调用merge_function_results合并结果并返回循环结束后达到最大尝试次数_reset_function_choice_settings(settings)清空工具配置再做一次不带函数调用的最终调用后返回。流式版本 get_streaming_chat_message_contents 结构类似额外要点包括单轮内把流式产出累积到all_messages通过reduce(lambda x, y: x y, all_messages)拼接出完整消息以提取FunctionCallContent工具执行结果经merge_streaming_function_results合并后根据_yield_function_result_messages判断是否向上游产出通过_get_ai_model_id为合并的流式消息补全ai_model_id保证多个流式消息可以正确拼接。与自动函数调用相关的可配置项定义在 function_choice_behavior.py 中默认最大自动调用次数DEFAULT_MAX_AUTO_INVOKE_ATTEMPTS 5并提供FunctionChoiceBehavior.Auto()模型自行决定是否调用、NoneInvoke()模型只描述如何调用但不执行、Required()模型必须调用指定函数三种工厂方法auto_invoke_kernel_functions属性由maximum_auto_invoke_attempts 0推导这也正是上述默认实现第 4 步判断的依据。各连接器的落地情况谁把开关打开了通过检索仓库源码可以看到SUPPORTS_FUNCTION_CALLING: ClassVar[bool] True已在下列连接器中显式覆盖连接器文件是否支持函数调用OpenAI / Azure OpenAIopen_ai_chat_completion_base.pyTrueAnthropicanthropic_chat_completion.pyTrueAzure AI Inferenceazure_ai_inference_chat_completion.pyTrueAWS Bedrockbedrock_chat_completion.pyTrueGoogle GeminiGoogle AIgoogle_ai_chat_completion.pyTrueGoogle Vertex AIvertex_ai_chat_completion.pyTrueMistral AImistral_ai_chat_completion.pyTrueOllamaollama_chat_completion.pyTrueONNX GenAIonnx_gen_ai_chat_completion.pyFalse保留默认值Realtime Client Baserealtime_client_base.pyFalse保留默认值这也印证了 ADR 的判断是否开启函数调用编排完全由连接器自身的能力决定不支持函数调用的连接器如本地 ONNX 推理直接走快速路径零额外开销。以 Anthropic 连接器 为例可以看到新抽象的完整配合方式覆盖_inner_get_chat_message_contents并用trace_chat_completion(MODEL_PROVIDER_NAME)装饰——这就是 ADR 所提到的单次模型调用的公共追踪接口的落地形态每次真实的模型请求都会被埋点记录覆盖_update_function_choice_settings_callback返回update_settings_from_function_call_configuration把 Kernel 的函数元数据翻译成 Anthropic 的tools请求字段覆盖_reset_function_choice_settings在最大尝试次数用尽后清空tool_choice与tools保证最后兜底调用不再带工具流式路径则把function_invoke_attempt透传到每条StreamingChatMessageContent让自动调用循环中的每一轮流式内容都可被区分。兼容性设计为什么不用abstractmethodADR 的 Revision 明确记录了一个重要的兼容性决策这两个新方法不添加abstractmethod装饰器。原因在于在 ADR 通过之前社区中已经存在基于旧版ChatCompletionClientBase实现的自定义连接器它们直接覆写了get_chat_message_contents等公共方法。如果新内部方法被声明为抽象方法任何未同步升级的自定义连接器都会在实例化时直接报错抽象类无法实例化形成硬性破坏性变更breaking change。改为基类默认实现中raise NotImplementedError后未升级的旧连接器依然可以实例化、继续覆写公共方法正常工作新编写的内置连接器若忘记实现内部方法只会在真正发起模型请求时才抛出NotImplementedError且错误信息清晰指向未实现的方法名存量用户获得平滑迁移窗口新抽象能力逐步铺开。这一软抽象策略是兼容性优先的典型工程取舍值得自定义连接器作者留意升级到新版 SDK 后建议尽快将实现迁移到_inner_*方法上以自动获得函数调用编排、流式合并与遥测追踪等公共能力。小结写给连接器作者的接入指南综合 ADR 与当前源码编写一个新的 chat completion 连接器并完整获得 Semantic Kernel 公共能力需要遵循以下约定继承ChatCompletionClientBasechat 场景或TextCompletionClientBasetext 补全场景实现get_prompt_execution_settings_class等AIServiceClientBase要求的接口实现_inner_get_chat_message_contents与_inner_get_streaming_chat_message_contentschat 场景或_inner_get_text_contents与_inner_get_streaming_text_contentstext 场景方法体内只需完成单次请求模型并转换为 Semantic Kernel 内容类型这一件事若模型支持 function calling将SUPPORTS_FUNCTION_CALLING覆盖为True并实现_update_function_choice_settings_callback把函数元数据翻译成厂商的工具参数与_reset_function_choice_settings清理工具参数若模型不支持保持False即可框架会自动跳过全部编排逻辑可选的_verify_function_choice_settings用于对 settings 做厂商级校验_prepare_chat_history_for_request用于定制消息序列化格式使用trace_chat_completion/trace_streaming_chat_completion装饰内部方法即可获得单次模型调用的 OpenTelemetry 追踪埋点。通过这套设计Semantic Kernel Python 将模型调用与函数编排清晰分层_inner_*只管发请求公共方法负责深度拷贝 settings、校验 kernel、配置工具、并行执行函数、合并流式结果与自动调用循环而SUPPORTS_FUNCTION_CALLING这一枚类级开关则决定了每个连接器最终接入哪一层能力。【免费下载链接】semantic-kernelIntegrate cutting-edge LLM technology quickly and easily into your apps项目地址: https://gitcode.com/GitHub_Trending/se/semantic-kernel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考