Semantic Kernel Python 多智能体编排实战从并发到 Magentic 的五种协同模式【免费下载链接】semantic-kernelIntegrate cutting-edge LLM technology quickly and easily into your apps项目地址: https://gitcode.com/GitHub_Trending/se/semantic-kernel本文围绕 Semantic Kernel Agent Framework 的多智能体编排Multi-agent orchestration能力展开逐一讲解 Concurrent并发、Sequential顺序、Handoff交接、GroupChat群聊与 Magentic规划器驱动五种协同模式的适用场景、核心 API 与完整示例并结合仓库源码剖析其运行机制。读完本文你将掌握如何让多个 Agent 以并行、流水线、动态交接或自由讨论的方式协同完成一个复杂任务并学会处理结构化输出、取消、流式回调与人在环Human-in-the-loop等实战细节。背景什么是多智能体编排Semantic Kernel 的 Agent Framework 现已支持编排多个 Agent 协同完成同一任务。在开始本文的编排示例之前建议先掌握以下前置概念Chat Completion 基础自动函数调用Auto Function Calling结构化输出Structured OutputAgent 入门示例更进阶的 Agent 示例本文全部示例位于仓库的 multi_agent_orchestration 目录共 14 个文件覆盖 5 种编排模式及其进阶变体multi_agent_orchestration/ ├── README.md ├── observability.py # 可观测性辅助Azure Monitor OpenTelemetry ├── step1_concurrent.py # 并发编排 ├── step1a_concurrent_structured_outputs.py # 并发编排 结构化输出 ├── step2_sequential.py # 顺序编排 ├── step2a_sequential_cancellation_token.py # 顺序编排 取消 ├── step2b_sequential_streaming_agent_response_callback.py # 顺序编排 流式回调 ├── step3_group_chat.py # 群聊编排轮询管理器 ├── step3a_group_chat_human_in_the_loop.py # 群聊 人在环 ├── step3b_group_chat_with_chat_completion_manager.py # 群聊 Chat Completion 管理器 ├── step4_handoff.py # 交接编排 ├── step4a_handoff_structured_inputs.py # 交接 结构化输入 ├── step4b_handoff_streaming_agent_response_callback.py # 交接 流式回调 ├── step4c_handoff_mix_agent_types.py # 交接 混合 Agent 类型 └── step5_magentic.py # Magentic 编排运行前提环境变量示例依赖以下两个环境变量使用 OpenAI 服务时OPENAI_API_KEYOpenAI API 密钥OPENAI_CHAT_MODEL_IDOpenAI Chat 模型 ID例如gpt-4o如果你使用其他模型服务可以在示例中自由切换为对应连接器环境变量的配置方式参考 setup 指南。以 OpenAI 为例openai_env_setup.py 展示了通过 pydantic settings 从环境变量读取api_key、org_id、chat_model_id等配置的机制其中chat_model_id会被传入OpenAIChatCompletion的ai_model_id参数。五种编排模式总览编排模式Orchestration描述Concurrent并发适用于需要多个 Agent 对同一任务进行独立分析、并行获益的场景。Sequential顺序适用于需要明确定义逐步执行路径的任务前一个 Agent 的输出作为后一个 Agent 的输入。Handoff交接适用于动态变化、没有固定逐步路径的任务Agent 之间可动态转移会话。GroupChat群聊适用于需要多个 Agent 输入、且对话流程可高度配置的场景。Magentic类似 GroupChat 但由基于规划器的管理器驱动其设计受 Microsoft 的 Magentic One一个面向复杂任务的通用多智能体系统研究启发。每种模式都遵循同一套核心 API 使用范式这也是理解全部示例的关键。公共骨架Runtime、Orchestration 与异步结果尽管五种编排的行为不同但它们的调用方式高度一致可以抽象为五步创建编排对象ConcurrentOrchestration/SequentialOrchestration/GroupChatOrchestration/HandoffOrchestration/MagenticOrchestration传入成员 Agent 列表创建运行时InProcessRuntime并启动调用编排对象的invoke(task..., runtime...)发起任务非阻塞立即返回在返回的OrchestrationResult上调用get(timeout...)等待结果调用await runtime.stop_when_idle()优雅停止运行时。从源码看这一范式由 orchestration_base.py 中的OrchestrationBase抽象基类统一定义。其invoke()方法见orchestration_base.py的invoke实现将字符串任务包装为ChatMessageContent(roleAuthorRole.USER, ...)创建后台asyncio.Task执行_start()并通过OrchestrationResult暴露结果。OrchestrationResult提供get(timeout: float | None)阻塞等待结果指定超时时若未完成会抛出TimeoutError但不会中止编排任务被取消时抛出RuntimeError(The invocation was canceled before it could complete.)cancel()取消编排——已收到消息的 Actor 会继续处理完但不会再处理新消息见orchestration_base.py中cancel()的注释说明。OrchestrationBase还支持泛型[TIn, TOut]与input_transform/output_transform用于自定义输入输出类型的转换默认实现会将 Pydantic 对象序列化为 JSON 字符串见_default_input_transform/_default_output_transform。模式一Concurrent 并发编排并发编排适用于多个专家对同一问题各自作答的场景。在 step1_concurrent.py 中物理学家与化学家两个 Agent 被并行要求回答温度是什么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), ) concurrent_orchestration ConcurrentOrchestration(members[physics_agent, chemistry_agent]) runtime InProcessRuntime() runtime.start() orchestration_result await concurrent_orchestration.invoke( taskWhat is temperature?, runtimeruntime, ) # 注意结果顺序与 Agent 在编排中的顺序不保证一致 value await orchestration_result.get(timeout20) for item in value: print(f# {item.name}: {item.content}) await runtime.stop_when_idle()运行后你会看到 PhysicsExpert 与 ChemistryExpert 各自的完整回答。并行结果是无序的代码注释与示例输出均强调这一点get返回的是结果列表遍历时应按item.name区分来源。并发编排 结构化输出step1a_concurrent_structured_outputs.py 演示了并发编排与结构化输出的结合用三个 Agent主题识别、情感分析、实体识别分析《哈姆雷特》全文摘要最终聚合为一个 Pydantic 模型class ArticleAnalysis(BaseModel): themes: list[str] sentiments: list[str] entities: list[str] concurrent_orchestration ConcurrentOrchestrationstr, ArticleAnalysis), )要点启用结构化输出必须同时指定output_transform与泛型类型参数[str, ArticleAnalysis]structured_outputs_transform位于 tools.py从semantic_kernel.agents.orchestration.tools导入所需的 Chat Completion 服务与模型必须支持结构化输出任务文本从resources目录的Hamlet_full_play_summary.txt文件读取相对路径为../resources/Hamlet_full_play_summary.txt结果通过value.model_dump_json(indent2)输出示例给出了包含themes、sentiments、entities三个字段的 JSON 结果。模式二Sequential 顺序编排顺序编排将 Agent 按列表顺序串联执行前一个 Agent 的输出自动成为后一个 Agent 的输入。 step2_sequential.py 用三段式流水线完成营销文案生产营销分析师提炼卖点 → 文案作者撰写 → 编辑校对润色sequential_orchestration SequentialOrchestration( members[concept_extractor_agent, writer_agent, format_proof_agent], agent_response_callbackagent_response_callback, ) orchestration_result await sequential_orchestration.invoke( taskAn eco-friendly stainless steel water bottle that keeps drinks cold for 24 hours, runtimeruntime, ) value await orchestration_result.get(timeout20) print(f***** Final Result *****\n{value})三个 Agent 的职责通过instructions明确分工ConceptExtractorAgent营销分析师从产品描述中识别关键特性、目标受众、独特卖点WriterAgent营销文案作者将要点写成约 150 词的短文案要求只输出单块文本FormatProofAgent编辑修正语法、提升清晰度、统一语气并润色输出。从源码看SequentialAgentActor见 sequential.py持有_next_agent_type处理完本 Agent 任务后会把响应路由给链上的下一个 Actor从而形成输出即输入的传递链。变体 A取消正在运行的编排step2a_sequential_cancellation_token.py 演示如何在编排完成前取消它orchestration_result await sequential_orchestration.invoke(task..., runtimeruntime) await asyncio.sleep(1) # 模拟延迟 orchestration_result.cancel() try: _ await orchestration_result.get(timeout20) except Exception as e: print(e) # 输出: The invocation was canceled before it could complete. finally: await runtime.stop_when_idle()取消后调用get()会抛出RuntimeError。示例同时展示了针对编排模块的日志配置logging.basicConfig(levellogging.WARNING) logging.getLogger(semantic_kernel.agents.orchestration.sequential).setLevel(logging.DEBUG)DEBUG 日志会显示各 Agent Actor 的注册信息如Registered agent actor of type ConceptExtractorAgent_uuid以及Sequential actor ... started processing.../finished processing.的处理过程方便你观察编排内部流转。变体 B流式响应回调step2b_sequential_streaming_agent_response_callback.py 用streaming_agent_response_callback实时打印每个 Agent 的流式输出is_new_message True def streaming_agent_response_callback(message: StreamingChatMessageContent, is_final: bool) - None: global is_new_message if is_new_message: print(f# {message.name}) is_new_message False print(message.content, end, flushTrue) if is_final: print() is_new_message True sequential_orchestration SequentialOrchestration( membersagents, streaming_agent_response_callbackstreaming_agent_response_callback, )回调接收两个参数流式消息内容StreamingChatMessageContent与结束标志is_final。示例利用模块级布尔标志在每个 Agent 开始时打印其名字并在is_final为真时换行实现按 Agent 分段的实时流式输出。模式三GroupChat 群聊编排群聊编排由**群聊管理器Group Chat Manager**控制对话流程。示例代码将管理器描述为一个状态机具有以下可能状态请求用户消息Request for user message终止Termination之后管理器会尝试从对话中过滤出结果继续Continuation管理器选择下一个发言的 Agent默认轮询管理器Round Robinstep3_group_chat.py 用默认的RoundRobinGroupChatManager让 Writer 与 Reviewer 两个 Agent 轮流发言迭代打磨一款电动 SUV 的广告语group_chat_orchestration GroupChatOrchestration( members[writer, reviewer], # max_rounds 为奇数保证 Writer 获得最后一轮发言权 managerRoundRobinGroupChatManager(max_rounds5), agent_response_callbackagent_response_callback, ) orchestration_result await group_chat_orchestration.invoke( taskCreate a slogan for a new electric SUV that is affordable and fun to drive., runtimeruntime, ) value await orchestration_result.get()示例输出展示了完整的迭代过程Writer 给出第一版广告语 → Reviewer 反馈 → Writer 修订 → Reviewer 再反馈 → Writer 给出最终版get()返回的最终结果是收敛后的广告语。代码注释特别指出max_rounds设为奇数是为了让 Writer 拿到最后一轮发言。变体 A人在环Human-in-the-loopstep3a_group_chat_human_in_the_loop.py 通过继承并覆写默认管理器实现用户介入。核心是覆写should_request_user_input本例要求Reviewer 发言后必须征求用户意见class CustomRoundRobinGroupChatManager(RoundRobinGroupChatManager): override async def should_request_user_input(self, chat_history: ChatHistory) - BooleanResult: if len(chat_history.messages) 0: return BooleanResult(resultFalse, reasonNo agents have spoken yet.) last_message chat_history.messages[-1] if last_message.name Reviewer: return BooleanResult(resultTrue, reasonUser input is needed after the reviewers message.) return BooleanResult(resultFalse, reasonUser input is not needed if the last message is not from the reviewer.)同时提供human_response_function获取键盘输入并包装为用户消息async def human_response_function(chat_history: ChatHistory) - ChatMessageContent: user_input input(User: ) return ChatMessageContent(roleAuthorRole.USER, contentuser_input) managerCustomRoundRobinGroupChatManager( max_rounds5, human_response_functionhuman_response_function, )示例输出展示用户输入希望广告语押韵、Reviewer 给出建议后用户再补充指示对话因此沿用户意图方向演进。BooleanResult等结果类型定义在 group_chat.py由GroupChatManagerResult泛型基类派生均包含result与reason两个字段。变体 B基于 Chat Completion 的管理器step3b_group_chat_with_chat_completion_manager.py 展示如何让 LLM 承担主持人角色8 个来自不同地域与职业背景的 Agent农民、开发者、教师、活动家、精神领袖、艺术家、移民、医生围绕什么是对你而言的好生活展开辩论管理器用 LLM 完成终止判断、发言人选择与结果归纳。核心是继承GroupChatManager并实现四个方法should_request_user_input返回本管理器不需要用户输入should_terminate向聊天历史注入termination_prompt判断讨论是否已达成结论是则回答 True调用服务并以BooleanResult作为response_format解析结果select_next_agent注入selection_prompt给出主题与参与者名单描述请求返回下一个发言者名字解析为StringResult并校验其确实在参与者列表中否则抛出RuntimeErrorfilter_results讨论结束后注入result_filter_prompt让 LLM 总结讨论并给出闭幕陈述返回MessageResult。示例中的三个提示词模板使用{{$topic}}、{{$participants}}等 Handlebars 风格变量通过KernelPromptTemplate渲染见_render_prompt辅助方法。运行日志会打印Should terminate: ... Reason: ...与Next participant: ... Reason: ...供观察。注意该方式要求所用模型支持结构化输出response_format绑定 Pydantic 模型。模式四Handoff 交接编排交接编排面向没有固定路径、按会话内容动态转移的场景。 step4_handoff.py 实现了一个客服分诊系统分诊 Agent 将客户按问题类型移交给退款 / 订单状态 / 退货 三个专业 Agent专业 Agent 也可把不属于自己职责的问题交回分诊。通过插件为 Agent 注入工具退款、订单状态、退货三个 Agent 各自挂载了一个原生插件class OrderStatusPlugin: kernel_function def check_order_status(self, order_id: str) - str: return fOrder {order_id} is shipped and will arrive in 2-3 days. order_status_agent ChatCompletionAgent( nameOrderStatusAgent, descriptionA customer support agent that checks order status., instructionsHandle order status requests., serviceAzureChatCompletion(credentialcredential), plugins[OrderStatusPlugin()], )定义交接关系OrchestrationHandoffs定义于 handoffs.py本质是源 Agent → {目标 Agent: 交接描述}的字典用链式 API 描述交接图谱handoffs ( OrchestrationHandoffs() .add_many( source_agentsupport_agent.name, target_agents{ refund_agent.name: Transfer to this agent if the issue is refund related, order_status_agent.name: Transfer to this agent if the issue is order status related, order_return_agent.name: Transfer to this agent if the issue is order return related, }, ) .add(source_agentrefund_agent.name, target_agentsupport_agent.name, descriptionTransfer to this agent if the issue is not refund related) .add(source_agentorder_status_agent.name, target_agentsupport_agent.name, descriptionTransfer to this agent if the issue is not order status related) .add(source_agentorder_return_agent.name, target_agentsupport_agent.name, descriptionTransfer to this agent if the issue is not order return related) ) handoff_orchestration HandoffOrchestration( membersagents, handoffshandoffs, agent_response_callbackagent_response_callback, human_response_functionhuman_response_function, )在HandoffOrchestration中所有 Agent 都能访问human_response_function与群聊中只有管理器可访问不同。示例中的agent_response_callback除了打印消息内容还会识别消息项中的FunctionCallContent与FunctionResultContent打印Handoff-transfer_to_OrderStatusAgent这类内部交接调用及插件函数调用如OrderStatusPlugin-check_order_status。示例运行输出展示了完整的动态转移链客户要求查单 → TriageAgent 移交 OrderStatusAgent → 查单插件被调用 → 客户改口要退货 → OrderStatusAgent 交回 TriageAgent → TriageAgent 再移交 OrderReturnAgent → 退货插件执行。任务结束时出现Handoff-complete_task函数调用携带task_summary随后get()返回任务总结。另有两个变体值得关注step4a_handoff_structured_inputs.py交接编排 结构化输入step4b_handoff_streaming_agent_response_callback.py交接编排 流式响应回调step4c_handoff_mix_agent_types.py混合不同类型的 Agent如ChatCompletionAgent与OpenAIAssistantAgent参与交接。模式五Magentic 编排Magentic 编排形似 GroupChat但由基于规划器的管理器驱动设计受 Microsoft 研究项目 Magentic One一个通用多智能体系统面向复杂任务启发。 step5_magentic.py 组建了两个能力互补的 Agent一个可联网搜索的ResearchAgent使用支持搜索的gpt-4o-search-preview模型一个具备代码解释器执行能力的CoderAgentOpenAIAssistantAgent code interpreter 工具共同完成对比 ResNet-50、BERT-base、GPT-2 的能耗与碳排放这类需要查资料 算数据的任务research_agent ChatCompletionAgent( nameResearchAgent, descriptionA helpful assistant with access to web search. Ask it to perform web searches., instructionsYou are a Researcher. You find information without additional computation or quantitative analysis., serviceOpenAIChatCompletion(ai_model_idgpt-4o-search-preview), ) client OpenAIAssistantAgent.create_client() code_interpreter_tool, code_interpreter_tool_resources OpenAIAssistantAgent.configure_code_interpreter_tool() definition await client.beta.assistants.create( modelOpenAISettings().chat_model_id, nameCoderAgent, descriptionA helpful assistant that writes and executes code to process and analyze data., instructionsYou solve questions using code. Please provide detailed analysis and computation process., toolscode_interpreter_tool, tool_resourcescode_interpreter_tool_resources, ) coder_agent OpenAIAssistantAgent(clientclient, definitiondefinition) magentic_orchestration MagenticOrchestration( membersawait agents(), managerStandardMagenticManager(chat_completion_serviceOpenAIChatCompletion()), agent_response_callbackagent_response_callback, )使用要点代码注释明确说明StandardMagenticManager的提示词经过精心调优但接受自定义提示词以适配高级用户与场景更进阶的需求可继承MagenticManagerBase自行实现管理器逻辑标准管理器要求Chat Completion 模型支持结构化输出。示例输出展示了 ResearchAgent 与 CoderAgent 反复协作检索资料 → 编写代码估算能耗 → 补充区域碳强度数据 → 重新校准 → 最终汇总成含推荐结论的报告表格get()返回最终报告。深入理解编排的底层实现从源码结构看目录 python/semantic_kernel/agents/orchestration/编排层包含以下关键模块文件职责orchestration_base.pyOrchestrationBase抽象基类与OrchestrationResult结果对象agent_actor_base.pyAgentActorBase/ActorBase每个 Agent 在编排中的执行单元concurrent.py并发编排实现sequential.py顺序编排实现SequentialAgentActor链式路由group_chat.py群聊编排、GroupChatManager及BooleanResult/StringResult/MessageResulthandoffs.py交接编排、OrchestrationHandoffs交接图谱定义magentic.pyMagentic 编排与StandardMagenticManagertools.pystructured_outputs_transform等工具函数可以推断出以下设计要点由代码结构佐证Actor 模式编排将每个 Agent 包装为 Actor见AgentActorBase注册到 Runtime通过消息如SequentialRequestMessage、GroupChatStartMessage、HandoffStartMessage驱动执行这与示例日志中Registered agent actor of type ...的 DEBUG 输出一致统一结果模型所有编排的invoke()都返回OrchestrationResult其eventasyncio.Event与cancellation_token共同支撑非阻塞调用、超时等待与取消语义类型泛化OrchestrationBase[TIn, TOut]通过__orig_class__推断类型参数见_set_types()配合input_transform/output_transform实现任意输入输出类型的转换结构化输出变体正是基于此机制。所有编排类目前均标注了experimental装饰器见 feature_stage_decorator 的使用意味着 API 仍处于实验阶段后续版本可能存在调整。可观测性追踪多智能体执行过程目录中的 observability.py 提供了基于 OpenTelemetry 与 Azure Monitor 的可观测性辅助set_up_logging()配置AzureMonitorLogExporter并将日志处理器挂到根 logger同时用过滤器排除semantic_kernel.functions.kernel_plugin等噪声命名空间set_up_tracing()配置AzureMonitorTraceExporter与TracerProviderenable_observability装饰器则会在设置APPINSIGHTS_CONNECTION_STRING环境变量后自动启用日志与追踪并打印当前Trace ID便于把一次完整的多智能体协同过程串联起来排障。小结与进一步学习本文覆盖了 Semantic Kernel Python 多智能体编排的五种核心模式Concurrent适合并行独立分析可配合structured_outputs_transform聚合结构化结果Sequential适合固定流水线支持取消cancel()与流式回调streaming_agent_response_callbackGroupChat提供高度可配置的对话流默认轮询管理器可覆写实现人在环也可替换为 LLM 驱动需结构化输出的管理器Handoff适合动态分诊通过OrchestrationHandoffs声明交接图谱全 Agent 共享用户交互函数可混合不同 Agent 类型Magentic由规划器管理器统筹检索型与执行型Agent适合需要反复查证与计算的复杂任务。所有模式共享同一套创建编排 → 启动 Runtime → invoke → get → stop的调用范式并统一由OrchestrationResult管理异步结果、超时与取消。若想继续深入可阅读 Agent 进阶示例 中的更多 Agent 组合用法以及 Getting Started with Agents 系列 从零搭建的完整入门路径。【免费下载链接】semantic-kernelIntegrate cutting-edge LLM technology quickly and easily into your apps项目地址: https://gitcode.com/GitHub_Trending/se/semantic-kernel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考