数据截至 2026-08-18(上游 commit 5e1f1fb87d9a)

Agent 层:线程、指令与托管 agent 的统一外壳

30 秒导读: 前三章讲的是内核——Kernel 与可调用单元、提示词与内容模型、自动函数调用循环。这一章讲 SK 在内核之上包出的那层壳:Agent。它要解决的核心问题只有一个——把"一个会说话的角色"和"这段对话的状态存在哪"拆成两个正交的东西,于是本地 ChatHistory、OpenAI 服务端 thread、Bedrock session、Copilot Studio conversation 可以插进同一套调用代码。多 agent 编排与 actor 运行时不在本章,见第 05 章。

本章还藏着一套多 agent 机制,别读漏: §9 讲的 AgentChat / AgentChannel / BroadcastQueue 群聊路径,和第 05 章的编排体系是两套并存的实现。它被放在本章而不是第 05 章,是因为它和 Agent / AgentChannel 抽象长在一起;两者的定位对比见 §9.5。

路径约定: 本章所有 file:line 引用相对 python/semantic_kernel/。例如 agents/agent.py:111 的完整路径是 python/semantic_kernel/agents/agent.py:111。文末「代码地图」给出完整路径。

1. 这是什么(零基础也能懂)

一句话定义: SK 的 Agent 层是一层薄外壳——把"系统指令 + 一个 kernel + 一套插件"打包成一个有名字的角色,再给它配一个可插拔的「对话状态容器」(thread)。

它替你解决的麻烦,是"状态存在哪"这件事。 只用第 01–03 章的内核,你得自己维护 ChatHistory:每轮把用户消息 append 进去、把模型回复 append 回去、把工具消息补进去、决定要不要裁剪。

一旦你想换成 OpenAI Assistants API 或 Azure AI Foundry,这套代码要全部重写——因为那边的对话历史根本不在你进程里,你手上只有一个服务端 thread ID。

Agent 层的做法:把"往对话里加一条消息"抽象成一个方法,本地实现就是 list.append,远端实现就是一次 HTTP 调用。调用方两边写法完全一样。

用起来什么样 —— 一个最小的真实用法(参考 samples/concepts/agents/chat_completion_agent/):

# 示意,非源码;演示 agent 层最小闭环
agent = ChatCompletionAgent(
    service=AzureChatCompletion(),
    name="Host",
    instructions="回答关于菜单的问题。",
    plugins=[MenuPlugin()],          # 插件直接挂进 agent 自带的 kernel
)

thread = None                        # 第一轮不传线程,agent 自己造一个
for user_text in ["有什么特色菜?", "刚才那道多少钱?"]:
    response = await agent.get_response(messages=user_text, thread=thread)
    print(response.content)
    thread = response.thread         # 重点看这里:线程从响应里拿回来,喂给下一轮

一句话直觉: 把 Agent 当作一个无状态的函数(指令 + 工具 + 模型),把 AgentThread 当作传给它的那本账本。同一个 agent 可以同时服务一百本账本;同一本账本也可以被好几个 agent 轮流写。

本节到此不涉及底层。下面开始拆。

2. 顶层全景(三件套怎么转)

怎么读这张图: 从左到右是一次调用的方向;虚线是"状态回流"——响应把线程带回给调用方,所以调用方永远不需要自己拼历史。

                            ┌──────────────────────────────┐
   messages(str/CMC/列表)──▶│           Agent              │
                            │  指令 + kernel + 插件 + 名字  │
                            │  get_response / invoke /     │
                            │  invoke_stream               │
                            └───────┬──────────────┬───────┘
                                    │              │
                     ①先把入参消息   │              │ ②调模型(第 03 章的
                     通知给线程      ▼              ▼   自动函数调用循环)
                            ┌──────────────┐   ┌─────────────────┐
                            │ AgentThread  │   │  Kernel + 服务   │
                            │ 对话状态在哪  │   └─────────────────┘
                            │ create/delete│              │
                            │ on_new_message│◀────────────┘
                            └──────┬───────┘   ③把新产生的消息补回线程
                                   │
                                   ▼
                       ┌───────────────────────┐
                       │  AgentResponseItem    │
                       │  message + thread     │──┈┈▶ 调用方拿 .thread 进下一轮
                       └───────────────────────┘

三件套各自的职责:

部件	干什么	定义位置
`Agent`	角色本身:名字、描述、指令、kernel、插件、三个调用方法	`agents/agent.py:246`
`AgentThread`	对话状态的抽象容器,只暴露 `create` / `delete` / `on_new_message`	`agents/agent.py:111`
`AgentResponseItem`	响应信封:把消息和它所属的线程绑在一起返回	`agents/agent.py:191`

主线走一遍(高层): 调用方给 messages 和(可选)thread → agent 确保线程存在并把入参消息写进去 → agent 从线程读回完整历史、渲染指令、调模型 → 模型循环中新产生的消息按序补回线程 → 最终回复包成 AgentResponseItem 返回,里面带着线程。

3. 三件套逐个拆

3.1 `AgentThread`:三个抽象方法撑起所有后端

它要解决的小问题: 「一段对话」在不同后端是完全不同的东西——本地是一个 list,OpenAI 是一个 thread_xxx ID,Bedrock 是一个 session,Responses API 甚至是一串 previous_response_id。要让上层代码统一,就得找出它们共同的最小动作集。

SK 的答案是三个抽象方法(agents/agent.py:167-183):_create / _delete / _on_new_message。

外面包了三个公有方法,负责状态机纪律:

create()(agents/agent.py:126)—— 幂等。已有 ID 直接返回;已删除则抛 RuntimeError。
delete()(agents/agent.py:140)—— 幂等。删过就 no-op;没 ID 也只是打个删除标记。
on_new_message()(agents/agent.py:156)—— 懒创建:线程还没 ID 就先 create(),再调 _on_new_message。

线程的生命周期状态:

   刚 new 出来              create()                 delete()
   _id = None  ──────────────────────▶  _id = "xxx"  ─────────▶  _is_deleted = True
       │                                    │                          │
       │ on_new_message()                   │ on_new_message()         │  再 create()
       │  → 自动补 create()  ───────────────┘  → _on_new_message()     │  → RuntimeError
       │                                                               │  访问 .id
       └───────────────────────────────────────────────────────────────┘  → RuntimeError

注意 id 属性(agents/agent.py:119-124)在已删除时直接抛 RuntimeError,而不是返回 None——这是刻意的:静默返回 None 会让调用方在一个死线程上继续写消息。

入口收口在一个地方: 所有 agent 的三个调用方法开头,都调基类的 _ensure_thread_exists_with_messages(agents/agent.py:492-531)。这个方法一次做四件事:

归一化入参 —— None → [],单个 → 单元素列表,裸字符串 → ChatMessageContent(role=USER)(agents/agent.py:504-509)。
没传线程就用调用方给的 construct_thread 工厂新建一个并 create()(agents/agent.py:511-513)。
类型校验 —— 传进来的线程类型不对,抛 AgentExecutionException(agents/agent.py:515-518)。这是"线程只能配对应的 agent"这条规则的落点。
给每条用户消息的 metadata 打上 agent_id,再逐条 on_new_message 推给线程(agents/agent.py:520-529)。

第 4 步的注释说明了打 agent_id 的动机:agent 可能已经被删了,但历史消息还得能追溯是谁的(agents/agent.py:520-521)。

3.2 `AgentResponseItem`:为什么响应要背着线程

问题: 第一轮调用时线程可能是 agent 自己 new 出来的,调用方手上什么都没有。

解法: 返回值不是裸消息,而是 message + thread 的二元组(agents/agent.py:191-200)。调用方从 response.thread 拿回线程,下一轮传回去——线程的所有权自然地在调用链里流转,不需要全局注册表。

这个信封还做了透明代理:__getattr__(agents/agent.py:231)把找不到的属性全转发给 message,同时显式暴露 content / items / metadata / name / role 几个高频属性(agents/agent.py:202-225)。所以 response.content、response.role 都能直接用,像在操作消息本身。

3.3 `Agent`:三个调用方法的分工

Agent 是 KernelBaseModel + ABC(agents/agent.py:246),字段很少:arguments / description / id / instructions / kernel / name / prompt_template(agents/agent.py:266-273)。

name 默认是 agent_<随机 ASCII>,并且受 AGENT_NAME_REGEX(utils/validation.py:3,即 ^[0-9A-Za-z_-]+$)约束——因为这个名字后面会当函数名用(见 §4)。

plugins 不是字段,而是一个 before-validator:_configure_plugins(agents/agent.py:282-293)把 plugins=[...] 吃掉,逐个 kernel.add_plugin 进(必要时新建的)kernel。所以构造函数里写 plugins=[MenuPlugin()] 和自己建 kernel 再 add_plugin 是同一件事。

三个抽象调用方法的取舍写在各自的 docstring 里(agents/agent.py:323-413):

方法	返回	中间步骤怎么给
`get_response`	单个 `AgentResponseItem`,阻塞到最终结果	拿不到
`invoke`	异步迭代完整消息(为多模态留的口子)	`on_intermediate_message` 回调
`invoke_stream`	异步迭代 `StreamingChatMessageContent` 分片	`on_intermediate_message` 回调(给的是成形消息)

get_response 的 docstring 明确解释了"为什么没有'只流式最终结果'这个选项":最终结果何时可用是未知的,阻塞调用方到那时在流式场景下不可接受(agents/agent.py:335-341)。

4. 每个 agent 出生自带一把工具柄

这是本章最值得记住的设计。 Agent.model_post_init(agents/agent.py:295)在每个 agent 实例构造完成时,当场闭包出一个 @kernel_function 并挂到实例上:

# agents/agent.py:298-319(节选)
@kernel_function(name=self.name, description=self.description or self.instructions)
async def _as_kernel_function(
    messages: Annotated[str | list[str], "The user messages for the agent."],
    instructions_override: Annotated[str | None, "Override agent instructions."] = None,
) -> Annotated[Any, "Agent response."]:
    ...
    response_item = await self.get_response(messages=messages, instructions_override=...)
    return response_item.content

setattr(self, "_as_kernel_function", _as_kernel_function)

三个细节值得注意:

函数名 = agent 名,描述 = 描述或指令。 所以模型在工具列表里看到的就是这个 agent 的身份;name 那条正则的存在就是为了这里。
签名被刻意压到最小——只有 messages 和 instructions_override,注释直呼其为 "A Minimal universal function for all agents"(agents/agent.py:303)。所有 agent 类型共用这一个形状。
用 setattr 挂到实例而非定义成方法,是为了绕开 Pydantic 对私有方法的校验(agents/agent.py:317-318)。

它凭什么能被 kernel 认出来? KernelPlugin.from_object(functions/kernel_plugin.py:215)用 inspect.getmembers(..., inspect.iscoroutinefunction) 扫描对象成员,再筛 hasattr(candidate, "__kernel_function__")(functions/kernel_plugin.py:239-247)。实例属性上的这个协程刚好被扫到。

于是 plugins=[some_agent] 就直接成立了——真实样例:

# samples/concepts/agents/chat_completion_agent/chat_completion_agent_as_kernel_function.py:64-75(节选)
triage_agent = ChatCompletionAgent(
    ...,
    plugins=[billing_agent, refund_agent],   # 两个 agent 直接当插件
)

一句诚实的更正: 这把工具柄是「agent 即工具」和 as_mcp_server(§10)的基础,但第 05 章的 handoff 编排并没有复用它——HandoffAgentActor._add_handoff_functions(agents/orchestration/handoffs.py:190-221)自己造了一批名为 transfer_to_<名字> 的函数(handoffs.py:194),配一个自动函数调用过滤器来终止当前 agent。全仓 grep _as_kernel_function 只有 agents/agent.py:299(定义)和 :319(setattr)两处命中。所以准确说法是:同一个"agent 变函数"的思路有两处独立实现,handoff 走的是自己那套。

5. 指令是模板,不是字符串

它要解决的小问题: 指令里常常要塞变量——"你在为 {{$customer}} 服务"。如果指令只是死字符串,你就得每轮自己拼。

format_instructions(agents/agent.py:443-459)把指令当提示词模板处理:没有 prompt_template 时,懒构造一个 KernelPromptTemplate(用 instructions 当模板体)缓存到 self.prompt_template,然后 render(kernel, arguments)。

渲染发生在每次调用,不是构造时——所以同一个 agent 换一组参数就换一套人格,不用重建。模板语法与渲染细节见第 02 章。

参数从哪来:参数覆盖顺序。 _merge_arguments(agents/agent.py:461-486)的规则一句话说完:本次调用传入的参数覆盖 agent 构造时的参数。

情况	结果
agent 无参数、调用无参数	空 `KernelArguments()`
agent 无参数、调用有参数	直接用调用的
agent 有参数、调用无参数	直接用 agent 的
两边都有	合并,`override_args` 赢(`agents/agent.py:479-486`)

一个要留意的副作用: 合并 execution settings 那两行是

# agents/agent.py:479-481
merged_execution_settings = self.arguments.execution_settings or {}
if override_args.execution_settings:
    merged_execution_settings.update(override_args.execution_settings)

merged_execution_settings 拿到的是 self.arguments.execution_settings 同一个 dict 对象,update 是就地写。也就是说,某一次调用传进来的 execution settings 会留在 agent 上影响后续调用(普通参数那边用 dict(...) 拷了副本,没这个问题,见 agents/agent.py:483-484)。用同一个 agent 跨请求复用且逐请求改采样参数时,这点值得知道。

6. 参考实现:`ChatCompletionAgent` 走一遍

这是 SK 自己的"纯本地" agent(agents/chat_completion/chat_completion_agent.py:117),也是理解整层的最佳标本——因为它是唯一一个对话循环完全跑在你进程里的实现。

6.1 它的线程:`ChatHistoryAgentThread`

定义在同一个文件的最上面(agents/chat_completion/chat_completion_agent.py:51)。三个抽象方法实现得几乎"不像实现":

_create 直接返回构造时就生成好的 thread_<uuid>(:72-74)——本地线程不需要跟谁申请 ID。
_delete 只是 self._chat_history.clear()(:77-79)。
_on_new_message 把消息 append 进本地 ChatHistory(:82-92)。

_on_new_message 里的去重守卫值得单独看(:87-92):如果消息 metadata 里带的 thread_id 就等于本线程 ID,跳过不加。含义是"这条消息本来就是从这个线程里读出来的",防止回写造成重复。托管型线程(Assistant / AzureAI)用的是同一条判据,只是加的动作变成了往服务端 POST(见 §8)。

reduce()(:107-113)是历史裁剪的挂钩:只有当底层 ChatHistory 实际是一个 ChatHistoryReducer 时才生效,否则返回 None。裁剪策略本身属于内容模型层(第 02 章)。

6.2 一次非流式调用的六步

怎么读这张图: 自上而下是时间顺序;右侧标注每步碰的是"线程"还是"本次请求的临时历史"——这两份历史不是一份东西,这是理解 _drain_mutated_messages 的关键。

  ① _ensure_thread_exists_with_messages          → 写【线程】
     入参消息归一化 + 打 agent_id + 推给线程
                    │
                    ▼
  ② 从线程读回全部消息 → chat_history            → 读【线程】
                    │
                    ▼
  ③ _prepare_agent_chat_history                  → 造【临时历史】
     [system: 渲染后的 instructions] + [线程历史]
     记下 start_idx = 此刻的长度
                    │
                    ▼
  ④ 调 chat completion 服务                      → 改【临时历史】
     第 03 章的自动函数调用循环会**就地**往这份
     临时历史里追加 assistant(tool_call) 和 tool 消息
                    │
                    ▼
  ⑤ _drain_mutated_messages(从 start_idx 到末尾)  → 补回【线程】
                    │
                    ▼
  ⑥ 最终回复 role != TOOL 才通知线程,然后 yield   → 写【线程】

对应代码:_inner_invoke(agents/chat_completion/chat_completion_agent.py:500-568),被 get_response(:276)和 invoke(:328)共用。

6.3 `_prepare_agent_chat_history`:指令不进线程

# agents/chat_completion/chat_completion_agent.py:570-581(节选)
formatted_instructions = await self.format_instructions(kernel, arguments)
messages = []
if formatted_instructions:
    messages.append(ChatMessageContent(role=AuthorRole.SYSTEM, content=formatted_instructions, name=self.name))
if history.messages:
    messages.extend(history.messages)
return ChatHistory(messages=messages)

它返回一份新的 ChatHistory,系统消息只活在这一次请求里。含义有两条:

线程里永远没有系统消息 —— 所以同一个线程可以被换了指令的 agent 接着用,不会累积一堆过期人格。
指令每轮重新渲染 —— 参数一变,下一轮的系统消息就跟着变。

6.4 `_drain_mutated_messages`:补齐工具消息又不重复通知

这是本章工程含量最高的一小段。 背景:第 03 章的自动函数调用循环在服务层内部跑完多轮工具调用,它把 assistant 的 tool_call 消息和 tool 结果消息就地追加进传进去的那份历史,但不会把它们放进返回值 responses 里。

如果不管,线程就只记得"用户问了什么、最终答了什么",中间的工具调用全丢——下一轮模型看不到自己刚查过什么。

做法是指针 + 排空:

# agents/chat_completion/chat_completion_agent.py:599-612
async def _drain_mutated_messages(self, history, start, thread) -> list[ChatMessageContent]:
    drained = []
    for i in range(start, len(history)):
        msg = history[i]
        msg.name = self.name
        await thread.on_new_message(msg)
        drained.append(msg)
    return drained

调用前 start_idx = len(agent_chat_history)(:531),调用后从 start_idx 排到末尾——按原顺序推进线程,顺带把 name 盖成 agent 名。

然后是防重复那一半(:561-568):遍历 responses 时,role == TOOL 的跳过通知。注释解释了为什么(:564-566):当自动函数调用循环被过滤器提前终止时,tool 消息本身就是返回的响应——而它刚刚已经被排空推过一次了,再推就是重复。

流式版本(invoke_stream:376-494)是同一套逻辑的增量版:start_idx 在每个 chunk 批次后重置为当前长度(:474-480),循环结束时再把攒下来的文本拼成一条完整消息推给线程,同样带 role != TOOL 守卫(:486-494)。

流式还多一层过滤:只有 role == ASSISTANT、且 items 里不含 FunctionCallContent / FunctionResultContent 的分片才 yield 给调用方(:463-470)——工具调用的分片不该被当成正文喷给用户。

7. 声明式 agent:一份 YAML 造一个 agent

它要解决的小问题: agent 的定义(指令、模型、工具清单)是配置,不该硬编码在 Python 里。

7.1 Spec 数据模型

agents/agent.py:47-102 定义了一组纯 Pydantic 模型:

模型	装什么	行
`AgentSpec`	顶层:`type` / `name` / `instructions` / `model` / `tools` / `template` / `inputs` / `outputs`	`agents/agent.py:89`
`ModelSpec`	`id` / `api` / `options` / `connection`	`agents/agent.py:70`
`ModelConnection`	`type` / `service_id` / `extras`	`agents/agent.py:62`
`ToolSpec`	`id` / `type` / `description` / `options`	`agents/agent.py:79`
`InputSpec` / `OutputSpec`	输入默认值与必填、输出描述	`agents/agent.py:47` / `:55`

一份真实的 spec(samples/concepts/resources/declarative_spec/azure_ai_agent_spec.yaml):

type: foundry_agent
name: FunctionCallingAgent
instructions: Use the provided functions to answer questions about the menu.
model:
  id: ${AzureAI:ChatModelId}
  connection:
    endpoint: ${AzureAI:Endpoint}
  options:
    temperature: 0.4
tools:
  - id: MenuPlugin.get_specials
    type: function

7.2 类型注册表与 `AgentRegistry`

type 字段怎么变成类?靠一个全局字典 AGENT_TYPE_REGISTRY 和装饰器 register_agent_type(agents/agent.py:642-658)。

装饰器只在模块被 import 时才执行,所以有 _preload_builtin_agents()(agents/agent.py:676-699):双检锁包着,逐个 importlib.import_module 那六个内建模块(agents/agent.py:666-673),任何一个失败就汇总报错。

当前可用的声明式类型(全仓 grep @register_agent_type 的结果,行号指装饰器所在行):

YAML `type`	类	定义位置
`chat_completion_agent`	`ChatCompletionAgent`	`agents/chat_completion/chat_completion_agent.py:116`
`foundry_agent`	`AzureAIAgent`	`agents/azure_ai/azure_ai_agent.py:352`
`openai_assistant`	`OpenAIAssistantAgent`	`agents/open_ai/openai_assistant_agent.py:228`
`azure_assistant`	`AzureAssistantAgent`	`agents/open_ai/azure_assistant_agent.py:41`
`openai_responses`	`OpenAIResponsesAgent`	`agents/open_ai/openai_responses_agent.py:256`
`azure_responses`	`AzureResponsesAgent`	`agents/open_ai/azure_responses_agent.py:45`

Bedrock、Copilot Studio、AutoGen 三家没有这个装饰器,因此不能从 YAML 声明式创建。

AgentRegistry(agents/agent.py:702)提供三个入口:create_from_yaml(:719)、create_from_dict(:779)、create_from_file(:834,只是读文件后转给 create_from_yaml)。

create_from_yaml 里一个容易看漏的细节:YAML 被解析了两次。

第一次 yaml.safe_load ──▶ 取出 type ──▶ 查注册表 ──▶ 拿到 agent_cls
                                                        │
                          agent_cls.resolve_placeholders(yaml_str, settings, extras)
                                                        │
第二次 yaml.safe_load ──────────────────────────────────┘──▶ agent_cls.from_dict(...)

依据 agents/agent.py:751-776。这么排的原因是占位符替换是每种 agent 自己的事——AzureAIAgent.resolve_placeholders(agents/azure_ai/azure_ai_agent.py:580-605)用正则把 ${AzureAI:Endpoint} 之类换成 AzureAIAgentSettings 里的字段。所以 type 必须在替换之前就能读出来。

7.3 `DeclarativeSpecMixin`:两个关键钩子

Mixin(agents/agent.py:889)给出 from_yaml / from_dict 的默认实现,把真正的构造留给子类的抽象方法 _from_dict(agents/agent.py:939-950)。

_normalize_spec_fields(agents/agent.py:966-1039) 是 spec → 构造参数的翻译层,四件事:

没 kernel 就新建一个;有 plugins 就逐个 add_plugin(:981-987)。
有 tools 就调 _validate_tools 校验(:990-991)。
model.options 变成 execution settings,不是普通参数(:1005-1012)。注释专门解释了这个区分:temperature、response_format 这类是执行设置,应该塞进 KernelArguments(settings=...);inputs 里的 default 才作为普通参数放进 arguments(:1014-1017)。
处理 template / prompt_template:构造 PromptTemplateConfig,若 YAML 同时给了 instructions,instructions 覆盖模板体,并把 fields["instructions"] 回写成最终用的模板(:1027-1037)。

_validate_tools(agents/agent.py:1042-1066)只校验,不创建。 它只管 type == "function" 的条目,要求 id 形如 PluginName.FunctionName,然后在 kernel 里查插件、查函数,任一步查不到就抛 AgentInitializationException。

这是最常见的踩坑点: YAML 里写了 MenuPlugin.get_specials,不等于 SK 会去把 MenuPlugin 变出来——你必须在调 create_from_yaml 时通过 plugins= 或预置的 kernel 把它交进去,否则直接报 "Plugin 'MenuPlugin' not found in kernel."。

8. 托管型 agent 家族:同一个契约,六种"线程"

这才是三件套抽象的回报。 每个托管后端的"对话状态"物理上完全不同,但都被压进 _create / _delete / _on_new_message 三个方法。

Agent	配套 Thread	线程 ID 实际是什么	状态存在哪	定义位置
`ChatCompletionAgent`	`ChatHistoryAgentThread`	本地生成的 `thread_<uuid>`	你的进程内存	`agents/chat_completion/chat_completion_agent.py:51`
`OpenAIAssistantAgent`	`AssistantAgentThread`	OpenAI 服务端 thread ID	OpenAI 服务端	`agents/open_ai/openai_assistant_agent.py:136`
`OpenAIResponsesAgent`	`ResponsesAgentThread`	上一条 response 的 ID	服务端或本地(见下)	`agents/open_ai/openai_responses_agent.py:144`
`AzureAIAgent`	`AzureAIAgentThread`	Foundry 项目的 thread ID	Azure AI 服务端	`agents/azure_ai/azure_ai_agent.py:260`
`BedrockAgent`	`BedrockAgentThread`	Bedrock session ID	AWS 服务端	`agents/bedrock/bedrock_agent.py:55`
`CopilotStudioAgent`	`CopilotStudioAgentThread`	Copilot conversation ID	Copilot Studio 服务端	`agents/copilot_studio/copilot_studio_agent.py:197`
`AutoGenConversableAgent`	`AutoGenConversableAgentThread`	本地 `thread_<uuid>`	你的进程内存	`agents/autogen/autogen_conversable_agent.py:42`

对外清单由 agents/__init__.pyi 统一给出(agents/__init__.pyi:15-41 的导入 + :42-89 的 __all__),里面还包含第 05 章的五种编排类。

四种典型的实现姿态,值得对照着看:

(a) 真·远端线程 —— Assistant / AzureAI。 _create 是一次真实的 HTTP 创建(agents/open_ai/openai_assistant_agent.py:171-175 调 client.beta.threads.create;agents/azure_ai/azure_ai_agent.py:296-300 调 client.agents.threads.create),失败统一包成 AgentThreadOperationException。_on_new_message 则是一次真实的消息上传(agents/open_ai/openai_assistant_agent.py:207)。两者的去重守卫和本地版一模一样(:201-205 / agents/azure_ai/azure_ai_agent.py:325-329)——同一条判据,不同的落地动作。

(b) ID 是滚动的 —— Responses。 ResponsesAgentThread 干脆把 id 属性重写成 response_id(agents/open_ai/openai_responses_agent.py:188-192),因为 Responses API 靠 previous_response_id 串对话。_create 只能返回空串,注释直说"ID 在发出第一条消息前不可用"(:203-204)。它还有个双模式:store_enabled 为真且已有 response_id 时,get_messages 从服务端拉;否则读本地 ChatHistory(:231-241)。

(c) 只读线程 —— Bedrock / Copilot Studio。 这两家的 _on_new_message 直接 raise NotImplementedError(agents/bedrock/bedrock_agent.py:104-109、agents/copilot_studio/copilot_studio_agent.py:251-255),理由写在异常信息里:消息与响应由服务自己管,SK 无从插手。这是抽象诚实的一面——做不到就明确报错,而不是假装成功。

BedrockAgentThread._delete 还有个语义脚注:它只 end_session,不删除 session(agents/bedrock/bedrock_agent.py:90-101)。

(d) 一个额外的坑:CopilotStudioAgent.channel_type = None(agents/copilot_studio/copilot_studio_agent.py:263)。基类的 get_channel_keys / create_channel 在 channel_type 为空时抛 NotImplementedError(agents/agent.py:425-437),所以这个 agent 不能加入 §9 的 AgentGroupChat。

9. 旧版群聊路径:`AgentChat` + `AgentChannel` + `BroadcastQueue`(与第 05 章的编排体系并存)

SK 里有两套让多个 agent 说话的机制。本章讲旧的这套,是因为它和 Agent/AgentChannel 抽象长在一起;新的那套(actor 运行时 + 五种编排模式)见第 05 章。两者的定位对比见 §9.5。

9.1 为什么需要 channel

问题: 群聊里 ChatCompletionAgent 和 OpenAIAssistantAgent 坐在一起。前者的历史在本地 list 里,后者的历史在 OpenAI 服务端。你没法给它们一份共享的 ChatHistory。

解法: 每种 agent 类型配一个 AgentChannel(agents/channels/agent_channel.py:15)——它是那种 agent 类型的"通信协议",五个抽象方法:receive / invoke / invoke_stream / get_history / reset。

群聊维护一份权威历史,同时给每种 channel 维护一份镜像。新消息产生时,群聊把它 receive 给所有其它 channel,各自用自己的方式落地。

channel 是按类型共享的,不是按 agent。 AgentChat._get_agent_hash(agents/group_chat/agent_chat.py:91-98)拿 agent.get_channel_keys()(基类实现就是 channel_type.__name__,agents/agent.py:419-427)算哈希。所以三个 ChatCompletionAgent 共用同一个 ChatHistoryChannel——历史只镜像一份。

9.2 `BroadcastQueue`:广播是异步的,读取时才等

它要解决的小问题: 一个 agent 说完话,要把消息同步给其它所有 channel。同步等待会让每轮都被最慢的后端拖住。

思路:每个 channel 一条队列 + 一个后台任务;入队不等待,只有真要读某个 channel 时才等它追平。

   agent 说完一句
        │
        ▼
   enqueue(channel_refs, messages)          ← 不 await 投递,立刻返回
        │
        ├──▶ [队列 A] ──(后台 task)──▶ channelA.receive(...)
        ├──▶ [队列 B] ──(后台 task)──▶ channelB.receive(...)
        └──▶ [队列 C] ──(后台 task)──▶ channelC.receive(...)

   要调用 agent B 之前:
        ensure_synchronized(B) ──▶ 轮询直到队列 B 空,才放行

enqueue(agents/group_chat/broadcast_queue.py:57-74):把消息 append 进对应队列,若没有活跃的接收任务就 asyncio.create_task 起一个。注释写明"能在不同步 channel 的情况下排队消息"(agents/group_chat/agent_chat.py:122-123)。
ensure_synchronized(:76-104):循环检查队列是否为空,不空就 await asyncio.sleep(block_duration)(默认 0.1 秒,:55)。后台任务里出的异常被存在 receive_failure 里,在这里才重新抛出(:91-96)——异步投递不丢错。
receive(:106-129):排空循环,注意它是先 peek 再投递,投递成功后才 popleft(:118-126),避免失败时消息丢失。

9.3 `AgentChat` / `AgentGroupChat`:单飞行锁 + 选择/终止策略

AgentChat(agents/group_chat/agent_chat.py:25)是基类,拿一把 threading.Lock 实现"同一时刻只能有一个 agent 活动":set_activity_or_throw(:41-46)/ clear_activity_signal(:48-51),每个公开方法都用 try/finally 包住。

主循环 invoke_agent(:141-162)四步:取或建 channel → channel.invoke(agent) 逐条产出 (是否可见, 消息) → 追加进权威历史 → 把这批消息广播给其它 channel。

_get_or_create_channel(:129-139)有个补课动作:新建的 channel 会先 receive 一遍已有的全部历史,新加入的 agent 能看到之前发生了什么。

AgentGroupChat(agents/group_chat/agent_group_chat.py:30)在上面加多轮:for _ in range(termination_strategy.maximum_iterations) 里,用 selection_strategy.next(...) 选人、跑一轮、用 termination_strategy.should_terminate(...) 判断是否结束(:149-163)。策略实现在 agents/strategies/selection/ 与 agents/strategies/termination/(含 SequentialSelectionStrategy、KernelFunctionSelectionStrategy 等)。

reduce_history(:215-228)做历史裁剪:裁完之后先 reset() 掉所有 channel,再把裁剪后的历史重新广播——因为各 channel 的镜像也必须跟着缩。

9.4 `ChatHistoryChannel`:去重与"可见性"

ChatHistoryChannel(agents/channels/chat_history_channel.py:33)同时继承 AgentChannel 和 ChatHistory——它自己就是那份镜像历史。

invoke(:46-96)要处理一个尴尬:agent 在跑的过程中会就地往这份历史里追加工具消息(§6.4 那个机制),同时又会 yield 最终响应。两个来源可能重叠。

做法是指针 + 集合去重 + 队列(:61-88):记住调用前的 message_count,每轮把新增部分收进 mutated_history 集合和 message_queue;若 response.message 不在集合里才额外入队。

_is_message_visible(:124-129)决定哪些消息给用户看:含 FunctionCallContent / FunctionResultContent 的消息默认不可见,除非它是队列里最后一条——即工具调用是过程,除非它就是结果。

receive(:131-154)做内容过滤:深拷贝每条消息,只保留 ALLOWED_CONTENT_TYPES(:38-44:Image / FunctionCall / FunctionResult / StreamingText / Text)里的 item,过滤后为空的整条丢弃。docstring 点明动机:不要引入只含文件引用的消息。

9.5 与第 05 章新体系的关系

代码里 AgentChat 和 AgentGroupChat 都标着 @experimental(agents/group_chat/agent_chat.py:24、agent_group_chat.py:29),没有 @deprecated 标记。两套机制在 agents/__init__.pyi 里并列导出。

不过两者的定位差别很清楚:

维度	`AgentGroupChat`(本章)	编排体系(第 05 章)
状态载体	`AgentChannel` + 镜像历史 + 广播队列	`AgentThread`(`agents/orchestration/agent_actor_base.py:125`)
并发模型	一把进程内锁,同一时刻只一个 agent	actor 运行时,消息驱动
协作形态	只有"轮流发言 + 选择/终止策略"	concurrent / sequential / handoff / group chat / magentic 五种
换 agent 类型的成本	得有对应的 `channel_type`,否则加不进去	只要能 `invoke`,不需要 channel

一句话取舍:AgentGroupChat 是"多种异构 agent 共享一段历史"的方案,代价是每种 agent 都得有 channel 实现;编排体系是"agent 之间传消息"的方案,不再要求共享历史。 第 05 章的编排类完全不碰 AgentChannel(grep agents/orchestration/*.py 只命中 AgentThread)。

10. 把 agent 变成一台 MCP server

Agent.as_mcp_server(agents/agent.py:559-588)一行转发:

# agents/agent.py:579-588(节选)
from semantic_kernel.connectors.mcp import create_mcp_server_from_functions
return create_mcp_server_from_functions(
    functions=self,                       # 把 agent 自己当作"可转插件的对象"传进去
    prompts=prompts,
    server_name=server_name or self.name, # 默认拿 agent 名当 server 名
    ...
)

它能成立,靠的就是 §4 那把工具柄。 create_mcp_server_from_functions(connectors/mcp.py:972)对非 KernelFunction 的入参走 kernel.add_plugin(func, plugin_name),于是 _as_kernel_function 被扫出来注册成唯一一个 tool。docstring 说得很直白:创建一个 MCP Server,里面只有一个 Tool,就是这个 agent 本身(agents/agent.py:570-571)。

真实样例在 samples/demos/mcp_server/agent_as_server.py 与 samples/concepts/mcp/servers/menu_agent_server.py。

这条路和 §4 是同一件事的两面: 对内,agent 是别的 agent 的一个函数;对外,agent 是一台标准 MCP server 上的一个 tool。同一个 @kernel_function 包装,换个出口。

11. 巧妙之处(可带走的技术)

"三个方法定义一段对话"。 _create / _delete / _on_new_message(agents/agent.py:167-183)是我见过最省的对话状态抽象——本地 list、OpenAI thread、Bedrock session、滚动的 previous_response_id 全都塞得进去。抽象小到极致,兼容面才大。
响应背着线程返回(agents/agent.py:191-200)。省掉了全局会话表和"线程 ID 传来传去"的样板;线程的所有权跟着调用链走。
懒创建 + 幂等的状态机纪律。 create 幂等、delete 幂等、on_new_message 自动补 create、已删除访问 id 直接抛错(agents/agent.py:119-165)。调用方几乎不可能把线程用到一个坏状态里。
指针 + 排空,解决"框架内部偷偷加消息"。 _drain_mutated_messages(agents/chat_completion/chat_completion_agent.py:599-612)配上 role != TOOL 守卫(:561-567),把自动函数调用循环的中间产物按序补进线程,又不重复通知。凡是"底层会就地改我传进去的容器"的场景,这个模式都能用。
广播异步、读取时才同步。 BroadcastQueue.enqueue 不等待,ensure_synchronized 才阻塞(agents/group_chat/broadcast_queue.py:57/:76),并且把后台任务的异常存起来在同步点重抛(:91-96)。快路径不被慢后端拖住,错误又不丢。
消息可见性是一个函数,不是一个字段。 _is_message_visible(agents/channels/chat_history_channel.py:124-129)用"是不是队列最后一条"来判断工具消息该不该露出——过程隐藏、结果显示,一行判据。
做不到就明确报错。 Bedrock / Copilot Studio 的 _on_new_message 直接 NotImplementedError 并写清原因(agents/bedrock/bedrock_agent.py:106-109)。抽象不合身时,诚实报错好过静默降级。

12. 边界与局限

线程与 agent 类型强绑定。 _ensure_thread_exists_with_messages 会做 isinstance 检查并抛 AgentExecutionException(agents/agent.py:515-518)。你不能把一个 ChatHistoryAgentThread 递给 OpenAIAssistantAgent——跨 agent 类型延续对话需要自己搬运消息。
两家托管 agent 的线程是只读的。 Bedrock 和 Copilot Studio 无法接受外部写入消息(§8-c),因此也无法做"预置几轮历史再开始"这类操作。
channel_type 为空的 agent 进不了 AgentGroupChat。 CopilotStudioAgent 明确设为 None(agents/copilot_studio/copilot_studio_agent.py:263),基类会抛 NotImplementedError(agents/agent.py:425-426)。
声明式只覆盖六种类型。 Bedrock / Copilot Studio / AutoGen 没有 @register_agent_type,YAML 造不出来(§7.2)。
_validate_tools 只验不建。 YAML 里的工具必须预先存在于 kernel(agents/agent.py:1042-1066),spec 本身不具备"按 id 自动加载插件"的能力。
_merge_arguments 会就地改 agent 的 execution settings。 见 §5 末尾(agents/agent.py:479-481)。长生命周期的 agent 跨请求复用时需留意。
AgentGroupChat 是单进程、单飞行的。 一把 threading.Lock(agents/group_chat/agent_chat.py:33)+ 0.1 秒轮询的同步点(agents/group_chat/broadcast_queue.py:55),不是为高吞吐设计的。需要并发协作请看第 05 章。
AutoGenConversableAgent 的类文档自己写明:这层封装"目前不支持在 SK 里使用 AgentGroupChat"(agents/autogen/autogen_conversable_agent.py:108-110)。

13. 代码地图(导航索引)

主题	文件路径	符号名
Agent 基类(字段、三个调用方法、channel 管理)	`python/semantic_kernel/agents/agent.py`	`Agent`
线程抽象与生命周期	`python/semantic_kernel/agents/agent.py`	`AgentThread`、`create`、`delete`、`on_new_message`
线程入口收口(归一化 / 建线程 / 类型校验 / 通知)	`python/semantic_kernel/agents/agent.py`	`_ensure_thread_exists_with_messages`、`_notify_thread_of_new_message`
响应信封	`python/semantic_kernel/agents/agent.py`	`AgentResponseItem`
agent 即工具(自带的 kernel_function 包装)	`python/semantic_kernel/agents/agent.py`	`model_post_init`、`_as_kernel_function`
指令模板化与参数覆盖	`python/semantic_kernel/agents/agent.py`	`format_instructions`、`_merge_arguments`
插件注入(构造期 validator)	`python/semantic_kernel/agents/agent.py`	`_configure_plugins`
导出为 MCP server	`python/semantic_kernel/agents/agent.py`	`as_mcp_server`
声明式 spec 数据模型	`python/semantic_kernel/agents/agent.py`	`AgentSpec`、`ModelSpec`、`ToolSpec`、`InputSpec`
类型注册表与内建模块预载	`python/semantic_kernel/agents/agent.py`	`AGENT_TYPE_REGISTRY`、`register_agent_type`、`_preload_builtin_agents`
YAML / dict / 文件三个入口	`python/semantic_kernel/agents/agent.py`	`AgentRegistry`、`create_from_yaml`、`create_from_file`
spec 字段翻译与工具校验	`python/semantic_kernel/agents/agent.py`	`DeclarativeSpecMixin`、`_normalize_spec_fields`、`_validate_tools`
参考实现:本地 agent 全流程	`python/semantic_kernel/agents/chat_completion/chat_completion_agent.py`	`ChatCompletionAgent`、`_inner_invoke`
指令注入与工具消息补齐	`python/semantic_kernel/agents/chat_completion/chat_completion_agent.py`	`_prepare_agent_chat_history`、`_drain_mutated_messages`
本地线程实现与历史裁剪	`python/semantic_kernel/agents/chat_completion/chat_completion_agent.py`	`ChatHistoryAgentThread`、`reduce`
托管:OpenAI Assistants	`python/semantic_kernel/agents/open_ai/openai_assistant_agent.py`	`OpenAIAssistantAgent`、`AssistantAgentThread`
托管:OpenAI Responses(滚动 response_id)	`python/semantic_kernel/agents/open_ai/openai_responses_agent.py`	`OpenAIResponsesAgent`、`ResponsesAgentThread`
托管:Azure AI Foundry(含占位符解析)	`python/semantic_kernel/agents/azure_ai/azure_ai_agent.py`	`AzureAIAgent`、`AzureAIAgentThread`、`resolve_placeholders`
托管:Bedrock(只读线程)	`python/semantic_kernel/agents/bedrock/bedrock_agent.py`	`BedrockAgent`、`BedrockAgentThread`
托管:Copilot Studio(只读线程,无 channel)	`python/semantic_kernel/agents/copilot_studio/copilot_studio_agent.py`	`CopilotStudioAgent`、`CopilotStudioAgentThread`
桥接:AutoGen 0.2	`python/semantic_kernel/agents/autogen/autogen_conversable_agent.py`	`AutoGenConversableAgent`、`AutoGenConversableAgentThread`
对外导出清单	`python/semantic_kernel/agents/__init__.pyi`	`__all__`
通信协议抽象	`python/semantic_kernel/agents/channels/agent_channel.py`	`AgentChannel`
本地 channel(去重、可见性、内容过滤)	`python/semantic_kernel/agents/channels/chat_history_channel.py`	`ChatHistoryChannel`、`_is_message_visible`
群聊基类(飞行锁、channel 复用、广播)	`python/semantic_kernel/agents/group_chat/agent_chat.py`	`AgentChat`、`invoke_agent`、`_get_or_create_channel`
群聊多轮(选择 / 终止 / 裁剪)	`python/semantic_kernel/agents/group_chat/agent_group_chat.py`	`AgentGroupChat`、`invoke`、`reduce_history`
异步广播队列	`python/semantic_kernel/agents/group_chat/broadcast_queue.py`	`BroadcastQueue`、`enqueue`、`ensure_synchronized`
选择 / 终止策略	`python/semantic_kernel/agents/strategies/`	`SelectionStrategy`、`TerminationStrategy`
agent 当插件的真实样例	`python/samples/concepts/agents/chat_completion_agent/chat_completion_agent_as_kernel_function.py`	`triage_agent`
声明式 spec 样例	`python/samples/concepts/resources/declarative_spec/azure_ai_agent_spec.yaml`	—

接着读: 05-multi-agent-orchestration.md —— 本章的 Agent 被摆到 actor 运行时上,变成五种协作模式;AgentThread 在那边由 actor 持有。往回读:03-function-calling-and-filters.md 解释了 §6.4 里"谁在往历史里偷偷加消息"。

1. 这是什么(零基础也能懂)​

2. 顶层全景(三件套怎么转)​

3. 三件套逐个拆​

3.1 AgentThread:三个抽象方法撑起所有后端​

3.2 AgentResponseItem:为什么响应要背着线程​

3.3 Agent:三个调用方法的分工​

4. 每个 agent 出生自带一把工具柄​

5. 指令是模板,不是字符串​

6. 参考实现:ChatCompletionAgent 走一遍​

6.1 它的线程:ChatHistoryAgentThread​

6.2 一次非流式调用的六步​

6.3 _prepare_agent_chat_history:指令不进线程​

6.4 _drain_mutated_messages:补齐工具消息又不重复通知​

7. 声明式 agent:一份 YAML 造一个 agent​

7.1 Spec 数据模型​

7.2 类型注册表与 AgentRegistry​

7.3 DeclarativeSpecMixin:两个关键钩子​

8. 托管型 agent 家族:同一个契约,六种"线程"​

9. 旧版群聊路径:AgentChat + AgentChannel + BroadcastQueue(与第 05 章的编排体系并存)​

9.1 为什么需要 channel​

9.2 BroadcastQueue:广播是异步的,读取时才等​

9.3 AgentChat / AgentGroupChat:单飞行锁 + 选择/终止策略​

9.4 ChatHistoryChannel:去重与"可见性"​

9.5 与第 05 章新体系的关系​

10. 把 agent 变成一台 MCP server​

11. 巧妙之处(可带走的技术)​

12. 边界与局限​

13. 代码地图(导航索引)​