AutoGen 分布式运行时与语义路由实战解析基于 core_semantic_router 示例的多 Agent 编排与意图路由【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogenAutoGen 仓库中的 core_semantic_router 示例 演示了一套**分布式 Agent 运行时distributed agent runtime**的核心用法由一个独立的 Host 进程维护事件引擎与发布/订阅消息系统由 Worker 进程承载 Agent 生命周期并托管一个语义路由器semantic router再由 User Proxy 承接用户界面与交互。本文以该示例为骨架先讲清三大组件与一个典型企业场景HR/Finance 意图路由再给出两终端启动方式、关键词交互与会话结束规则最后深入到源码层说明意图分类器Intent Classifier、Agent 注册表Agent Registry、Topic 发布/订阅机制以及 Closure Agent 消息回路的实现细节帮助你掌握 AutoGen 中把 Agent 按职责拆到不同进程、通过消息主题解耦的编排范式。示例定位三层分布式运行时架构原文档明确指出本系统由三个主要组件构成它们职责分离且运行在不同的进程中因而可以部署在不同机器上Agent Host RuntimeAgent 宿主运行时负责管理事件引擎eventing engine与发布/订阅消息系统。Worker RuntimeWorker 运行时负责分布式 Agent 的生命周期管理其中就包括本示例的主角语义路由器semantic router。User Proxy用户代理负责管理用户界面以及用户与 Agent 之间的交互。对应到仓库文件三者各自落在独立的入口中组件运行入口关键类源码位置Agent Host Runtimerun_host.pyGrpcWorkerAgentRuntimeHost_worker_runtime_host.pyWorker Runtime含语义路由器run_semantic_router.pyGrpcWorkerAgentRuntime_worker_runtime.pyUser Proxy内嵌于 Worker 进程的UserProxyAgent_agents.pyHost 与 Worker 之间通过 gRPC 通信默认服务地址为localhost:50051Host 侧GrpcWorkerAgentRuntimeHost(addresslocalhost:50051)创建宿主服务host.start()在后台启动该服务host.stop_when_signal()等待进程信号后优雅停机参见 run_host.pyWorker 侧GrpcWorkerAgentRuntime(host_addresslocalhost:50051)连接到宿主await agent_runtime.start()建立通道后即可在运行时上注册 Agent 与订阅参见 run_semantic_router.py。需要特别说明的是虽然把多个 Agent 放在一个系统里协作解决复杂任务是多 Agent 框架的常见用法但本示例的目标场景与之不同它展示的是企业如何管理大量各自面向独立任务的 Agent并把用户路由到与当前任务最相关的那个 Agent 上。示例中 Agent 的业务逻辑刻意保持简单意在说明这种路由能力由 AutoGen 的分布式运行时提供与 Agent 自身能力强弱无关。示例场景HR / Finance 双 Agent 的意图路由本示例虚构了一个企业人力资源与财务运营场景存在两个相互独立、互不依赖的分布式 Agent一个HR Agent处理hr、human resources、employee等关键词相关的请求一个Finance Agent处理finance、money、budget等关键词相关的请求。当用户发起一轮会话时Semantic Router Agent会先识别用户消息的意图intent再找到与该意图最匹配的 Agent然后把用户路由给该 Agent随后由该 Agent 接管会话与用户以对话方式持续交互。这里需要澄清原文档隐含的一个概念层次所谓语义路由在代码上被抽象成了两个独立且可替换的环节见 _semantic_router_components.py 中定义的抽象基类意图分类器Intent Classifier定义抽象接口IntentClassifierBase.classify_intent(message) - str负责把一条用户消息归类为某个意图标识Agent 注册表Agent Registry定义抽象接口AgentRegistryBase.get_agent(intent) - str负责把意图标识映射为具体的 Agent 类型名。原文档指出当前示例使用过度简单的字符串匹配方式识别意图。这一说法在示例代码中可以直接印证MockIntentClassifier仅对消息做子串包含判断命中任一关键词即返回对应意图全部未命中返回generalMockAgentRegistry则用一个dict维护意图 → Agent 名称的映射见 run_semantic_router.py。同时源码注释给出了一条明显的演进路径更复杂的实现可以让意图分类器更健壮注册表可以基于 Azure AI Search 之类的技术托管大量 Agent 的定义见 run_semantic_router.py 的模块 docstring此处仅作为示例作者对扩展方向的设想陈述。关键词与意图的对应关系完整如下意图intent触发关键词路由目标 Agentfinance_intentfinance、money、budgetfinancehr_intenthr、human resources、employeehrgeneral未命中任何关键词—无可路由 Agent触发TerminationMessage结束会话运行准备与启动步骤环境安装原文档的Getting Started一节说明需要安装autogen-core及其依赖。结合源码进一步核实示例同时导入了autogen_core核心 Agent/Topic/消息 API与autogen_ext.runtimes.grpcgRPC 宿主与 Worker 运行时后者位于autogen-ext包中且 gRPC 运行时是autogen-ext的可选依赖声明于 python/packages/autogen-ext/pyproject.toml 的[project.optional-dependencies]段grpcextra 引入grpcio~1.70.0等。因此完整的依赖安装应包括# 在仓库 python/ 目录下该目录以 uv 管理多包工作区见 python/pyproject.toml 与 python/uv.lock # 方式一使用 uv 安装本示例所需的包 uv pip install -e packages/autogen-core uv pip install -e packages/autogen-ext[grpc] # 方式二也可使用 pip 以可编辑模式安装需保证 python 环境一致 pip install -e packages/autogen-core pip install -e packages/autogen-ext[grpc]安装完成后请在示例目录下运行代码下文命令均默认在python/samples/core_semantic_router目录中执行因为示例文件间使用同目录相对导入如from _agents import ...。在两个终端中启动由于本示例意在演示分布式运行时其各组件被设计为运行在不同进程即不同终端中# 终端 1运行 Agent Host Runtime python run_host.py# 终端 2运行 Worker Runtime语义路由器及全部 Worker Agent 都注册在这里 python run_semantic_router.py运行后第一个终端会记录一系列事件日志——各 Agent 注册到运行时的过程run_host.py以logging.DEBUG级别输出事件日志来自TRACE_LOGGER_NAME.host记录器。在第二个终端中程序打印Agents registered, starting conversation后提示Enter a message:此时即可输入与 finance 或 hr 场景相关的请求例如Finance 场景I need a budget report命中budgetHR 场景I want to talk to HR about my employee benefits命中hr/employee。随后你会在终端 1 和终端 2 上看到 Host 与 Worker 运行时来回收发消息、把请求路由到正确 Agent 的过程最终由终端 2 打印出 Agent 的响应。会话随后可以在被选中的 Agent 上持续进行直到用户发送一条包含END的消息——此时该 Agent 会与用户断开一轮新会话可以重新开始。消息流与 Topic 发布/订阅机制原文档说明借助 Agent Host Runtime 的Topic特性本系统的工作流程如下mermaid 序列图。对这张序列图结合 run_semantic_router.py、_agents.py 与 _semantic_router_agent.py 可以还原出消息在各 Topic 上的真实流动。图中所有消息都通过publish_message(...)发布到某个DefaultTopicId并由相应 Agent 的订阅接收。示例实际使用了以下 Topic 类型Topic 类型谁发布谁订阅用途default顶层启动代码 / Closure Agent新会话SemanticRouterAgentdefault_subscription接收每条待路由的初始用户消息finance/hrSemanticRouterAgent.contact_agent对应WorkerAgent路由后的用户消息送达目标 Agentuser_proxyWorkerAgent、SemanticRouterAgentUserProxyAgentAgent 回复或终止事件回传给 User ProxyresponseUserProxyAgentClosureAgent把结果暴露给外部系统如 API以返回给用户初始消息如何进入系统顶层代码通过await agent_runtime.publish_message(UserProxyMessage(contentmessage, sourceuser), topic_idDefaultTopicId(typedefault, sourceuser))把第一条消息发布到defaultTopic见 run_semantic_router.py。SemanticRouterAgent上标注了default_subscription装饰器见 _semantic_router_agent.py该装饰器在注册时自动为 Agent 建立对defaultTopic 类型的订阅其实现见 _default_subscription.py从而保证路由器能收到所有待路由消息。路由器如何决定发给谁SemanticRouterAgent.route_to_agent处理器_semantic_router_agent.py执行三步从消息上下文中取出session_id ctx.topic_id.source作为该轮会话的分隔标识本示例中为user调用_identify_intent即把消息交给IntentClassifierBase实现完成意图识别调用_find_agent即通过AgentRegistryBase实现把意图映射成 Agent 名称。_find_agent中有一个值得注意的容错分支当注册表查询抛出KeyError意图不在映射表中时会返回哨兵值termination。随后contact_agent_semantic_router_agent.py据此分流若是termination向user_proxyTopic 发布TerminationMessage(reasonNo relevant agent found, ...)即示例源码 docstring 中所说的无法归类为 hr 或 finance 的请求会以一条终止消息结束对话否则把原UserProxyMessage转发到DefaultTopicId(type目标Agent, sourcesession_id)例如类型为finance或hr的 Topic。WorkerAgent 在注册时即通过agent_runtime.add_subscription(DefaultSubscription(topic_typefinance, agent_typefinance))建立了Topic 类型 → Agent 类型的订阅见 run_semantic_router.py因此消息到达对应 Topic 后会被对应的 Worker 实例接收。会话中继与后续消息为何直达 Worker在序列图中可以看到初次路由后后续消息是User 直接发给 Worker_Agent的中间不再经过 Semantic_Router。其机制如下WorkerAgent 收到UserProxyMessage后把回复封装为WorkerAgentMessage发布到user_proxyTopicsource取ctx.topic_id.type即finance/hr用于表明回复来自哪个领域UserProxyAgent.on_agent_message收到该消息后不做改写直接把它转发到responseTopic见 _agents.pyClosureAgent订阅responseTopic。它由ClosureAgent.register_closure(..., output_result, subscriptionslambda: [DefaultSubscription(topic_typeresponse, agent_typeclosure_agent)])注册见 run_semantic_router.py其回调output_result负责把 Agent 回复打印给用户随即再次以input()获取用户下一条输入用户下一句回复由output_result发布到DefaultTopicId(typemessage.source, sourceuser)——由于message.source此时就是finance或hr这条消息直接进入对应 Worker 的 Topic从而跳过语义路由器实现用户与所选 Agent 的持续对话。Closure Agent 是示例中面向外部的会话出口源码注释明确指出它把最终结果暴露给外部系统例如一个 API使系统得以与用户交互。其运行时抽象ClosureAgent与编程接口ClosureContext含send_message与publish_message定义于 _closure_agent.py注册入口register_closure位于同一文件的 第 142 行。会话如何结束与重启当用户在后续消息中发送包含END的内容时WorkerAgent 的消息处理器会发布TerminationMessage(reasonuser terminated conversation, content..., sourceself.type)到user_proxyTopic见 _agents.py。同样如果语义路由器发现无法路由意图也会走user_proxyTopic 的TerminationMessage路径。两条路径殊途同归UserProxyAgent.on_terminate处理器收到TerminationMessage将其封装为FinalResult发布到responseTopic_agents.pyClosureAgent.output_result根据消息类型分流收到WorkerAgentMessage时继续同 Agent 会话收到FinalResult时打印Conversation ended并提示Enter a new conversation start:见 run_semantic_router.py新一轮会话的开始消息被发布回defaultTopic重新进入语义路由器的管辖范围——一条用户消息发送END即可断开当前 Agent开启新会话。分层消息类型设计整个示例在 _semantic_router_components.py 中定义了清晰的消息类型体系均以source消息来源为基座消息类型继承关系额外字段语义TextMessageBaseMessagecontent带文本内容的通用消息UserProxyMessageTextMessage—用户发往系统、等待路由的消息WorkerAgentMessageTextMessage—Worker Agent 发回给用户的回复TerminationMessageTextMessagereason系统通知用户会话结束并携带结束原因FinalResultTextMessage—Agent 最终输出标志一轮会话的收尾这一继承层次与 AutoGen Core 的消息类型驱动路由机制相配合每个message_handler通过参数类型标注声明自己能处理哪种消息例如SemanticRouterAgent只处理UserProxyMessageUserProxyAgent分别用on_terminate与on_agent_message处理TerminationMessage与WorkerAgentMessage。本示例可借鉴的工程要点综合原文档与源码core_semantic_router在工程上至少有四点值得复用用进程边界解耦系统角色Host事件引擎 pub/sub 总线与 WorkerAgent 生命周期分属不同进程天然支持把大规模 Agent 集合分散到不同机器符合企业维护大量任务型 Agent的运营形态把路由策略与Agent 逻辑彻底分离意图分类与 Agent 查找都通过抽象基类注入示例用字符串匹配只是占位实现想要替换成 LLM 意图识别或基于向量检索的 Agent 注册表只需要提供新的IntentClassifierBase/AgentRegistryBase实现路由 Agent 与 Worker 代码无需改动用 Topic 而非点对点调用来织网default待路由、finance/hr领域、user_proxy会话中枢、response对外出口四类 Topic 各司其职消息发布者与接收者完全解耦新增一个领域的 Agent 只需注册 建订阅 往注册表加一行会话维度由 Topic 的source承载示例统一使用sourceuser作为会话标识传递session_id保证了同一轮会话内路由、回复、转发都落在同一命名空间为将来扩展多用户并发会话预留了结构上的空间。如果想要在更大规模上观察同一种分布式运行时的用法可以继续对照仓库中的 core_distributed-group-chat分布式群聊与 core_grpc_worker_runtime多 Worker gRPC 运行时示例它们与core_semantic_router共用GrpcWorkerAgentRuntime/GrpcWorkerAgentRuntimeHost这套基础设施。最后说明本文分析所基于的示例由仓库原 README 维护Contributors 信息见原文档结尾。以上所有关于类的职责与消息流向的结论均可回到本示例自身的六个源文件以及autogen-core、autogen-ext的源码中逐行核对。【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogen创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
