企业官网建设流程全解析

Semantic Kernel 多智能体编排（Multi-agent Orchestration）架构解析与实践指南

【免费下载链接】semantic-kernelIntegrate cutting-edge LLM technology quickly and easily into your apps项目地址: https://gitcode.com/GitHub_Trending/se/semantic-kernel

本文以 0071-multi-agent-orchestration.md 这一架构决策记录（ADR）为核心骨架，结合当前仓库中 python/semantic_kernel/agents/orchestration 的真实实现与 multi_agent_orchestration 示例 展开。读者将掌握 Semantic Kernel 多智能体编排的设计动机、核心概念（Actor、Runtime、Orchestration）、五种预置编排模式（Concurrent、Sequential、Handoff、GroupChat、Magentic）的流程与源码对应关系，以及如何用OrchestrationBase+InProcessRuntime在应用中落地一个可运行的多智能体系统。

背景：从单智能体到多智能体编排

业界正沿着"基础模型 → RAG 系统 → 单个 AI 智能体 → 多智能体系统"的路径不断上移抽象层级。当单个 Agent 无法独立完成复杂任务时，多个智能体协作就成了刚需。Semantic Kernel Agent Framework 已提供稳定的 Agent 抽象（python/semantic_kernel/agents/agent.py）并支持 OpenAI Assistant、Chat Completion 等多种 Agent 服务，但在此框架之上仍缺少一个让多个智能体"协同工作"的层。本 ADR 正是为了解决这一问题：在 Agent Framework 之上构建多智能体编排框架。

这一方案还建立在与 AutoGen 团队合作的基础上——双方共享了 Agent Runtime 抽象，Semantic Kernel 的多智能体编排将直接依赖该共享运行时抽象，而不是把运行时实现耦合进编排框架本身。

核心术语表

这里刻意使用 "actor"（参与者）一词，是为了与 Semantic Kernel Agent Framework 中的 "agent" 区分开。另外，"pattern"（模式）与 "orchestration" 语义几乎等同，后者强调对模式的"管理与执行"；可以认为 "patterns" 是 "orchestrations" 的一种类型——例如 "concurrent orchestration" 就是遵循并发模式的一类编排。

来自 AutoGen 的共享运行时抽象

AutoGen 团队构建了一个运行时抽象（并附带一个进程内运行时实现），支持系统中 Actor 之间的 pub-sub（发布-订阅）通信。Semantic Kernel 直接复用了这份成果，形成了共享的 Agent Runtime 抽象。关键设计约束是：

• 根据运行时实现的不同，Actor 可以是本地的，也可以是分布式的；
• Semantic Kernel 的 Agent 框架不绑定任何特定运行时实现，即 runtime agnostic（运行时无关）。

这一约束在 python/semantic_kernel/agents/runtime/core/core_runtime.py 中得到验证：CoreRuntime是一个@runtime_checkable的 Protocol，定义了send_message、publish_message、register_factory、add_subscription、remove_subscription、get等统一接口，任何实现该协议的运行时（进程内、分布式）都可以被编排层使用。

设计考虑（Considerations）

预置编排（Orchestrations）总览

框架第一版提供覆盖最常见协作模式的预置编排，后续将按客户反馈持续扩充，并允许用户用框架提供的积木（building blocks）自行创建编排：

每个编排的详细流程见后文「五种预置编排的运作机制」一节。

应用层职责与生命周期约定

• Runtime 实例的生命周期由应用管理，且应在所有编排之外；
• 编排只在被 invoke 时才需要 runtime 实例，创建编排时不需要。

也就是说，编排是"运行时无关的模板"，应用负责创建、启动、停止 runtime，编排负责在运行时上注册 Actor 并驱动消息流。

图状结构与惰性求值

编排应被视为一个描述"智能体之间如何交互"的模板，类似一张有向图：

• Actor 在执行开始前才注册到 runtime，而不是在编排创建时注册；
• runtime 负责创建 Actor 并管理其生命周期。

这正是"惰性求值（lazy evaluation）"的体现：invoke时才把成员注册进运行时、建立通信通道。在实现中，_prepare抽象方法负责"注册 Actor 与订阅"，_start负责"真正启动流程"（见 orchestration_base.py）。

独立且隔离的调用（Invocation）

同一编排可以被多次 invoke，每次调用相互独立、彼此隔离，且可共享同一个 runtime 实例。为避免冲突（例如 Actor 名称或 ID 碰撞），必须定义清晰的调用边界。实现上，每次invoke都会生成一个唯一的内部 topic 类型：internal_topic_type = uuid.uuid4().hex（orchestration_base.py），并把它拼进 Actor 类型名（如f"{agent.name}_{internal_topic_type}"），从而保证共享同一 runtime 的多次调用互不干扰。

支持结构化输入与输出

编排需要接受结构化输入并返回结构化输出，方便非聊天型编排在代码层面被使用（尽管内部 Agent 仍是聊天式的）。这一需求由TIn/TOut类型参数与input_transform/output_transform转换函数实现，细节见下文「数据转换逻辑」。

核心提案：四大构建积木（Building Blocks）

总体结构如下图所示（编排内部：成员 Agent Actor、内部 Topic、可选 Actor 之间的直接消息与广播关系）：

Agent Actor：Agent 与运行时之间的桥梁

AgentActorBase是对 Semantic Kernel Agent 的包装，使其能够在运行时中收发消息，它继承自 AutoGen 的RoutedAgent类。ADR 中的原型如下：

class AgentActorBase(RoutedAgent): """A agent actor for multi-agent orchestration running on Agent runtime.""" def __init__(self, agent: Agent) -> None: """Initialize the agent container. Args: agent (Agent): An agent to be run in the container. """ self._agent = agent self._agent_thread = None # Chat history to temporarily store messages before the agent thread is created self._chat_history = ChatHistory() RoutedAgent.__init__(self, description=agent.description or "Semantic Kernel Agent")

在实际实现（python/semantic_kernel/agents/orchestration/agent_actor_base.py）中，AgentActorBase进一步演化为继承自ActorBase（其本身继承RoutedAgent），并补充了：

• internal_topic_type：该 actor 所属编排的内部 topic 类型，用于消息路由与隔离；
• exception_callback：异常回调，配合ActorBase.exception_handler装饰器，在消息处理异常时通知编排结果；
• agent_response_callback与streaming_agent_response_callback：观察者回调，分别在全量响应与流式响应（含is_final标记）产生时被调用；
• _message_cache（ChatHistory）：在每次 invoke 前暂存消息；
• _invoke_agent：内部通过self._agent.invoke_stream(...)以流式方式调用底层 Agent，将分片缓冲为完整ChatMessageContent返回。

各编排会派生自己的 Agent actor，因为每种编排都有自己的消息处理器集合。例如群聊编排的 actor：

class GroupChatAgentActor(AgentActorBase): """An agent actor for agents that process messages in a group chat.""" @message_handler async def _handle_start_message(self, message: GroupChatStartMessage, ctx: MessageContext) -> None: """Handle the initial message(s) provided by the user.""" ... @message_handler async def _handle_response_message(self, message: GroupChatResponseMessage, ctx: MessageContext) -> None: """Handle the response message from other agents in the group chat.""" ... @message_handler async def _handle_request_message(self, message: GroupChatRequestMessage, ctx: MessageContext) -> None: """Handle the request message from the group manager.""" ...

其他编排的 actor 处理的消息类型或数量不同。提案对编排内部 Actor 之间的交互方式不做任何限制——交互规则由各编排自行定义。

数据转换逻辑（Data Transform Logic）

转换函数签名如下：

DefaultTypeAlias = ChatMessageContent | list[ChatMessageContent] TIn = TypeVar("TIn", default=DefaultTypeAlias) TOut = TypeVar("TOut", default=DefaultTypeAlias) input_transform: Callable[[TIn], Awaitable[DefaultTypeAlias] | DefaultTypeAlias] output_transform: Callable[[DefaultTypeAlias], Awaitable[TOut] | TOut]

其中TIn表示编排接受的输入类型，TOut表示编排返回给调用者的输出类型；默认类型是ChatMessageContent与list[ChatMessageContent]——即编排默认接受"单条聊天消息或消息列表"作为输入，返回"单条消息或消息列表"。

框架还提供了一套默认转换与工具函数以提升开发体验：

• 默认转换逻辑内置在 orchestration_base.py 中：字符串输入会被包装为ChatMessageContent(role=AuthorRole.USER, content=...)；自定义TIn类型的输入会被json.dumps(input_message.__dict__)序列化进 user 消息；TOut为自定义 Pydantic 模型时，会按json.loads(output_message.content)反序列化回目标模型。
• 仓库还提供了structured_outputs_transform工具（tools.py）：给定目标 Pydantic 结构与支持结构化输出的ChatCompletionClientBase服务，返回一个输出转换函数，用target_structure.model_validate_json(response.content)将 LLM 输出解析为结构化对象。

Orchestration 基类：模板化执行

一个编排就是"一组 Semantic Kernel Agent + 它们之间交互的规则"。具体实现必须提供两段逻辑：

• **如何启动（start）**一次调用；
• **如何准备（prepare）**一次调用——即把 Actor 注册进 runtime，并按编排类型建立 Actor 之间的通信通道。

class OrchestrationBase(ABC, Generic[TIn, TOut]): def __init__( self, members: list[Agent], input_transform: Callable[[TIn], Awaitable[DefaultTypeAlias] | DefaultTypeAlias] | None = None, output_transform: Callable[[DefaultTypeAlias], Awaitable[TOut] | TOut] | None = None, ) -> None: """Initialize the orchestration base. Args: members (list[Agent]): The list of agents or orchestrations to be used. input_transform (Callable | None): A function that transforms the external input message. output_transform (Callable | None): A function that transforms the internal output message. """ ... async def invoke( self, task: str | DefaultTypeAlias | TIn, runtime: AgentRuntime, ) -> OrchestrationResult: """Invoke the orchestration and return an result immediately which can be awaited later. The runtime is supplied by the application at invocation time, not at creation time. Orchestrations are runtime-agnostic and can be used with any runtime that implements the runtime abstraction. """ orchestration_result = OrchestrationResult[TOut]() async def result_callback(result: DefaultTypeAlias) -> None: """Callback function that is called when the result is ready.""" ... ... # This unique topic type is used to isolate the invocation from others. internal_topic_type = uuid.uuid4().hex await self._prepare(runtime, internal_topic_type, result_callback) ... await self._start(runtime, internal_topic_type, orchestration_result.cancellation_token) return orchestration_result @abstractmethod async def _start( self, runtime: AgentRuntime, internal_topic_type: str, cancellation_token: CancellationToken, ) -> None: ... @abstractmethod async def _prepare( self, runtime: AgentRuntime, internal_topic_type: str, result_callback: Callable[[DefaultTypeAlias], Awaitable[None]] | None = None, ) -> str: ...

这一设计在实际实现中得到了完整落地，并且invoke是非阻塞的：它立即返回OrchestrationResult，并把_start包装为后台asyncio.Task执行（orchestration_base.py）。类型参数既可以在类上显式指定（如ConcurrentOrchestrationstr, ArticleAnalysis），也可以通过 TypeVar 的默认值推导。

用户使用编排时，可以按需设置TIn/TOut并传入输入/输出转换函数。Python 示例：

class MyTypeA: pass class MyTypeB: pass sequential_orchestration = SequentialOrchestrationMyTypeA, MyTypeB

框架提供默认值，因此只有高级用户才需要显式指定TIn/TOut。在 .NET 中则可以通过"非泛型密封类继承泛型基类"来达到类似效果：

public class SequentialOrchestration<TIn, TOut> : AgentOrchestration<TIn, TOut> { ... } public sealed class SequentialOrchestration : SequentialOrchestration<ChatMessageContent, ChatMessageContent> { ... }

OrchestrationResult：异步获取结果

编排结果对象设计如下：

class OrchestrationResult(KernelBaseModel, Generic[TOut]): value: TOut | None = None event: asyncio.Event = Field(default_factory=lambda: asyncio.Event()) cancellation_token: CancellationToken = Field(default_factory=lambda: CancellationToken()) async def get(self, timeout: float | None = None) -> TOut: """Get the result of the invocation. Args: timeout (float | None): The timeout in seconds. If None, wait indefinitely. Raises: TimeoutError: If the timeout is reached before the result is ready. RuntimeError: If the invocation is cancelled. Returns: TOut: The result of the invocation. """ ... def cancel(self) -> None: """Cancel the invocation. This method will cancel the invocation and set the cancellation token. Actors that have received messages will continue to process them, but no new messages will be processed. """ ...

实际实现（orchestration_base.py）在此基础上补充了background_task与exception字段：异常既可能由内部result_callback路径产生，也可能来自_start后台任务本身（通过add_done_callback捕获并写入结果对象），因此get()在超时、取消、异常、无结果四种情形下都有明确的错误语义。cancel()会取消调用：已经收到消息的 Actor 会继续处理完，但不再处理新消息。

五种预置编排的运作机制（Appendix A）

Concurrent Orchestration（并发编排）

执行步骤：

1. 编排被以一个任务 invoke；
2. 编排将任务广播给所有 Actor；
3. Actor 各自开始处理任务，并把结果发送给 result collector；
4. result collector 收集结果，当收到期望数量的结果时，调用回调函数以宣告编排结束。

实现要点（concurrent.py）：ConcurrentOrchestration._start通过runtime.publish_message(ConcurrentRequestMessage(...), TopicId(internal_topic_type, ...))广播任务；每个ConcurrentAgentActor处理完ConcurrentRequestMessage后，把ConcurrentResponseMessage直接发送给CollectionActor；CollectionActor内部用asyncio.Lock保护结果列表，当len(self._results) == self._expected_answer_count（即成员数量）时触发result_callback。注意：并发结果的返回顺序不保证与成员列表顺序一致。

Sequential Orchestration（顺序编排）

执行步骤：

1. 编排被以一个任务 invoke；
2. 编排将任务发送给第一个Actor；
3. 第一个 Actor 处理任务，并把结果发送给下一个 Actor；
4. 最后一个 Actor 处理结果，并发送给 result collector；
5. result collector 调用回调函数宣告编排结束。

实现要点（sequential.py）：成员按逆序注册到 runtime，使得"当前 Actor 的下一跳 Actor 类型"在注册时就已知（next_actor_type从 collector 开始反向逐级链接）；_start只向members[0]发送首个SequentialRequestMessage；每个SequentialAgentActor处理后把结果作为新的SequentialRequestMessage发给下一个 Actor；最后的CollectionActor收到消息即触发result_callback。成员列表顺序即执行顺序。

Handoff Orchestration（移交编排）

执行步骤：

1. 编排被以一个任务 invoke；
2. 编排将任务发送给所有 Actor（广播会话上下文）；
3. 编排向第一个 Actor 发送 "request to speak" 消息；
4. 第一个 Actor 处理任务、广播会话上下文，并决定是否需要将任务移交给另一个 Actor；
5. 若决定移交，则向目标 Actor 发送 "request to speak" 消息；
6. 目标 Actor 处理任务并决定是否需要继续移交；
7. 过程持续进行，直到最后一个 Actor 判定任务完成，调用回调宣告编排结束。

实现要点（handoffs.py）：这是实现细节最丰富的一种编排。OrchestrationHandoffs是一个dict[str, AgentHandoffs]，描述"源 Agent → 目标 Agent 及其移交描述"的连接图，并提供链式 APIadd/add_many；HandoffOrchestration.__init__会校验：handoffs 不能为空、连接双方必须都是成员、Agent 不能移交给自己。每个HandoffAgentActor会在克隆的 Kernel上动态注入一个名为Handoff的插件，包含：

• 每个移交连接对应一个transfer_to_{agent_name}函数（KernelFunctionFromMethod+partial），供 LLM 通过函数调用触发移交；
• 一个complete_task(task_summary)函数，用于宣告任务完成并携带总结；
• 一个AUTO_FUNCTION_INVOCATION过滤器：当模型调用Handoff插件函数时，设置context.terminate = True终止当前 Agent 的自动函数调用循环。

当 Actor 被请求发言时，它会以_invoke_agent_with_potentially_no_response调用 Agent（与_invoke_agent不同，该方法在无响应时返回None而非抛错，因为移交函数可能终止调用循环）；随后进入决策循环：若设置了移交目标则广播HandoffRequestMessage，否则广播响应。HITL 方面，human_response_function对所有Agent 可见（群聊中则仅 manager 可见）。

Group Chat Orchestration（群聊编排）

执行步骤：

1. 编排被以一个任务 invoke；
2. 编排将任务发送给所有 Actor；
3. 编排将任务发送给 group manager，触发群聊管理器启动编排；
4. group manager 根据会话状态做出以下决策之一：
   • Request User Input → 调用回调函数并等待用户输入；
   • Terminate（终止）；
   • Next Actor（选择下一位发言者）；
5. 若需要继续，group manager 选择下一个 Actor 并发送 "request to speak" 消息；
6. Actor 处理请求并把响应广播到内部 topic；
7. 所有其他 Actor 收到响应并加入各自的会话上下文；
8. group manager 收到响应后回到第 4 步；
9. 若会话结束，group manager 取出结果并调用回调宣告编排结束。

实现要点（group_chat.py）：GroupChatManagerActor是状态机式的可选 Actor，其决策循环由_determine_state_and_take_action驱动，依次执行should_request_user_input→should_terminate→select_next_agent；终止时用filter_results从聊天历史中提取最终结果，并把termination_reason、filter_result_reason写入结果的 metadata。GroupChatOrchestration._start会先asyncio.gather向所有成员并发发送GroupChatStartMessage，再向 manager 发送启动消息——因为若 manager 处理过快而其他 Actor 太慢，可能在成员尚未具备必要上下文时就发出"请求发言"，导致上下文缺失。另外注意：群聊编排要求所有成员都必须有 description（构造时校验），因为 manager 需要借助成员描述来挑选下一位发言者。

群聊管理器接口定义如下：

class GroupChatManager(KernelBaseModel, ABC): """A group chat manager that manages the flow of a group chat.""" user_input_func: Callable[[ChatHistory], Awaitable[str]] | None = None @abstractmethod async def should_request_user_input(self, chat_history: ChatHistory) -> bool: raise NotImplementedError @abstractmethod async def should_terminate(self, chat_history: ChatHistory) -> bool: raise NotImplementedError @abstractmethod async def select_next_agent(self, chat_history: ChatHistory, participant_descriptions: dict[str, str]) -> str: raise NotImplementedError @abstractmethod async def filter_results(self, chat_history: ChatHistory) -> ChatMessageContent: raise NotImplementedError

在实际实现中，接口演化为返回类型化的BooleanResult/StringResult/MessageResult（子类化GroupChatManagerResult[T]，因为 OpenAI 等模型服务不支持泛型类名），并新增了current_round、max_rounds与human_response_function字段。内置的RoundRobinGroupChatManager提供了默认实现：不请求用户输入、按(current_index + 1) % len(participants)轮询选择下一位、把聊天历史的最后一条消息作为结果。仓库还提供了基于 Chat Completion 选择发言者的ChatCompletionGroupChatManager（见 step3b 示例）。

Magentic One Orchestration

Magentic One 是一种类群聊编排，但使用特殊的 group manager（基于 planner），整体灵感来自微软研究院的 Magentic One 通用型多智能体系统。在仓库中对应 magentic.py 与 step5_magentic.py：

• MagenticOrchestration+StandardMagenticManager（继承MagenticManagerBase）组成了编排主体；
• Standard manager 使用了经过精心调校的提示词（task ledger、progress ledger 等，见 prompts/_magentic_prompts.py），支持替换自定义提示词，甚至子类化MagenticManagerBase实现自己的管理器逻辑；
• 注意前提条件：manager 需要一个支持结构化输出的聊天补全模型（示例使用gpt-4o-search-preview等模型驱动的 Research/Coder 双 Agent 协作）。

端到端使用模式：如何运行一个多智能体编排

无论是哪种编排，其使用模式完全一致，与 ADR 中的示例吻合（下面以仓库实际示例为准）：

import asyncio from azure.identity import AzureCliCredential from semantic_kernel.agents import Agent, ChatCompletionAgent, ConcurrentOrchestration from semantic_kernel.agents.runtime import InProcessRuntime from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion def get_agents() -> list[Agent]: credential = AzureCliCredential() physics_agent = ChatCompletionAgent( name="PhysicsExpert", instructions="You are an expert in physics. You answer questions from a physics perspective.", service=AzureChatCompletion(credential=credential), ) chemistry_agent = ChatCompletionAgent( name="ChemistryExpert", instructions="You are an expert in chemistry. You answer questions from a chemistry perspective.", service=AzureChatCompletion(credential=credential), ) return [physics_agent, chemistry_agent] async def main(): agents = get_agents() concurrent_orchestration = ConcurrentOrchestration(members=agents) # 创建并启动运行时 runtime = InProcessRuntime() runtime.start() # invoke 是非阻塞的，立即返回 OrchestrationResult orchestration_result = await concurrent_orchestration.invoke( task="What is temperature?", runtime=runtime, ) # 等待结果（可指定超时；并发结果顺序不保证与成员顺序一致） value = await orchestration_result.get(timeout=20) for item in value: print(f"# {item.name}: {item.content}") # 空闲时优雅停止运行时 await runtime.stop_when_idle() if __name__ == "__main__": asyncio.run(main())

完整可运行示例位于 python/samples/getting_started_with_agents/multi_agent_orchestration，每个示例对应一种编排：

关于 InProcessRuntime

进程内运行时（python/semantic_kernel/agents/runtime/in_process/in_process_runtime.py）是所有示例的基础设施。其关键行为：

• start()：在后台任务中启动消息处理循环（仅可调用一次）；
• send_message/publish_message：分别实现"点对点直接发送（RPC 语义，等待响应）"与"向订阅了某 topic 的所有 Actor 广播"两种通信原语，消息通过单一asyncio.Queue排队、每条消息在独立 task 中并发处理；
• stop()/stop_when_idle()/close()：立即停止 / 队列清空后停止 / 停止并关闭所有实例化 Agent。stop_when_idle是文档推荐的常用停止方式；
• ignore_unhandled_exceptions：构造参数，默认True——设为False时，若执行期间发生异常，runtime 会停止并抛出；
• register_factory：按唯一 type 注册 Agent 工厂，编排的_prepare正是通过它把 Agent actor 注册进 runtime（工厂内部可通过AgentInstantiationContext访问当前 runtime 与 agent ID，即"惰性创建"）；
• save_state/load_state：保存/恢复所有实例化 Agent 的状态（暂不包含订阅状态）。

多次调用与调用隔离

同一编排可以多次 invoke，且可共享同一个 runtime。ADR 中的示例（task_1与task_2完全独立、不共享上下文）：

agent_1 = ChatCompletionAgent(...) agent_2 = ChatCompletionAgent(...) group_chat = GroupChatOrchestration(members=[agent_1, agent_2], manager=RoundRobinGroupChatManager()) runtime = InProcessRuntime() runtime.start() task_1 = await group_chat.invoke(task=TASK_1, runtime=runtime) task_2 = await group_chat.invoke(task=TASK_2, runtime=runtime) result_1 = await task_1.get(timeout=20) result_2 = await task_2.get(timeout=20) await runtime.stop_when_idle()

如前文所述，隔离的物理基础是每次 invoke 生成的唯一internal_topic_type（uuid.uuid4().hex）以及将其拼入 Actor 类型名的命名策略（f"{agent.name}_{internal_topic_type}"），从源码结构看，这保证了即使多个编排/多次调用共享同一 runtime，Actor 类型与内部 topic 也不会碰撞。

开放讨论：面向未来的设计空间

以下议题属于 ADR 记录的开放讨论（Open Discussions），不阻塞首版实现，但为后续迭代预留了设计空间。

状态管理（State Management）

• Resume（恢复）：进程仍存活但处于空闲状态，等待某些事件以继续；runtime 从空闲状态恢复进程。
• Restart（重启）：进程已停止（手动停止或出错）；编排可以从零开始重启，也可以从之前的 checkpoint 重启。重启是幂等的——同一 checkpoint 可以多次重启，而不会对编排、runtime 和 Agent 产生副作用。

编排既可能长时运行（数小时、数天甚至数年），也可能短时运行（几分钟、几秒甚至更短）。其状态可能包括：活跃运行但空闲等待用户输入或其他事件、进入错误状态等。从空闲状态恢复由 runtime 负责（保存 Actor 状态、恢复时重新水合）；Agent 的对话上下文（threads 与 memories）则属于另一类状态，需要与编排框架协同设计。

Agent 上下文（Agent Context）

编排不管理 Agent 状态，但希望支持"在已有 Agent 上下文上 invoke/restart 编排"。一种候选方案是引入 context provider：按 Agent ID 提供 Agent 上下文，并附着到 Agent actor 上供其读取与更新；每次新的调用会返回编排的文本表示（见"声明式编排"），用于后续重新水合编排。

错误处理（Error Handling）

应用管理 runtime，因此编排无法捕获发生在 runtime 与 actor 层面的错误。当前InProcessRuntime提供ignore_unhandled_exceptions标志（默认True，构造时设置；设为False会让 runtime 停止并在执行异常时抛出）。分布式 runtime 场景下错误处理会更复杂，还需要在 runtime 层面考虑重试与幂等。

人在回路（Human in the Loop）

这是自主系统的关键组成部分，需要支持：取消一次调用、向用户通知重要事件、支持分布式场景（客户端与编排不在同一系统）。当前群聊与移交编排已提供实验性的人机交互能力：

• 群聊：manager 的human_response_function（见 step3a 示例）；
• 移交：所有 Agent 共享的human_response_function；
• 取消：OrchestrationResult.cancel()与CancellationToken。

组合（Composition）

组合允许把已有编排当作积木去构建更强大的编排（例如把编排中的一个 Agent 替换为另一个编排）。挑战包括：编排输入/输出类型不匹配的处理、Actor 与编排之间的通信、嵌套编排的生命周期管理、嵌套编排事件的向上传播，以及使用/实现两方面的简洁性。

分布式编排（Distributed Orchestrations）

编排虽不与特定 runtime 绑定，但仍需回答：Actor 工厂是否需要分布式？runtime 如何处理分布式 Actor 故障？分布式编排的取消如何实现？分布式场景下结果如何通过回调或其他机制返回？

声明式编排（Declarative Orchestrations）

声明式编排为用户提供低代码方案，可与已有的声明式 Agent（declarative agents）工作复用，实现声明式编排。

护栏（Guardrails）

安全是优先级之一：编排能力越强，潜在危害越大。需要讨论护栏应放在编排层、actor 层还是 agent 层（类似 OpenAI Agent SDK 的 guardrails 概念）。

可观测性（Observability）

作为企业级方案，编排框架需要纳入可观测性设计。

运行时之前的安全中间层

可以考虑在 runtime 之前增加一层，标准化所有 Actor 间的消息，以获得：

• 内置幂等与重试：标准化消息携带 id、causation_id、retry_count、ttl 等字段，支持确定性去重、用于遥测的因果图和安全重投递；
• 一流的可观测性：标准化消息字段可 1:1 映射到 OpenTelemetry 属性，实现每一跳的可追踪与指标；
• 持久化/重新水合：标准化消息可序列化存储并按需反序列化；
• 护栏集中化：统一包装层让策略/护栏检查集中在 runtime，确保没有消息未经检查就到达 Agent。

范围外（Out of Scope）与版本状态

• Runtime 实现本身不在本提案范围内（本文档只约束编排层如何依赖运行时抽象）；
• 开放讨论中的议题不在首版实现内，但会为未来扩展预留空间。

需要说明的是，ADR 元数据中标明status: proposed、日期 2025-04-30；而从当前仓库的实际代码来看，该提案中的核心设计（OrchestrationBase、AgentActorBase、五种预置编排、InProcessRuntime、CoreRuntime协议、OrchestrationResult等）均已在python/semantic_kernel/agents/orchestration/与python/semantic_kernel/agents/runtime/中以@experimental标记落地，并配套了成体系的入门示例，读者可直接对照本文各节给出的文件路径深入研读源码与测试。

【免费下载链接】semantic-kernelIntegrate cutting-edge LLM technology quickly and easily into your apps项目地址: https://gitcode.com/GitHub_Trending/se/semantic-kernel