尧图网站建设 
swarms AgentRouter 实战指南基于 Embedding 语义相似度的多 Agent 智能路由【免费下载链接】swarmsThe Enterprise-Grade Multi-Agent Orchestration Framework. Website: https://swarms.ai项目地址: https://gitcode.com/GitHub_Trending/swar/swarmsAgentRouter 是 swarms 框架中负责把任务交给最合适的 Agent的核心组件它不再依赖硬编码的 if-else 或关键词规则而是将每个 Agent 的能力描述、系统提示词与任务文本一起投影到向量空间通过余弦相似度计算实现语义级的路由决策。本文以仓库中的 agent_router_example.py 为主线结合 AgentRouter 源码 与其 测试用例带你完整掌握 AgentRouter 的参数、工作流程、动态更新机制与实战用法读完即可在自己的多 Agent 系统中落地智能分诊。一、为什么需要 Agent 路由任务分发的问题本质在多 Agent 系统中一组 Agent 往往各自擅长不同领域医疗分诊、代码审查、文案撰写、数据处理等。如果每次任务都广播给所有 Agent或者固定调用某一个会出现两个问题一是无关 Agent 产生大量无效输出浪费算力与 Token二是难以保证正确的人干正确的事响应质量不稳定。仓库 READMEexamples/multi_agent/agent_router_examples/README.md给出的定位是Agent routing 能够根据能力capabilities、可用性availability或任务特征task characteristics在多个 Agent 之间智能分配任务从而实现高效的负载均衡load balancing与最优 Agent 选择optimal agent selection。而 AgentRouter 的具体实现走的是能力 任务特征的语义匹配路线把 Agent 的身份信息向量化任务到来时再向量化任务文本用相似度打分选出胜者。二、快速上手医学分诊 Agent 路由示例全解仓库中最直接的入门示例是 agent_router_example.py它构建了一个患者症状 → 专科 Agent的路由场景。核心代码如下from swarms.structs.agent import Agent from swarms.structs.agent_router import AgentRouter agent_router AgentRouter( embedding_modeltext-embedding-ada-002, n_agents1, agents[ Agent( agent_nameSymptom Checker, agent_descriptionExpert agent for initial triage and identifying possible causes based on symptom input., system_prompt( You are a medical symptom checker agent. Ask clarifying questions about the patients symptoms, duration, severity, and related risk factors. Provide a list of possible conditions and next diagnostic steps, but do not make a final diagnosis. ), ), Agent( agent_nameDiagnosis Synthesizer, agent_descriptionAgent specializing in synthesizing diagnostic possibilities from patient information and medical history., system_prompt( You are a medical diagnosis assistant. Analyze the patients reported symptoms, medical history, and any test results. Provide a differential diagnosis, and highlight the most likely conditions a physician should consider. ), ), Agent( agent_nameLab Interpretation Expert, agent_descriptionSpecializes in interpreting laboratory and imaging results for diagnostic support., system_prompt( You are a medical lab and imaging interpretation agent. Take the patients test results, imaging findings, and vitals, and interpret them in context of their symptoms. Suggest relevant follow-up diagnostics or considerations for the physician. ), ), ], ) result agent_router.run( I have a headache, fever, and cough. What could be wrong? ) print(result.agent_name)这段代码演示了三条关键信息Agent 的路由身份由三段文本共同决定agent_name名称、agent_description能力描述、system_prompt系统提示词。在 add_agent 的实现 中三者会被拼接成一行文本f{agent.name} {agent.description} {agent.system_prompt}再交给 Embedding 模型生成向量——因此这三段描述写得越精准路由区分度越高。任务进入路由agent_router.run(I have a headache, fever, and cough. What could be wrong?)会把用户描述的症状文本向量化与三个 Agent 的向量逐一计算相似度返回语义上最匹配的那个 Agent 对象随后通过print(result.agent_name)输出命中者的名称。返回的是 Agent 实例而非字符串run()的结果可以直接继续调用result.run(...)执行真正的任务路由与执行无缝衔接这一点在测试用例test_find_best_agent_success中也有体现见 tests/structs/test_agent_router.py。同样的场景还有一个更精简的变体 refactor_smoke_example.py它注册了 Researcher / Coder / Writer 三个 Agent并额外演示了update_agent_history(Coder)与find_best_agent(Help me write a Python function.)的组合用法。三、AgentRouter 核心 API 与参数详解根据 AgentRouter 源码构造函数的签名与参数含义如下参数类型默认值作用embedding_modelstrtext-embedding-ada-002用于生成向量的 Embedding 模型名称。源码注释列举了可选范围text-embedding-ada-002、text-embedding-3-small、text-embedding-3-large、cohere/embed-english-v3.0、huggingface/microsoft/codebert-base等经 LiteLLM 统一调用因此也支持provider/model前缀形式n_agentsint1每次查询期望返回的 Agent 数量。当前版本实现中find_best_agent只返回相似度最高的单个 Agent见下文当前实现说明api_keystr \| NoneNoneEmbedding 服务的 API Key不传时从环境变量读取api_basestr \| NoneNoneEmbedding 服务的自定义 API Base URL自建网关或代理场景agentsList[AgentType] \| NoneNone初始化时一次性注册的 Agent 列表不传则后续通过add_agent/add_agents动态添加其中AgentType在 omni_agent_types.py 中被定义为Union[Agent, Callable, Any]即除了标准Agent实例还允许传入可调用对象或任意对象适配灵活接入场景。核心方法一览add_agent(agent)为单个 Agent 生成 Embedding 并注册。源码 add_agent 同时维护三个内部结构——self.agentsAgent 列表、self.agent_embeddings向量列表、self.agent_metadata{name: ..., text: ...}元数据三者索引一一对应。add_agents(agents)批量注册内部逐个调用add_agent。find_best_agent(task)给定任务文本生成向量后与所有 Agent 向量计算余弦相似度返回得分最高的 Agent路由中没有 Agent 时返回None。run(task)等价于find_best_agent(task)即 run 的实现 直接委托给find_best_agent。update_agent_history(agent_name)把 Agent 的对话历史合并进路由文本并重新向量化详见第五节。值得注意的是add_agent与find_best_agent都使用了tenacity的retry装饰器stop_after_attempt(3)、wait_exponential(multiplier1, min4, max10)即网络抖动时最多重试 3 次、指数退避等待增强了生产环境下的鲁棒性。四、底层原理Embedding 生成与余弦相似度匹配AgentRouter 的智能并不复杂核心是三步流水线第一步文本向量化_generate_embedding。源码 通过litellm.embedding(model..., input[text])调用 Embedding 模型。返回值同时兼容两种响应结构response.data[0].embedding对象属性形式与response.data[0][embedding]字典键值形式并对无法解析的结构抛出ValueError。如果显式传入了api_key/api_base会合并进请求参数。第二步余弦相似度计算_cosine_similarity。实现 是标准的余弦公式两个向量长度不一致时抛出ValueError(Vectors must have the same length)任一向量模长为 0 时返回0.0以避免除零错误。测试用例test_cosine_similarity_identical_vectors/test_cosine_similarity_orthogonal_vectors/test_cosine_similarity_opposite_vectors分别验证了同向相似度 1.0、正交0.0、反向-1.0三种边界见 tests/structs/test_agent_router.py。第三步取最大相似度 Agent。find_best_agent 对任务文本生成向量后与每个已注册 Agent 的向量逐一计算相似度similarities.index(max(similarities))取最高分对应下标返回self.agents[best_index]并把命中分数记入日志如Found best matching agent: xxx (similarity: 0.9123)。当前实现说明虽然构造函数保留了n_agents参数默认 1测试用例test_agent_router_initialization_custom中传入了n_agents3但从源码看find_best_agent目前只返回 Top-1 的单个 Agentn_agents尚未参与 Top-K 筛选。如果你需要一次取回多个候选再人工仲裁可以在此基础上自行扩展排序截断逻辑——这是源码结构similarities全量列表已就绪所支持的合理推断。五、动态更新让路由随 Agent 经验成长AgentRouter 的一个进阶能力是update_agent_historyAgent 每次执行任务后其短期记忆short_memory.return_history_as_string()会沉淀下来把这段历史追加到路由文本中重新生成向量就能让该 Agent 的路由画像包含它真实处理过的任务类型使后续匹配更贴合实际。实现细节 如下agent find_agent_by_name(self.agents, agent_name) history agent.short_memory.return_history_as_string() history_text .join(history) updated_text f{agent.name} {agent.description} {agent.system_prompt} {history_text} # ...用 updated_text 重新生成 embedding 并替换 agent_embeddings[agent_index]这里有两个值得注意的工程细节缓存加速find_agent_by_name来自 ma_blocks.py首次查找时为某个 Agent 列表构建name - agent索引上限 256 条后续查询从 O(n) 线性扫描降为 O(1)。仓库 changelogexamples/changelogs/v13/CHANGELOG.md也提到AgentRouter 的双线性扫描被共享缓存查找替代这一重构正是这条路径。容错目标 Agent 不在列表中时不会抛异常而是记录logger.warning后直接返回——对应测试用例test_update_agent_history_agent_not_found见 tests/structs/test_agent_router.py。六、测试验证从测试用例看路由器的行为边界test_agent_router.py 完整覆盖了 AgentRouter 的行为契约可作为你理解和使用该组件的行为手册初始化test_agent_router_initialization_default验证默认embedding_modeltext-embedding-ada-002、n_agents1、agents[]test_agent_router_initialization_custom验证自定义参数与批量传 Agent。匹配正确性test_find_best_agent_success构造数据抽取与摘要两个 Agent输入Extract data from documents能命中数据抽取 Agenttest_find_best_agent_no_agents验证空路由返回None。边界条件test_agent_router_edge_cases验证空字符串任务返回None超长任务1000 个test 也能被优雅处理而不崩溃。与流式执行的集成test_router_with_agent_streaming与test_router_find_and_run_with_streaming验证了路由命中的 Agent 可以直接带streaming_callback执行并流式返回这一链路说明路由结果与 Agent 的run接口完全兼容。这些测试同时印证了路由命中基于语义相似度而非关键词包含因此即使任务表述与 Agent 描述用词不完全一致只要语义相关仍能正确分派。七、实战建议与注意事项结合示例与源码落地 AgentRouter 时有几点实操建议认真写agent_description它和system_prompt共同决定路由向量。描述应突出擅长什么、处理什么类型输入避免含糊表述命名用语义化名称如Symptom Checker有助于日志可读性。按需指定 Embedding 模型默认text-embedding-ada-002对英文文本友好如需中文场景可切换为text-embedding-3-small或自建模型如huggingface/microsoft/codebert-base适合代码场景并注意环境变量中的 API Key 配置。先分诊、再执行推荐的调用模式是best router.find_best_agent(task)然后best.run(task, streaming_callback...)或把best交给上层编排。这样路由失败返回None时还可以走兜底逻辑。善用update_agent_history对高频 Agent 在任务结束后更新其历史向量能让路由画像持续贴近真实负载但注意它每次都会触发一次新的 Embedding 调用需权衡成本。理解能力边界当前实现的find_best_agent只返回 Top-1n_agents参数暂不参与 Top-K 筛选若需要多候选路由、负载均衡或按可用性过滤可基于现有similarities数组做二次扩展或参考仓库中MultiAgentRouter见 multi_agent_router.py等其他路由实现的设计思路。总而言之AgentRouter 用一套文本 → 向量 → 余弦相似度 → 取最优的简洁链路为 swarms 多 Agent 系统提供了开箱即用的语义路由能力。从 agent_router_example.py 起步对照 源码 与 测试 理解其行为边界你就能在自己的多 Agent 架构中快速构建出自动分诊、按需调度的任务分发层。【免费下载链接】swarmsThe Enterprise-Grade Multi-Agent Orchestration Framework. Website: https://swarms.ai项目地址: https://gitcode.com/GitHub_Trending/swar/swarms创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 swarms AgentRouter 实战指南：基于 Embedding 语义相似度的多 Agent 智能路由
swarms AgentRouter 实战指南基于 Embedding 语义相似度的多 Agent 智能路由【免费下载链接】swarmsThe Enterprise-Grade Multi-Agent Orchestration Framework. Website: https://swarms.ai项目地址: https://gitcode.com/GitHub_Trending/swar/swarmsAgentRouter 是 swarms 框架中负责把任务交给最合适的 Agent的核心组件它不再依赖硬编码的 if-else 或关键词规则而是将每个 Agent 的能力描述、系统提示词与任务文本一起投影到向量空间通过余弦相似度计算实现语义级的路由决策。本文以仓库中的 agent_router_example.py 为主线结合 AgentRouter 源码 与其 测试用例带你完整掌握 AgentRouter 的参数、工作流程、动态更新机制与实战用法读完即可在自己的多 Agent 系统中落地智能分诊。一、为什么需要 Agent 路由任务分发的问题本质在多 Agent 系统中一组 Agent 往往各自擅长不同领域医疗分诊、代码审查、文案撰写、数据处理等。如果每次任务都广播给所有 Agent或者固定调用某一个会出现两个问题一是无关 Agent 产生大量无效输出浪费算力与 Token二是难以保证正确的人干正确的事响应质量不稳定。仓库 READMEexamples/multi_agent/agent_router_examples/README.md给出的定位是Agent routing 能够根据能力capabilities、可用性availability或任务特征task characteristics在多个 Agent 之间智能分配任务从而实现高效的负载均衡load balancing与最优 Agent 选择optimal agent selection。而 AgentRouter 的具体实现走的是能力 任务特征的语义匹配路线把 Agent 的身份信息向量化任务到来时再向量化任务文本用相似度打分选出胜者。二、快速上手医学分诊 Agent 路由示例全解仓库中最直接的入门示例是 agent_router_example.py它构建了一个患者症状 → 专科 Agent的路由场景。核心代码如下from swarms.structs.agent import Agent from swarms.structs.agent_router import AgentRouter agent_router AgentRouter( embedding_modeltext-embedding-ada-002, n_agents1, agents[ Agent( agent_nameSymptom Checker, agent_descriptionExpert agent for initial triage and identifying possible causes based on symptom input., system_prompt( You are a medical symptom checker agent. Ask clarifying questions about the patients symptoms, duration, severity, and related risk factors. Provide a list of possible conditions and next diagnostic steps, but do not make a final diagnosis. ), ), Agent( agent_nameDiagnosis Synthesizer, agent_descriptionAgent specializing in synthesizing diagnostic possibilities from patient information and medical history., system_prompt( You are a medical diagnosis assistant. Analyze the patients reported symptoms, medical history, and any test results. Provide a differential diagnosis, and highlight the most likely conditions a physician should consider. ), ), Agent( agent_nameLab Interpretation Expert, agent_descriptionSpecializes in interpreting laboratory and imaging results for diagnostic support., system_prompt( You are a medical lab and imaging interpretation agent. Take the patients test results, imaging findings, and vitals, and interpret them in context of their symptoms. Suggest relevant follow-up diagnostics or considerations for the physician. ), ), ], ) result agent_router.run( I have a headache, fever, and cough. What could be wrong? ) print(result.agent_name)这段代码演示了三条关键信息Agent 的路由身份由三段文本共同决定agent_name名称、agent_description能力描述、system_prompt系统提示词。在 add_agent 的实现 中三者会被拼接成一行文本f{agent.name} {agent.description} {agent.system_prompt}再交给 Embedding 模型生成向量——因此这三段描述写得越精准路由区分度越高。任务进入路由agent_router.run(I have a headache, fever, and cough. What could be wrong?)会把用户描述的症状文本向量化与三个 Agent 的向量逐一计算相似度返回语义上最匹配的那个 Agent 对象随后通过print(result.agent_name)输出命中者的名称。返回的是 Agent 实例而非字符串run()的结果可以直接继续调用result.run(...)执行真正的任务路由与执行无缝衔接这一点在测试用例test_find_best_agent_success中也有体现见 tests/structs/test_agent_router.py。同样的场景还有一个更精简的变体 refactor_smoke_example.py它注册了 Researcher / Coder / Writer 三个 Agent并额外演示了update_agent_history(Coder)与find_best_agent(Help me write a Python function.)的组合用法。三、AgentRouter 核心 API 与参数详解根据 AgentRouter 源码构造函数的签名与参数含义如下参数类型默认值作用embedding_modelstrtext-embedding-ada-002用于生成向量的 Embedding 模型名称。源码注释列举了可选范围text-embedding-ada-002、text-embedding-3-small、text-embedding-3-large、cohere/embed-english-v3.0、huggingface/microsoft/codebert-base等经 LiteLLM 统一调用因此也支持provider/model前缀形式n_agentsint1每次查询期望返回的 Agent 数量。当前版本实现中find_best_agent只返回相似度最高的单个 Agent见下文当前实现说明api_keystr \| NoneNoneEmbedding 服务的 API Key不传时从环境变量读取api_basestr \| NoneNoneEmbedding 服务的自定义 API Base URL自建网关或代理场景agentsList[AgentType] \| NoneNone初始化时一次性注册的 Agent 列表不传则后续通过add_agent/add_agents动态添加其中AgentType在 omni_agent_types.py 中被定义为Union[Agent, Callable, Any]即除了标准Agent实例还允许传入可调用对象或任意对象适配灵活接入场景。核心方法一览add_agent(agent)为单个 Agent 生成 Embedding 并注册。源码 add_agent 同时维护三个内部结构——self.agentsAgent 列表、self.agent_embeddings向量列表、self.agent_metadata{name: ..., text: ...}元数据三者索引一一对应。add_agents(agents)批量注册内部逐个调用add_agent。find_best_agent(task)给定任务文本生成向量后与所有 Agent 向量计算余弦相似度返回得分最高的 Agent路由中没有 Agent 时返回None。run(task)等价于find_best_agent(task)即 run 的实现 直接委托给find_best_agent。update_agent_history(agent_name)把 Agent 的对话历史合并进路由文本并重新向量化详见第五节。值得注意的是add_agent与find_best_agent都使用了tenacity的retry装饰器stop_after_attempt(3)、wait_exponential(multiplier1, min4, max10)即网络抖动时最多重试 3 次、指数退避等待增强了生产环境下的鲁棒性。四、底层原理Embedding 生成与余弦相似度匹配AgentRouter 的智能并不复杂核心是三步流水线第一步文本向量化_generate_embedding。源码 通过litellm.embedding(model..., input[text])调用 Embedding 模型。返回值同时兼容两种响应结构response.data[0].embedding对象属性形式与response.data[0][embedding]字典键值形式并对无法解析的结构抛出ValueError。如果显式传入了api_key/api_base会合并进请求参数。第二步余弦相似度计算_cosine_similarity。实现 是标准的余弦公式两个向量长度不一致时抛出ValueError(Vectors must have the same length)任一向量模长为 0 时返回0.0以避免除零错误。测试用例test_cosine_similarity_identical_vectors/test_cosine_similarity_orthogonal_vectors/test_cosine_similarity_opposite_vectors分别验证了同向相似度 1.0、正交0.0、反向-1.0三种边界见 tests/structs/test_agent_router.py。第三步取最大相似度 Agent。find_best_agent 对任务文本生成向量后与每个已注册 Agent 的向量逐一计算相似度similarities.index(max(similarities))取最高分对应下标返回self.agents[best_index]并把命中分数记入日志如Found best matching agent: xxx (similarity: 0.9123)。当前实现说明虽然构造函数保留了n_agents参数默认 1测试用例test_agent_router_initialization_custom中传入了n_agents3但从源码看find_best_agent目前只返回 Top-1 的单个 Agentn_agents尚未参与 Top-K 筛选。如果你需要一次取回多个候选再人工仲裁可以在此基础上自行扩展排序截断逻辑——这是源码结构similarities全量列表已就绪所支持的合理推断。五、动态更新让路由随 Agent 经验成长AgentRouter 的一个进阶能力是update_agent_historyAgent 每次执行任务后其短期记忆short_memory.return_history_as_string()会沉淀下来把这段历史追加到路由文本中重新生成向量就能让该 Agent 的路由画像包含它真实处理过的任务类型使后续匹配更贴合实际。实现细节 如下agent find_agent_by_name(self.agents, agent_name) history agent.short_memory.return_history_as_string() history_text .join(history) updated_text f{agent.name} {agent.description} {agent.system_prompt} {history_text} # ...用 updated_text 重新生成 embedding 并替换 agent_embeddings[agent_index]这里有两个值得注意的工程细节缓存加速find_agent_by_name来自 ma_blocks.py首次查找时为某个 Agent 列表构建name - agent索引上限 256 条后续查询从 O(n) 线性扫描降为 O(1)。仓库 changelogexamples/changelogs/v13/CHANGELOG.md也提到AgentRouter 的双线性扫描被共享缓存查找替代这一重构正是这条路径。容错目标 Agent 不在列表中时不会抛异常而是记录logger.warning后直接返回——对应测试用例test_update_agent_history_agent_not_found见 tests/structs/test_agent_router.py。六、测试验证从测试用例看路由器的行为边界test_agent_router.py 完整覆盖了 AgentRouter 的行为契约可作为你理解和使用该组件的行为手册初始化test_agent_router_initialization_default验证默认embedding_modeltext-embedding-ada-002、n_agents1、agents[]test_agent_router_initialization_custom验证自定义参数与批量传 Agent。匹配正确性test_find_best_agent_success构造数据抽取与摘要两个 Agent输入Extract data from documents能命中数据抽取 Agenttest_find_best_agent_no_agents验证空路由返回None。边界条件test_agent_router_edge_cases验证空字符串任务返回None超长任务1000 个test 也能被优雅处理而不崩溃。与流式执行的集成test_router_with_agent_streaming与test_router_find_and_run_with_streaming验证了路由命中的 Agent 可以直接带streaming_callback执行并流式返回这一链路说明路由结果与 Agent 的run接口完全兼容。这些测试同时印证了路由命中基于语义相似度而非关键词包含因此即使任务表述与 Agent 描述用词不完全一致只要语义相关仍能正确分派。七、实战建议与注意事项结合示例与源码落地 AgentRouter 时有几点实操建议认真写agent_description它和system_prompt共同决定路由向量。描述应突出擅长什么、处理什么类型输入避免含糊表述命名用语义化名称如Symptom Checker有助于日志可读性。按需指定 Embedding 模型默认text-embedding-ada-002对英文文本友好如需中文场景可切换为text-embedding-3-small或自建模型如huggingface/microsoft/codebert-base适合代码场景并注意环境变量中的 API Key 配置。先分诊、再执行推荐的调用模式是best router.find_best_agent(task)然后best.run(task, streaming_callback...)或把best交给上层编排。这样路由失败返回None时还可以走兜底逻辑。善用update_agent_history对高频 Agent 在任务结束后更新其历史向量能让路由画像持续贴近真实负载但注意它每次都会触发一次新的 Embedding 调用需权衡成本。理解能力边界当前实现的find_best_agent只返回 Top-1n_agents参数暂不参与 Top-K 筛选若需要多候选路由、负载均衡或按可用性过滤可基于现有similarities数组做二次扩展或参考仓库中MultiAgentRouter见 multi_agent_router.py等其他路由实现的设计思路。总而言之AgentRouter 用一套文本 → 向量 → 余弦相似度 → 取最优的简洁链路为 swarms 多 Agent 系统提供了开箱即用的语义路由能力。从 agent_router_example.py 起步对照 源码 与 测试 理解其行为边界你就能在自己的多 Agent 架构中快速构建出自动分诊、按需调度的任务分发层。【免费下载链接】swarmsThe Enterprise-Grade Multi-Agent Orchestration Framework. Website: https://swarms.ai项目地址: https://gitcode.com/GitHub_Trending/swar/swarms创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考