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的流程与源码对应关系以及如何用OrchestrationBaseInProcessRuntime在应用中落地一个可运行的多智能体系统。背景从单智能体到多智能体编排业界正沿着基础模型 → 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 的多智能体编排将直接依赖该共享运行时抽象而不是把运行时实现耦合进编排框架本身。核心术语表TermDefinitionActor运行时中可以发送和接收消息的实体Runtime负责促进 Actor 之间的通信并管理它们的状态与生命周期Runtime Abstraction为不同运行时实现提供统一接口的抽象层Agent一个 Semantic Kernel 智能体Orchestration包含 Actor 以及它们之间如何交互的规则这里刻意使用 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自行创建编排OrchestrationsDescriptionConcurrent并发适用于需要多个智能体独立分析同一任务并从中获益的场景Sequential顺序适用于需要明确、逐步推进的任务Handoff移交适用于动态变化、没有固定步骤的任务GroupChat群聊适用于需要多个智能体输入、且对话流程高度可配置的任务Magentic One类似 GroupChat但由基于 planner 的 manager 驱动灵感来自微软研究院的 Magentic One 系统每个编排的详细流程见后文「五种预置编排的运作机制」一节。应用层职责与生命周期约定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().hexorchestration_base.py并把它拼进 Actor 类型名如f{agent.name}_{internal_topic_type}从而保证共享同一 runtime 的多次调用互不干扰。支持结构化输入与输出编排需要接受结构化输入并返回结构化输出方便非聊天型编排在代码层面被使用尽管内部 Agent 仍是聊天式的。这一需求由TIn/TOut类型参数与input_transform/output_transform转换函数实现细节见下文「数据转换逻辑」。核心提案四大构建积木Building BlocksComponentDetailsAgent actorSemantic Kernel Agent 的包装器持有 Agent 上下文thread 与 historyData transform logic提供钩子将编排的输入/输出与自定义类型相互转换Orchestration由多个 Agent actor 及其他可选的编排专属 actor 组成Optional actors非 Agent actor 的其他 actor例如群聊编排中的 group manager actor总体结构如下图所示编排内部成员 Agent Actor、内部 Topic、可选 Actor 之间的直接消息与广播关系Agent ActorAgent 与运行时之间的桥梁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, descriptionagent.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_cacheChatHistory在每次 invoke 前暂存消息_invoke_agent内部通过self._agent.invoke_stream(...)以流式方式调用底层 Agent将分片缓冲为完整ChatMessageContent返回。各编排会派生自己的 Agent actor因为每种编排都有自己的消息处理器集合。例如群聊编排的 actorclass 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, defaultDefaultTypeAlias) TOut TypeVar(TOut, defaultDefaultTypeAlias) input_transform: Callable[[TIn], Awaitable[DefaultTypeAlias] | DefaultTypeAlias] output_transform: Callable[[DefaultTypeAlias], Awaitable[TOut] | TOut]其中TIn表示编排接受的输入类型TOut表示编排返回给调用者的输出类型默认类型是ChatMessageContent与list[ChatMessageContent]——即编排默认接受单条聊天消息或消息列表作为输入返回单条消息或消息列表。框架还提供了一套默认转换与工具函数以提升开发体验默认转换逻辑内置在 orchestration_base.py 中字符串输入会被包装为ChatMessageContent(roleAuthorRole.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 SequentialOrchestrationTIn, TOut : AgentOrchestrationTIn, TOut { ... } public sealed class SequentialOrchestration : SequentialOrchestrationChatMessageContent, ChatMessageContent { ... }OrchestrationResult异步获取结果编排结果对象设计如下class OrchestrationResult(KernelBaseModel, Generic[TOut]): value: TOut | None None event: asyncio.Event Field(default_factorylambda: asyncio.Event()) cancellation_token: CancellationToken Field(default_factorylambda: 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 AConcurrent Orchestration并发编排执行步骤编排被以一个任务 invoke编排将任务广播给所有 ActorActor 各自开始处理任务并把结果发送给 result collectorresult collector 收集结果当收到期望数量的结果时调用回调函数以宣告编排结束。实现要点concurrent.pyConcurrentOrchestration._start通过runtime.publish_message(ConcurrentRequestMessage(...), TopicId(internal_topic_type, ...))广播任务每个ConcurrentAgentActor处理完ConcurrentRequestMessage后把ConcurrentResponseMessage直接发送给CollectionActorCollectionActor内部用asyncio.Lock保护结果列表当len(self._results) self._expected_answer_count即成员数量时触发result_callback。注意并发结果的返回顺序不保证与成员列表顺序一致。Sequential Orchestration顺序编排执行步骤编排被以一个任务 invoke编排将任务发送给第一个Actor第一个 Actor 处理任务并把结果发送给下一个 Actor最后一个 Actor 处理结果并发送给 result collectorresult collector 调用回调函数宣告编排结束。实现要点sequential.py成员按逆序注册到 runtime使得当前 Actor 的下一跳 Actor 类型在注册时就已知next_actor_type从 collector 开始反向逐级链接_start只向members[0]发送首个SequentialRequestMessage每个SequentialAgentActor处理后把结果作为新的SequentialRequestMessage发给下一个 Actor最后的CollectionActor收到消息即触发result_callback。成员列表顺序即执行顺序。Handoff Orchestration移交编排执行步骤编排被以一个任务 invoke编排将任务发送给所有 Actor广播会话上下文编排向第一个 Actor 发送 request to speak 消息第一个 Actor 处理任务、广播会话上下文并决定是否需要将任务移交给另一个 Actor若决定移交则向目标 Actor 发送 request to speak 消息目标 Actor 处理任务并决定是否需要继续移交过程持续进行直到最后一个 Actor 判定任务完成调用回调宣告编排结束。实现要点handoffs.py这是实现细节最丰富的一种编排。OrchestrationHandoffs是一个dict[str, AgentHandoffs]描述源 Agent → 目标 Agent 及其移交描述的连接图并提供链式 APIadd/add_manyHandoffOrchestration.__init__会校验handoffs 不能为空、连接双方必须都是成员、Agent 不能移交给自己。每个HandoffAgentActor会在克隆的 Kernel上动态注入一个名为Handoff的插件包含每个移交连接对应一个transfer_to_{agent_name}函数KernelFunctionFromMethodpartial供 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群聊编排执行步骤编排被以一个任务 invoke编排将任务发送给所有 Actor编排将任务发送给 group manager触发群聊管理器启动编排group manager 根据会话状态做出以下决策之一Request User Input → 调用回调函数并等待用户输入Terminate终止Next Actor选择下一位发言者若需要继续group manager 选择下一个 Actor 并发送 request to speak 消息Actor 处理请求并把响应广播到内部 topic所有其他 Actor 收到响应并加入各自的会话上下文group manager 收到响应后回到第 4 步若会话结束group manager 取出结果并调用回调宣告编排结束。实现要点group_chat.pyGroupChatManagerActor是状态机式的可选 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 OrchestrationMagentic One 是一种类群聊编排但使用特殊的 group manager基于 planner整体灵感来自微软研究院的 Magentic One 通用型多智能体系统。在仓库中对应 magentic.py 与 step5_magentic.pyMagenticOrchestrationStandardMagenticManager继承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( namePhysicsExpert, instructionsYou are an expert in physics. You answer questions from a physics perspective., serviceAzureChatCompletion(credentialcredential), ) chemistry_agent ChatCompletionAgent( nameChemistryExpert, instructionsYou are an expert in chemistry. You answer questions from a chemistry perspective., serviceAzureChatCompletion(credentialcredential), ) return [physics_agent, chemistry_agent] async def main(): agents get_agents() concurrent_orchestration ConcurrentOrchestration(membersagents) # 创建并启动运行时 runtime InProcessRuntime() runtime.start() # invoke 是非阻塞的立即返回 OrchestrationResult orchestration_result await concurrent_orchestration.invoke( taskWhat is temperature?, runtimeruntime, ) # 等待结果可指定超时并发结果顺序不保证与成员顺序一致 value await orchestration_result.get(timeout20) 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每个示例对应一种编排示例文件演示内容step1_concurrent.py并发编排多专家并行回答同一问题step1a_concurrent_structured_outputs.py并发编排 结构化输出ConcurrentOrchestration[str, ArticleAnalysis]structured_outputs_transformstep2_sequential.py顺序编排概念提取 → 文案撰写 → 校对润色step2a_sequential_cancellation_token.py顺序编排 取消令牌step2b_sequential_streaming_agent_response_callback.py顺序编排 流式响应回调step3_group_chat.py群聊编排Writer 与 Reviewer 轮询迭代打磨标语RoundRobinGroupChatManager(max_rounds5)step3a_group_chat_human_in_the_loop.py群聊编排 人在回路manager 的human_response_functionstep3b_group_chat_with_chat_completion_manager.py基于 Chat Completion 的群聊管理器step4_handoff.py移交编排客服三线系统分诊/退款/订单状态/退货各 Agent 通过transfer_to_*函数互转step4a_handoff_structured_inputs.py移交编排 结构化输入step4b_handoff_streaming_agent_response_callback.py移交编排 流式响应回调step4c_handoff_mix_agent_types.py移交编排 混合 Agent 类型step5_magentic.pyMagentic 编排Research Coder代码解释器协作关于 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], managerRoundRobinGroupChatManager()) runtime InProcessRuntime() runtime.start() task_1 await group_chat.invoke(taskTASK_1, runtimeruntime) task_2 await group_chat.invoke(taskTASK_2, runtimeruntime) result_1 await task_1.get(timeout20) result_2 await task_2.get(timeout20) await runtime.stop_when_idle()如前文所述隔离的物理基础是每次 invoke 生成的唯一internal_topic_typeuuid.uuid4().hex以及将其拼入 Actor 类型名的命名策略f{agent.name}_{internal_topic_type}从源码结构看这保证了即使多个编排/多次调用共享同一 runtimeActor 类型与内部 topic 也不会碰撞。开放讨论面向未来的设计空间以下议题属于 ADR 记录的开放讨论Open Discussions不阻塞首版实现但为后续迭代预留了设计空间。状态管理State ManagementResume恢复进程仍存活但处于空闲状态等待某些事件以继续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声明式编排为用户提供低代码方案可与已有的声明式 Agentdeclarative 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考