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 仓库中的架构决策记录（ADR）0052-python-ai-connector-new-abstract-methods.md 为主体，讲解 Python 版ChatCompletionClientBase与TextCompletionClientBase如何通过新增一组_inner_*内部方法与SUPPORTS_FUNCTION_CALLING类变量，把"一次模型调用"与"自动函数调用（auto function invocation）编排"解耦为两层。读完本文，你将理解该抽象层的设计动机、默认实现的工作流程，以及如何基于这套约定为 Semantic Kernel 编写新的 AI 连接器。

背景与问题：自动函数调用带来的分层需求

在 Semantic Kernel 中，ChatCompletionClientBase是所有聊天补全（chat completion）AI 服务连接器（connector）的基类。在引入本 ADR 之前，该类只暴露两个抽象方法：

• get_chat_message_contents：非流式地获取模型返回的聊天消息内容列表；
• get_streaming_chat_message_contents：以异步生成器方式获取流式聊天消息内容。

这两个方法为上层（Kernel、Agent、插件编排）提供了与具体模型无关的标准化接口。

随着众多模型开始支持 function calling，Semantic Kernel 实现了auto function invocation（自动函数调用）特性：当模型在回复中请求调用某个 Kernel 函数时，框架会自动执行该函数、把结果写回ChatHistory并再次请求模型，循环往复，直到模型给出最终答案。开发者无需手动解析 function call、逐个调用插件再拼装消息，开发体验因此大幅简化。

但自动函数调用有一个重要的副作用：一次对get_chat_message_contents（或流式版本）的调用，底层可能触发对模型的多次调用（一次原始请求 + 多轮工具调用往返）。这说明原有的两个抽象方法承担了两种职责：①真正与模型进行单次 HTTP 通信；②围绕单次通信做函数调用编排。这正是一个引入新抽象层的绝佳机会——让"单次模型调用"成为独立的、可被专门追踪与监控的单元。

设计目标：三层收益

ADR 明确了这次引入抽象层的三个收益：

1. 简化连接器实现：在基类中为get_chat_message_contents和get_streaming_chat_message_contents提供默认实现，派生类只需实现真正发送单次请求的内部方法，无需重复编写函数调用编排逻辑；
2. 可观测性：可以围绕"单次模型调用"建立公共的追踪（tracing）接口，提升系统的监控与管理能力（这一点在源码中体现为@trace_chat_completion、@trace_streaming_chat_completion装饰器，见下文）；
3. 降低新连接器的接入成本：新 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 NotImplementedError

async 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 为例，其完整流程为：

1. 深拷贝并规范化 settings：copy.deepcopy(settings)避免修改调用方传入的对象；若非本连接器对应的 settings 类型，则通过get_prompt_execution_settings_from_settings转换；
2. 快速路径：若not self.SUPPORTS_FUNCTION_CALLING，说明连接器不支持函数调用，直接调用self._inner_get_chat_message_contents(chat_history, settings)并返回——这也是不开启函数调用时几乎所有场景走的路径；
3. 校验与配置：若settings.function_choice_behavior非空，则要求kwargs中必须携带kernel（否则抛出ServiceInvalidExecutionSettingsError），并调用_verify_function_choice_settings做连接器级校验；随后调用function_choice_behavior.configure(...)，通过_update_function_choice_settings_callback()把可用函数列表写入请求参数（如 OpenAI 的tools、Anthropic 的tools等）；
4. 无自动调用时退化为单次调用：若function_choice_behavior为空、或auto_invoke_kernel_functions为False，则同样直接走内部方法；
5. 自动调用主循环：在use_span(...)（OpenTelemetry span，span 名为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已在下列连接器中显式覆盖：

这也印证了 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，让自动调用循环中的每一轮流式内容都可被区分。

兼容性设计：为什么不用@abstractmethod

ADR 的 Revision 明确记录了一个重要的兼容性决策：这两个新方法不添加@abstractmethod装饰器。

原因在于：在 ADR 通过之前，社区中已经存在基于旧版ChatCompletionClientBase实现的自定义连接器（它们直接覆写了get_chat_message_contents等公共方法）。如果新内部方法被声明为抽象方法，任何未同步升级的自定义连接器都会在实例化时直接报错（抽象类无法实例化），形成硬性破坏性变更（breaking change）。

改为"基类默认实现中raise NotImplementedError"后：

• 未升级的旧连接器依然可以实例化、继续覆写公共方法正常工作；
• 新编写的内置连接器若忘记实现内部方法，只会在真正发起模型请求时才抛出NotImplementedError，且错误信息清晰指向未实现的方法名；
• 存量用户获得平滑迁移窗口，新抽象能力逐步铺开。

这一"软抽象"策略是兼容性优先的典型工程取舍，值得自定义连接器作者留意：升级到新版 SDK 后，建议尽快将实现迁移到_inner_*方法上，以自动获得函数调用编排、流式合并与遥测追踪等公共能力。

小结：写给连接器作者的接入指南

综合 ADR 与当前源码，编写一个新的 chat completion 连接器并完整获得 Semantic Kernel 公共能力，需要遵循以下约定：

1. 继承ChatCompletionClientBase（chat 场景）或TextCompletionClientBase（text 补全场景），实现get_prompt_execution_settings_class等AIServiceClientBase要求的接口；
2. 实现_inner_get_chat_message_contents与_inner_get_streaming_chat_message_contents（chat 场景），或_inner_get_text_contents与_inner_get_streaming_text_contents（text 场景），方法体内只需完成"单次请求模型并转换为 Semantic Kernel 内容类型"这一件事；
3. 若模型支持 function calling，将SUPPORTS_FUNCTION_CALLING覆盖为True，并实现_update_function_choice_settings_callback（把函数元数据翻译成厂商的工具参数）与_reset_function_choice_settings（清理工具参数）；若模型不支持，保持False即可，框架会自动跳过全部编排逻辑；
4. 可选的_verify_function_choice_settings用于对 settings 做厂商级校验，_prepare_chat_history_for_request用于定制消息序列化格式；
5. 使用@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