数据截至 (上游 commit 5e1f1fb87d9a)
第 03 章 · 主线:自动函数调用循环与三类过滤器
本章讲什么: 模型说「我要调
math-Add」之后,到这次调用的结果重新进入对话之间,SK 到底做了多少事。这是全书工程含量最高的一章——把工具暴露、schema 生成、循环本体、单次调用的容错、以及包在外面的三类过滤器,一层层拆开。前置: 第 01 章 的
Kernel/KernelFunction/ 插件,第 02 章 的ChatHistory与内容模型。Agent 层怎么复用这套东西,见 第 04 章——本章只讲底座。
3.1 先把问题讲清楚
大模型不会真的执行任何东西。它只会在回复里吐一段结构化的话:「请帮我调用名叫 math-Add 的函数,参数是 {"input": 3, "amount": 4}」。
从这句话到「结果回到对话里,模型继续往下说」,中间有五件必须有人干的事:
| 序号 | 要干的事 | 白话 |
|---|---|---|
| ① | 告诉模型有哪些工具可用 | 把插件里的函数翻译成 provider 认识的 tools 数组 |
| ② | 接住模型的调用请求 | 从回复里挑出 FunctionCallContent |
| ③ | 找到并执行真函数 | 名字对不对、参数齐不齐、执行会不会炸 |
| ④ | 把结果塞回对话 | 变成一条 tool 角色的消息追加进 ChatHistory |
| ⑤ | 再问模型一次 | 让它看着结果继续——可能又要调工具,于是循环 |
SK 把这五件事全塞进了 ChatCompletionClientBase 的一个 for 循环里,让调用方一次 await 就拿到最终答案,中间几轮工具往返完全不用管。
一张图看全流程
从上往下是一轮的时间顺序;右边虚线框是「本轮结束后回到循环开头」。
用户消息 + ChatHistory
│
┌────────────▼─────────────┐
│ ① 暴露工具 │ FunctionChoiceBehavior.configure
│ 把可用函数写进 settings │ → settings.tools / tool_choice
└────────────┬─────────────┘
│
┌────────────▼─────────────┐
│ 发请求给模型 │ _inner_get_chat_message_contents
└────────────┬─────────────┘
│
┌───────▼────────┐ 没有工具调用
│ ② 有 tool call?├──────────────► 直接返回,循环结束
└───────┬────────┘
│ 有(可能好几个)
┌────────────▼─────────────┐
│ ③ 并行执行每个调用 │ asyncio.gather(invoke_function_call…)
│ ┌───────────────────┐ │
│ │ 过滤器洋葱 │ │ auto_function_invocation filters
│ │ └─ 真函数 │ │
│ └───────────────────┘ │
└────────────┬─────────────┘
│
┌────────────▼─────────────┐
│ ④ 结果写回 ChatHistory │ FunctionResultContent → tool 消息
└────────────┬─────────────┘
│ terminate?
┌───────▼────────┐ 是
│ ⑤ 回到循环开头 ├──────► 立刻返回工具结果
└───────┬────────┘
│ 否,轮次 +1;超上限则「关掉工具」再问最后一次
└ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┘
后面五节按这张图逐格展开。
3.2 第一步:决定给模型看哪些工具(FunctionChoiceBehavior)
它要解决的小问题: Kernel 上可能挂了几十个函数,但这一次对话只该让模型看到其中几个;而且要能控制「模型可以不调」「必须调一个」「只准描述不准真调」。
SK 把这三件事全压进一个对象:FunctionChoiceBehavior(python/semantic_kernel/connectors/ai/function_choice_behavior.py:26-224)。
三个 类方法 = 三种策略
| 类方法 | type_ | maximum_auto_invoke_attempts 默认 | 语义 |
|---|---|---|---|
Auto(auto_invoke=True) | AUTO | 5(DEFAULT_MAX_AUTO_INVOKE_ATTEMPTS) | 模型自己决定调不调、调哪个 |
Required(auto_invoke=True) | REQUIRED | 1 | 模型必须从给定函数里挑一个 |
NoneInvoke() | NONE | 0 | 工具照样告诉模型,但一个都不真调 |
三处源码分别是 function_choice_behavior.py:126、:173、:149 的 kwargs.setdefault(...)。注意 Required 默认只给 1 次——因为「必须调」的语义下,让它循环 5 轮很容易变成死循环。
一个数字兼职当开关
整个类里没有 auto_invoke: bool 这个字段。是否自动执行,由次数是否大于 0 推出来:
@property
def auto_invoke_kernel_functions(self):
"""Return True if auto_invoke_kernel_functions is enabled."""
return self.maximum_auto_invoke_attempts > 0
function_choice_behavior.py:65-68。反过来 setter(:70-73)把 True 写成 5、False 写成 0。
这个设计的直接后果:NoneInvoke() 之所以「不真调」,不是因为有个开关关了,而是因为它的上限是 0——循环体在 chat_completion_client_base.py:130-134 就被这个属性挡掉,压根不进循环。
filters 白名单:四个键, 两两互斥
filters 字段(function_choice_behavior.py:59-62)接受四个键:
| 键 | 作用 | 匹配对象 |
|---|---|---|
included_plugins | 只暴露这些插件 | 插件名 |
excluded_plugins | 排除这些插件 | 插件名 |
included_functions | 只暴露这些函数 | 全限定名(math-Add) |
excluded_functions | 排除这些函数 | 全限定名 |
included_* 和 excluded_* 同类不能同时用,否则直接 ValueError——校验在 python/semantic_kernel/functions/kernel_function_extension.py:399-402,过滤逻辑在 :392-401 的 get_list_of_function_metadata_filters。
坑: 函数级过滤用的是全限定名,分隔符是连字符
-(DEFAULT_FULLY_QUALIFIED_NAME_SEPARATOR,python/semantic_kernel/const.py:9),不是点号。所以from_dict专门做了一次替换:name.replace(".", "-")(function_choice_behavior.py:196)——用户在 YAML 里习惯写math.Add,这里帮忙纠正。
configure:把「可用函数」写进 provider 的 settings
configure 本身只有十来行(function_choice_behavior.py:90-103),干三件事:
enable_kernel_functions为假就直接返回;- 调
get_config(kernel)拿到FunctionCallChoiceConfiguration(里面就一个available_functions列表); - 把 config、settings、
type_交给 connector 传进来的回调update_settings_callback。
关键是第 3 步的解耦:FunctionChoiceBehavior 自己不知道 OpenAI 的字段叫 tools 还是 Anthropic 的叫别的。它只负责「算出哪些函数可用」,格式化交给 connector。
回调从 ChatCompletionClientBase._update_function_choice_settings_callback() 取(python/semantic_kernel/connectors/ai/chat_completion_client_base.py:390-398),基类默认是个空 lambda。OpenAI connector 覆盖它,返回 update_settings_from_function_call_configuration(python/semantic_kernel/connectors/ai/open_ai/services/open_ai_chat_completion_base.py:150-154),那个函数往 settings 上写两个字段:
settings.tool_choice = type
settings.tools = [
kernel_function_metadata_to_function_call_format(f)
for f in function_choice_configuration.available_functions
]
python/semantic_kernel/connectors/ai/function_calling_utils.py:35-39。注意 hasattr(settings, "tool_choice") and hasattr(settings, "tools") 的守卫(:32-34)——settings 类没这俩字段就什么都不做,静默跳过。
get_config 还有第二个用途:开 span 时拿函数清单打标签,见 §3.9。
分叉点一览
调用方给的 behavior 决定走哪条路(chat_completion_client_base.py:112-137):
SUPPORTS_FUNCTION_CALLING == False ──► 直接单次请求(:112-113)
behavior 非空但 kernel 是 None ──────► 抛 ServiceInvalidExecutionSettingsError(:116-118)
behavior is None ─┐
behavior.auto_invoke == False ─┴─► 配好 tools,单次请求就返回(:130-134)
其余 ──► 进 auto invoke 循环(:137)
第三条是 NoneInvoke() 的实际归宿:工具照样出现在请求里(第 121-128 行的 configure 已经跑过了),模型也会返回 FunctionCallContent,只是没人去执行它——这正好用来做「让模型说说它打算怎么调」的场景。
3.3 工具 schema 是怎么长出来的
它要解决的小问题: Python 函数签名 def add(self, input: Annotated[int, "第一个加数"], amount: int) -> int 要变成一份 JSON Schema,模型才知道怎么填参数。
SK 分两步走,而且第一步在函数注册的那一刻就完成了,不是每次请求现算。
第一步:参数元数据自带 schema
KernelParameterMetadata 有个 pydantic 前置校验器,构造时若没给 schema_data 就现场推一份:
@model_validator(mode="before")
@classmethod
def form_schema(cls, data: Any) -> Any:
"""Create a schema for the parameter metadata."""
python/semantic_kernel/functions/kernel_parameter_metadata.py:24-35,真正推断在 :37-62 的 infer_schema——有真实类型对象(type_object)就走 KernelJsonSchemaBuilder.build,只有类型名字符串就走 build_from_type_name。
一个贴心细节:只有类型名时,默认值会被拼进 description(:52-59),变成 "第一个加数 (default value: 0)"。因为 JSON Schema 的 default 字段很多模型不看,写进描述里反而更管用。
第二步:schema builder 的分派表
KernelJsonSchemaBuilder.build(python/semantic_kernel/schema/kernel_json_schema_builder.py:36-63)是一串 if,顺序就是优先级:
| 判断 | 走向 | 源码 |
|---|---|---|
| 是字符串类型名 | build_from_type_name | :117-138 |
是 KernelBaseModel 实例 / 有 __annotations__ | build_model_schema | :65-107 |
是 Enum 子类 | build_enum_schema | :220-240 |
有 __args__(泛型) | handle_complex_type | :154-218 |
| 其余 | 查 TYPE_MAPPING 表 | :141-152 |
TYPE_MAPPING(:12-31)把 Python 类型和类型名同时映射到 JSON Schema 类型,查不到一律给 "object"——不报错,降级。
几个值得记的处理:
Optional[T]被翻成"type": [T, "null"]而不是anyOf(:199-206),这是 OpenAI structured output 要求的写法;- 联合类型的字符串形式用逗号分隔(
PARSED_ANNOTATION_UNION_DELIMITER),build_from_type_name见到逗号就拆成anyOf(:128-132); dict[str, X]在 Python 3.10 上会产出{"type": "object"},代码专门补了一个空properties抹平版本差异(:182-184)。
第三步:拼成 provider 的 tools 条目
最后一层薄薄的格式化:
def kernel_function_metadata_to_function_call_format(
metadata: "KernelFunctionMetadata",
) -> dict[str, Any]:
python/semantic_kernel/connectors/ai/function_calling_utils.py:42-59。它把每个参数的 schema_data 当作 properties 的一项,把 is_required 的挑出来当 required。
这里有个隐藏的开关: 两处推导式都带 if param.include_in_function_choices(:54、:56)。这个字段(kernel_parameter_metadata.py:22)默认 True,但可以在 @kernel_function 的 Annotated 里用字典关掉:
# 示意,非源码
from typing import Annotated
@kernel_function
def summarize(
text: Annotated[str, "要摘要的文本"],
# 这个参数由框架注入,不该让模型看见,更不该让它填
kernel: Annotated["Kernel", {"include_in_function_choices": False}],
) -> str:
...
装饰器解析时还会顺手把它的 is_required 强制改成 False(python/semantic_kernel/functions/kernel_function_decorator.py:190-192),否则「必填但模型看不见」会直接把调用卡死。这是 kernel、arguments、service 这类框架注入参数的标准做法(说明见 :33-36 的 docstring)。
一个具体的前后对照
# 示意,非源码 —— 左边是你写的,右边是模型收到的
@kernel_function(name="Add", description="两数相加")
def add(self,
input: Annotated[int, "第一个加数"],
amount: Annotated[int, "第二个加数"]) -> int:
return input + amount
# ↓ 注册时烤好 schema_data,发请求时拼成:
{
"type": "function",
"function": {
"name": "math-Add", # 插件名-函数名,连字符
"description": "两数相加",
"parameters": {
"type": "object",
"properties": {
"input": {"type": "integer", "description": "第一个加数"},
"amount": {"type": "integer", "description": "第二个加数"}
},
"required": ["input", "amount"]
}
}
}
重点看 name:模型之后回传的就是这个 math-Add,FunctionCallContent.__init__ 会按连字符拆回 plugin_name / function_name(python/semantic_kernel/contents/function_call_content.py:81-85)。
3.4 循环本体(非流式)
它要解决的小问题: 模型可能连着调好几轮工具才给出最终答案。调用方不该为此写 while。
先用伪代码建立直觉
# 示意,非源码 —— 循环的骨架
for attempt in range(max_attempts): # 默认 5
reply = await ask_model(history, settings)
calls = [x for x in reply.items if is_function_call(x)]
if not calls:
return reply # 模型不调工具了,收工
history.add(reply) # 先把「我要调工具」这条记下
results = await gather(*[run(c) for c in calls]) # 并行跑,结果自动进 history
if any(r.terminate for r in results):
return merge(history.last_n(len(results))) # 有人喊停,直接交工具结果
else:
settings.tools = None # 用光了次数:摘掉工具
return await ask_model(history, settings) # 逼模型用自然语言收尾
重点看 else——它挂在 for 上而不是 if 上,是 Python 的 for...else:循环把 range 走完(没被 return 打断)才执行。这就是「达到上限后的兜底」。
真实实现:逐行要点
真源码在 python/semantic_kernel/connectors/ai/chat_completion_client_base.py:137-174,不到 40 行。逐格对应:
| 行号 | 做什么 | 值得注意的点 |
|---|---|---|
:137 | use_span(...) 包住整个循环 | 一个 span 覆盖全部轮次,不是每轮 一个 |
:138 | for request_index in range(maximum_auto_invoke_attempts) | 轮次号一路往下传,进 filter context |
:139 | _inner_get_chat_message_contents | 唯一的抽象点,各 connector 实现 |
:142 | 从 completions[0].items 挑 FunctionCallContent | 只看第 0 个 completion |
:143-144 | 没有调用 → 原样返回 | 循环的正常出口 |
:147 | 把模型那条 assistant 消息加进 history | 必须先加,否则 tool 消息没有配对的 tool_call |
:154-167 | asyncio.gather 并行跑所有调用 | 结果由 invoke_function_call 自己写回 history |
:169-170 | 任一结果 terminate → merge_function_results | 提前退出 |
:171-174 | for...else:重置 settings + 无工具再问一次 | 上限兜底 |
三处不显然的设计
(a) 只取 completions[0]。 注释直说了理由:多 completion 的情况「应该已经被 _verify_function_choice_settings 挡掉」(:140-141)。OpenAI connector 的实现确实这么干——number_of_responses > 1 时直接抛错(open_ai_chat_completion_base.py:141-149)。这是「用前置校验换掉循环里的分支」的典型手法。
(b) 结果不是 return 出来的,是被写进 history 的。 gather 收到的 results 里,元素要么是 None,要么是一个 AutoFunctionInvocationContext。只有 terminate 为真时才返回 context(python/semantic_kernel/kernel.py:463)。所以 results 的唯一用途就是检查有没有人喊停;真正的工具输出走的是 chat_history 这条副作用通道。
(c) 达上限后的最后一问要先「摘掉工具」。 若不摘,模型看见 tools 还在,大概率又要调一次,这一轮就白费了。摘的动作是 _reset_function_choice_settings(settings)(chat_completion_client_base.py:173),基类空实现(:400-408),OpenAI 版把 tool_choice 和 tools 都置回 None(open_ai_chat_completion_base.py:156-161)。
注意这一步动的是
settings的深拷贝——方法一进来就settings = copy.deepcopy(settings)(:107),所以调用方传进来的 settings 对象不会被这次调用改坏。
terminate 的返回值形态
terminate 触发时返回的不是模型的话,而是 merge_function_results(chat_history.messages[-len(results):])(:170)。这个函数(function_calling_utils.py:103-122)把最后 N 条消息里的 FunctionResultContent 全抠出来,合并成一条 role=TOOL 的 ChatMessageContent。
也就是说:终止时调用方拿到的是工具的原始输出,不是模型的总结。 这是有意的——filter 喊停通常正是因为 「这个工具结果本身就是最终答案,别再让模型加工了」。
3.5 流式版:四处不一样
流式循环在 chat_completion_client_base.py:256-318,骨架一样,但有四个必须知道的差异。
| 方面 | 非流式(:137-174) | 流式(:256-318) |
|---|---|---|
| 拿到调用的方式 | 一次返回,直接取 items | 边收边 yield,收完用 reduce 把碎片加起来(:279) |
| 是否传轮次给 connector | 不传 | 传 request_index 作 function_invoke_attempt(:261-263) |
| 工具结果怎么给调用方 | 只在 terminate 时返回 | 每轮都 yield 一次合并后的结果消息(:309-315) |
| 达上限后 | for...else 再问一次(无工具) | 没有 else 分支,循环跑完直接结束 |
先 yield 再合并
流式的难点是:FunctionCallContent 的参数 JSON 是一个字符一个字符流过来的,你必须先把整条流收齐才知道模型到底要调什么。
代码的做法是「照收照转,同时攒着」:
async for messages in self._inner_get_streaming_chat_message_contents(
chat_history, settings, request_index
):
for msg in messages:
if msg is not None:
all_messages.append(msg)
if not function_call_returned and any(
isinstance(item, FunctionCallContent) for item in msg.items
):
function_call_returned = True
yield messages
:261-271。用户那边照样看到逐字输出;循环这边攒下 all_messages,流结束后一把 reduce(lambda x, y: x + y, all_messages)(:279)拼成完整消息。
拼接靠的是 FunctionCallContent.__add__(python/semantic_kernel/contents/function_call_content.py:108-132):它按 id / index / call_id 校验是不是同一个调用,然后 combine_arguments 把参数字符串接起来。
每轮都吐工具结果
流式版每轮 gather 完,不管终不终止都要把工具结果发给调用方:
function_result_messages = merge_streaming_function_results(
messages=chat_history.messages[-len(results):],
ai_model_id=ai_model_id,
function_invoke_attempt=request_index,
)
if self._yield_function_result_messages(function_result_messages):
yield function_result_messages
:309-315。merge_streaming_function_results(function_calling_utils.py:125-158)和非流式版做同样的合并,只是产出 StreamingChatMessageContent,并带上 ai_model_id 和 function_invoke_attempt——这两个字段是为了让两条流式消息能相加(不同轮次的消息不该被合并到一起)。
_yield_function_result_messages(:436-441)是个防空守卫:结果为空就不 yield,避免给调用方发一条没内容的消息。
没有 else 分支意味着什么
非流式跑满 5 轮会「摘掉工具再问一次」,保证一定有段自然语言收尾。流式不会——for 跑完就是流结束,调用方最后收到的可能是一条工具结果消息,而不是模型的话。做 UI 时要自己处理这个情况。
3.6 单次工具调用:一条「几乎不抛异常」的流水线
Kernel.invoke_function_call(python/semantic_kernel/kernel.py:326-463)是本章最该细读的一段。它的设计原则一句话:
模型犯的任何错,都不该炸掉程序,而应该变成一段文字告诉模型「你错在哪、重来」。
五道关卡
FunctionCallContent
│
①名字为空 / 不在白名单 / 函数不存在 ──► 文本:"该工具不在提供的工具列表里,请核对名字"
│ 通过 (:358-368)
②参数缺失 / 有多余参数 ───────────────► 文本:"缺少 [x];收到多余 [y];请对照签名修正"
│ 通过 (:383-397)
③参数不是合法 JSON ───────────────────► 文本:"参数格式错误,必须是 JSON,请重试"
│ 通过 (:401-408)
④必填个数再兜一次 ────────────────────► 文本:"需要 N 个参数,只收到 M 个"
│ 通过 (:410-424)
⑤过滤器洋葱 → 真函数执行
│ 执行内部炸了也被 handler 接住 ──► 文本:"调用 X 时出错:<异常信息>"
▼ (:477-484)
结果 deepcopy → FunctionResultContent → 写进 ChatHistory
四道关卡的出口形态完全一样:构造 FunctionResultContent.from_function_call_content_and_result(...)、chat_history.add_message(frc.to_chat_message_content())、return None。返回 None 意味着「这次调用没有喊停」,循环照常进入下一轮——模型看到那段错误文本,自己纠正重试。
白名单校验:behavior 的 filters 在这里第二次生效
if function_behavior is not None and function_behavior.filters:
allowed_functions = [
func.fully_qualified_name for func in self.get_list_of_function_metadata(function_behavior.filters)
]
if function_call.name not in allowed_functions:
raise FunctionExecutionException(
f"Only functions: {allowed_functions} are allowed, {function_call.name} is not allowed."
)
kernel.py:342-349。为什么已经只把白名单函数发给模型了,这里还要再查一遍?因为模型可以幻觉出一个没给过的名字,而 kernel 上真的挂着那个函数。少了这道校验,一个「只准用 math 插件」的会话就可能被模型骗着调到 email-Send。
没传 function_behavior 时不做校验,但会打一条 debug 日志明确提示「本次未做白名单校验」(:350-356)——这是留给直接调 invoke_function_call 的调用方的警告。
注意 raise 之后立刻被同一个 try 的 except Exception 接住(:358),转成给模型的文本。异常在这里只是控制流,不外泄。
参数校验的两层
第一层按名字比对(:374-397):算出 missing_params(必填但没给)和 unexpected_params(给了但签名里没有),两者有一个非空就拼一句话回去。消息刻意包含排序后的参数名(sorted(...)),让模型能精确定位。
第二层按个数(:410-424):必填参数数量 vs 收到的参数数量。这层在第一层之后,属于额外的保险。
parsed_args 来自 function_call.to_kernel_arguments()(python/semantic_kernel/contents/function_call_content.py:171-178),它内部的 parse_arguments(:151-169)还做了一次容错解析:JSON 解析失败时,把非转义的单引号替换成双引号再试一次——专治模型输出 Python 风格的 {'a': 1}。两次都失败才抛 FunctionCallInvalidArgumentsException,被 :401 接住。
返回值 deepcopy 快照
# Snapshot the tool's return value so later mutations don't leak back
if invocation_context.function_result and invocation_context.function_result.value is not None:
invocation_context.function_result.value = deepcopy(invocation_context.function_result.value)
kernel.py:450-452。工具返回的如果是个可变对象(比如插件内部持有的 list),后续轮次里插件把它改了,已经写进 ChatHistory 的那条记录也会跟着变——对话历史就成了「会自己变的历史」。深拷贝把这条路堵死。
流式与非流式的消息形态
is_streaming = any(isinstance(message, StreamingChatMessageContent) for message in chat_history.messages)
message = frc.to_streaming_chat_message_content() if is_streaming else frc.to_chat_message_content()
kernel.py:458-459。方法签名上明明有 is_streaming 参数(:335),这里却用扫描 chat_history 的结果把它覆盖了。判据是「历史里出现过流式消息,那这条也该是流式的」——保证同一条 history 里的消息类型一致,后面 reduce 相加时不会因为类型不匹配而失败。
两个构造方法都很短,只是把 role 定成 TOOL:python/semantic_kernel/contents/function_result_content.py:161-165 和 :167-171。
3.7 执行体与异常兜底
洋葱最里面那层是 _inner_auto_function_invoke_handler(kernel.py:465-484),二十行:
result = await context.function.invoke(
context.kernel,
context.arguments,
metadata=context.function_call_content.metadata | context.function_call_content.to_dict()
if context.function_call_content
else {},
)
:468-474。两个细节:
- metadata 是合并出来的:原始 metadata 加上整个调用内容的字典形式。这让下游的
function_invocationfilter 和遥测能看到tool_call_id之类的信息; context.function.invoke是第 01 章那个通用调用入口——也就是说,自动函数调用最终还是走了普通函数调用的全部管线,包括function_invocation过滤器。两层洋葱是嵌套关系,不是并列。
异常兜底在 :477-484:任何异常都被记进日志,然后把错误文本塞进 context.function_result.value,函数正常返回。所以外层 invoke_function_call 拿到的永远是一个有值的 result,不需要再包一层 try。
这也是「工具报错」和「工具不存在」的处理归口不同的原因:前者在这里(异常 → 文本),后者在 §3.6 的关卡①(校验失败 → 文本)。两条路殊途同归,最后都是一段给模型看的话。
3.8 过滤器管线:洋葱是怎么套出来的
它要解决的小问题: 想在函数执行前后插一段自己的代码(打日志、鉴权、缓存、改参数、改结果、喊停),又不想改函数本身。
SK 的答案是 ASP.NET 中间件那一套:每个 filter 收 (context, next),自己决定什么时候调 next,以及调完之后再干什么。
注册:头插
getattr(self, FILTER_MAPPING[filter_type.value]).insert(0, (id(filter), filter))
python/semantic_kernel/filters/kernel_filters_extension.py:56。注意是 insert(0) 不是 append——新加的 filter 排在列表最前面。存的是 (id(filter), filter) 元组,那个 id() 是给 remove_filter 用的句柄(:73-106)。
除了 add_filter,还有个装饰器写法 kernel.filter(FilterTypes.X)(:60-71),内部就是调 add_filter。
组装:partial 反向套
def construct_call_stack(self, filter_type, inner_function):
"""Construct the call stack for the given filter type."""
stack: list[Any] = [inner_function]
for _, filter in getattr(self, FILTER_MAPPING[filter_type]):
filter_with_next = partial(filter, next=stack[0])
stack.insert(0, filter_with_next)
return stack[0]
:108-118。每一轮把当前最外层当作下一个 filter 的 next,再把新的塞到最外面。走一遍就清楚了——先加 A 再加 B:
存储顺序(头插的结果): [B, A]
组装过程:
起点 stack = [inner]
遇到 B stack = [B(next=inner), inner]
遇到 A stack = [A(next=B), B(next=inner), inner]
返回 stack[0] = A
运行时的洋葱:
┌─ A:前置代码 ────────────────────────┐
│ ┌─ B:前置代码 ────────────────── ┐ │
│ │ ┌─ inner:真正干活 ────────┐ │ │
│ │ └────────────────────────┘ │ │
│ └─ B:后置代码 ──────────────────┘ │
└─ A:后置代码 ────────────────────────┘
结论(和 add_filter 的 docstring 一致,:39-43):先注册的先执行前置、后执行后置。 「后加的在外面」这个直觉是错的。
签名坑:
partial(filter, next=stack[0])用的是关键字参数。所以你的 filter 第二个参数必须字面叫next,叫call_next会直接TypeError。仓库里的样例全都遵守这一点(如python/samples/concepts/filtering/function_invocation_filters.py:26-29)。
教学示例:一个缓存 filter
# 示意,非源码 —— 演示 next 前后各干一件事
cache = {}
@kernel.filter(FilterTypes.FUNCTION_INVOCATION)
async def caching_filter(context, next):
key = (context.function.fully_qualified_name, str(context.arguments))
if key in cache:
context.result = cache[key] # 直接给结果
return # 不调 next,真函数根本不跑
await next(context) # 放行,让内层执行
cache[key] = context.result # 回来的路上收割结果
重点看不调 next 就等于短路——这是 filter 能做缓存、能做熔断的根本原因。FunctionInvocationContext 的 docstring 也明说了缓存是设计用途之一(python/semantic_kernel/filters/functions/function_invocation_context.py:14-16)。
三类 filter 一览
FilterTypes 只有三个成员(python/semantic_kernel/filters/filter_types.py:6-11):
| 类型 | 包住什么 | context 类 | 挂载点源码 |
|---|---|---|---|
PROMPT_RENDERING | 提示词模板渲染 | PromptRenderContext | python/semantic_kernel/functions/kernel_function_from_prompt.py:281-284 |
FUNCTION_INVOCATION | 任何 KernelFunction 的一次执行 | FunctionInvocationContext | python/semantic_kernel/functions/kernel_function.py:274-277(流式版 :346-349) |
AUTO_FUNCTION_INVOCATION | 自动调用循环里的一次工具调用 | AutoFunctionInvocationContext | python/semantic_kernel/kernel.py:444-447 |
三个 context 都继承 FilterContextBase(python/semantic_kernel/filters/filter_context_base.py:13-19),共享四个字段:function、kernel、arguments、is_streaming。各自的扩展字段:
| context | 独有字段 | 典型用法 |
|---|---|---|
PromptRenderContext | rendered_prompt、function_result | 渲染很贵时直接给结果跳过(prompt_render_context.py:14-16) |
FunctionInvocationContext | result | 记日志、缓存、改结果(function_invocation_context.py:27) |
AutoFunctionInvocationContext | chat_history、function_call_content、function_result、execution_settings、request_sequence_index、function_sequence_index、function_count、terminate | 见下 |
AutoFunctionInvocationContext 的字段定义在 python/semantic_kernel/filters/auto_function_invocation/auto_function_invocation_context.py:39-46。
位置感知的三个计数器
AutoFunctionInvocationContext 独有的三个 int,拼起来能回答「我是第几轮、第几个」:
| 字段 | 含义 | 赋值处 |
|---|---|---|
request_sequence_index | 第几轮(循环的 request_index) | kernel.py:439 |
function_sequence_index | 本轮里的第几个调用 | kernel.py:441-442(来自 function_call.index) |
function_count | 本轮一共几个调用 | kernel.py:438 |
有了这三个,filter 就能写出「只在第一轮的第一个调用上做某事」这类逻辑,而不需要自己在外面维护状态。
terminate:让 filter 决定终止循环
这是三类 filter 里唯一能改变外层循环走向的开关:
return invocation_context if invocation_context.terminate else None
kernel.py:463。回到 §3.4,循环在 chat_completion_client_base.py:169 检查 any(result.terminate ...),为真就立刻返回工具结果。
典型场景是 human-in-the-loop:filter 里发现这个工具需要人确认、或者已经拿到了足够的答案,就把 context.terminate = True,整个 auto-invoke 循环当场收工。
值得注意的是执行顺序:terminate 是在洋葱跑完之后才读的,所以 filter 既可以在 next 之前就置位(此时通常配合「不调 next」一起用,连函数都不执行),也可以在 next 之后根据结果决定。
3.9 可观测性锚点
整个自动调用循环被一个 span 罩着:
span = tracer.start_span(AUTO_FUNCTION_INVOCATION_SPAN_NAME)
chat_completion_client_base.py:417,常量值是 "AutoFunctionInvocationLoop"(python/semantic_kernel/const.py:11)。这个 span 在 :137(非流式)和 :256(流式)用 use_span(..., end_on_exit=True) 打开,覆盖全部轮次,不是每轮一个。
span 上挂一个属性:所有可用函数的全限定名,逗号拼接。
available_functions = settings.function_choice_behavior.get_config(kernel).available_functions or []
span.set_attribute(
AVAILABLE_FUNCTIONS,
",".join([f.fully_qualified_name for f in available_functions]),
)
:420-424。属性名是 "sk.available_functions"(python/semantic_kernel/utils/telemetry/model_diagnostics/gen_ai_attributes.py:45)——注意前缀是 sk. 而不是 gen_ai.,这是 SK 自己的扩展,不在 OpenTelemetry GenAI 语义约定里。
三层 span 的嵌套关系
AutoFunctionInvocationLoop (整个循环,1 个)
├─ chat <model> (每轮一次模型调用)
│ trace_chat_completion 装饰器
└─ <函数名> (每次工具执行)
function_tracer.start_as_current_span
| 层 | 谁开的 | 源码 |
|---|---|---|
| 循环 | _start_auto_function_invocation_activity | chat_completion_client_base.py:410-426 |
| 模型调用 | trace_chat_completion / trace_streaming_chat_completion | python/semantic_kernel/utils/telemetry/model_diagnostics/decorators.py:93-140 / :144 |
| 函数执行 | KernelFunction.invoke 里的 span | python/semantic_kernel/functions/kernel_function.py:264(流式 :336) |
最内层还会在敏感事件开关打开时记下参数和结果:
gen_ai.tool.call.arguments(kernel_function.py:269)gen_ai.tool.call.result(kernel_function.py:288,流式版:372)
开关是 function_tracer.are_sensitive_events_enabled()(python/semantic_kernel/utils/telemetry/model_diagnostics/function_tracer.py:31)。默认关——工具参数里可能有用户隐私,不该无条件进 trace。
同一处还有个直方图记录耗时:self.invocation_duration_histogram.record(duration, attributes)(kernel_function.py:296),标签是函数全限定名,可以直接拿来做「哪个工具最慢」的看板。
3.10 巧妙之处与边界
值得抄走的四个设计
① 错误全部转成对话文本。 §3.6 那五道关卡,没有一道会把异常抛给调用方。模型犯错 → 变成一句给模型看的话 → 模型下一轮自己改。这把「参数校验」从工程问题变成了对话问题,省掉了整套重试框架。
② 用一个整数同时表达开关和上限。 maximum_auto_invoke_attempts > 0 就是 auto_invoke(function_choice_behavior.py:65-68)。少一个字段,少一处「两个字段不一致」的 bug。
③ 白名单查两次。 一次在暴露给模型时(get_list_of_function_metadata(filters)),一次在真要执行时(kernel.py:342-349)。第二次专治模型幻觉出的函数名,是安全边界而非冗余。
④ 结果走 history 这条副作用通道,返回值只用来表达控制流。 invoke_function_call 返回 None 或 context,语义是「要不要停」,不是「结果是什么」。这让 asyncio.gather 的用法极其干净——不需要按顺序收集结果再拼装。
会在这里崩 / 需要小心的地方
| 情况 | 会发生什么 | 依据 |
|---|---|---|
number_of_responses > 1 + auto invoke | OpenAI connector 直接抛 ServiceInvalidExecutionSettingsError | open_ai_chat_completion_base.py:141-149 |
有 function_choice_behavior 但没传 kernel | 抛 ServiceInvalidExecutionSettingsError | chat_completion_client_base.py:116-118 |
filter 第二个参数没叫 next | TypeError,因为 partial 用的关键字传参 | kernel_filters_extension.py:116 |
| 流式模式跑满上限 | 没有「摘掉工具再问一次」的兜底,流直接结束 | chat_completion_client_base.py:257-318 无 else 分支 |
Required 配合多轮期待 | 默认上限是 1,不是 5 | function_choice_behavior.py:173 |
直接调 invoke_function_call 不传 function_behavior | 不做白名单校验,只打 debug 日志 | kernel.py:350-356 |
| 工具返回可变对象 | 已被 deepcopy 快照,但深拷贝失败的对象(如带锁的句柄)会抛出来 | kernel.py:450-452(inferred) |
刻意不做的事
- 没有内建的「工具重试」策略。 模型给错参数就是靠「把错误说给它听」让它自己重试,重试次数由
maximum_auto_invoke_attempts一并管着,没有独立的退避配置。 - 没有工具级的超时。 循环用
asyncio.gather一把等齐所有调用,任何一个卡住整轮就卡住;要超时得自己写function_invocationfilter 包asyncio.wait_for。 - 不做工具结果的裁剪。 工具返回多大就往
ChatHistory里塞多大,超上下文窗口是调用方的事。
3.11 代码地图
| 主题 | 文件路径(相对克隆根) | 关键符号 |
|---|---|---|
| 工具暴露策略与上限 | python/semantic_kernel/connectors/ai/function_choice_behavior.py | FunctionChoiceBehavior、DEFAULT_MAX_AUTO_INVOKE_ATTEMPTS、Auto、Required、NoneInvoke、configure、get_config、auto_invoke_kernel_functions |
| 三种选择类型枚举 | python/semantic_kernel/connectors/ai/function_choice_type.py | FunctionChoiceType |
| 自动调用循环(非流式 + 流式) | python/semantic_kernel/connectors/ai/chat_completion_client_base.py | get_chat_message_contents、get_streaming_chat_message_contents、_reset_function_choice_settings、_update_function_choice_settings_callback、_start_auto_function_invocation_activity、_yield_function_result_messages |
| tools 格式化与结果合并 | python/semantic_kernel/connectors/ai/function_calling_utils.py | kernel_function_metadata_to_function_call_format、update_settings_from_function_call_configuration、merge_function_results、merge_streaming_function_results、_combine_filter_dicts |
| 单次工具调用的容错流水线 | python/semantic_kernel/kernel.py | Kernel.invoke_function_call、_inner_auto_function_invoke_handler |
| 过滤器注册与洋葱组装 | python/semantic_kernel/filters/kernel_filters_extension.py | KernelFilterExtension.add_filter、filter、remove_filter、construct_call_stack、FILTER_MAPPING |
| 三类过滤器的枚举与上下文 | python/semantic_kernel/filters/ | FilterTypes、FilterContextBase、AutoFunctionInvocationContext、FunctionInvocationContext、PromptRenderContext |
| 函数执行处的 filter 挂载 | python/semantic_kernel/functions/kernel_function.py | KernelFunction.invoke、invoke_stream |
| 提示词渲染处的 filter 挂载 | python/semantic_kernel/functions/kernel_function_from_prompt.py | _render_prompt、_inner_render_prompt |
| 参数 schema 推断 | python/semantic_kernel/functions/kernel_parameter_metadata.py | KernelParameterMetadata.form_schema、infer_schema、include_in_function_choices |
| JSON Schema 生成 | python/semantic_kernel/schema/kernel_json_schema_builder.py | KernelJsonSchemaBuilder.build、build_model_schema、build_from_type_name、handle_complex_type、build_enum_schema、TYPE_MAPPING |
| 函数白名单过滤 | python/semantic_kernel/functions/kernel_function_extension.py | get_list_of_function_metadata_filters、get_full_list_of_function_metadata、get_function |
| 调用/结果内容与流式拼接 | python/semantic_kernel/contents/function_call_content.py、function_result_content.py | FunctionCallContent.__add__、parse_arguments、to_kernel_arguments、FunctionResultContent.from_function_call_content_and_result、to_chat_message_content、to_streaming_chat_message_content |
| OpenAI connector 的三个钩子 | python/semantic_kernel/connectors/ai/open_ai/services/open_ai_chat_completion_base.py | _verify_function_choice_settings、_update_function_choice_settings_callback、_reset_function_choice_settings |
| 遥测常量与装饰器 | python/semantic_kernel/const.py、python/semantic_kernel/utils/telemetry/model_diagnostics/ | AUTO_FUNCTION_INVOCATION_SPAN_NAME、AVAILABLE_FUNCTIONS、TOOL_CALL_ARGUMENTS、TOOL_CALL_RESULT、trace_chat_completion、trace_streaming_chat_completion、are_sensitive_events_enabled |
| 可运行的过滤器样例 | python/samples/concepts/filtering/ | auto_function_invoke_filters.py、function_invocation_filters.py、prompt_filters.py、retry_with_filters.py |
下一章: 第 04 章 · Agent 层:线程、指令与托管 agent 的统一外壳 —— 本章讲的循环是「一次对话内」的,Agent 层在它外面再包一层线程与指令管理。