使用 Sprix SAGE Router 路由 AI 智能体任务

什么是 Sprix SAGE Router？

Sprix SAGE Router 是一个开源 Python 研究项目，用于在任务已经开始执行后，决定 AI 智能体应如何继续完成任务。SAGE 是 State-Aware Graph Exchange（状态感知图交换）的缩写。它位于智能体发现与任务执行之间，在选择可行的智能体配置时，会考虑当前进度、能力、权限、成本、延迟、依赖关系和上下文传输损失。

该路由器会比较三种执行模式：

• SELF：当现任智能体的能力和已积累上下文足够时，由其独自继续执行。
• COLLABORATE：现任智能体保留任务所有权，由互补智能体补足缺失的需求。
• HANDOFF：当某个对等智能体的专业优势超过上下文传输成本时，由其接管全部任务所有权。

SAGE 旨在构建于 Agent2Agent 协议之上。A2A 定义了可互操作的 Agent Card、任务、消息、产物、身份验证和传输机制。SAGE 解决的是另一个问题：应由哪种符合条件的配置、以何种模式执行任务，以及为何如此选择？

当前原型只返回路由决策，并不会向智能体传输任务，因此必须由 A2A 客户端或其他执行层实施所选路线。

核心路由功能

SAGE 将路由、调度和在线学习整合到一个可检查的决策流程中。其主要功能包括：

• 三模式路由：SELF、COLLABORATE 和 HANDOFF 候选方案基于同一效用目标进行竞争。
• 进度感知重规划：已积累进度、活跃执行者、失败记录、已完成的任务节点和可传输上下文，都会影响切换是否值得。
• 互补团队选择：团队因增加需求覆盖率而获得更高评价，而不是仅仅汇集功能重复的高排名智能体。
• 上下文信任：系统会针对每个智能体和每项需求学习可靠性，而不是将其简化为单一的全局信誉分数。
• 任务 DAG 调度：需求会分配给执行者，而依赖边则决定执行顺序、通信拓扑和关键路径延迟。
• 有界搜索：束搜索会评估多个协作团队前缀，而不是仅做一次贪心选择。
• 约束优先过滤：无论预测质量多高，不符合条件的智能体都会在排名前被移除。
• 证据感知学习：需求评分、实际成本和实际延迟可用于更新信任度、协同效应、成功模型和报价保真度。
• 可审计结果：决策会公开所选模式、任务分配、拓扑、预测成功率、覆盖率、成本、延迟、风险、效用和选择理由。

算法如何评估路线

对于每项任务需求，SAGE 会将全局信任与按需求设定条件的信任相结合，以估算智能体经过校准的能力。当成员提供有用且不重复的能力时，团队覆盖率就会提高。随后，每项剩余需求都会分配给团队中经过校准后能力最强的成员。

调度器会遵循需求 DAG。分配给同一智能体的工作将串行执行，而分配给不同智能体的独立工作则可并行执行。构建候选团队后，SAGE 会根据任务预算和截止时间，检查团队层面的成本与关键路径延迟。

系统使用效用函数对可行候选方案进行排名，在预测成功价值与成本、延迟、风险、上下文传输损失、协调开销和不确定性之间取得平衡。因此，即使某个专家能力更强，如果任务交接会丢失过多有用上下文，其排名仍可能低于现任智能体。同样，一个能力互补的团队也可能胜过由知名但能力重复的智能体组成的团队。

安装并验证项目

前提条件

参考实现要求使用 Python 3.10 或更高版本。根据项目 README，该项目没有运行时依赖项。

克隆代码库

git clone https://github.com/wang2122/sprix-sage-router.git
cd sprix-sage-router
python demo.py

demo.py 脚本提供了一个易于理解的端到端示例。由于代码库通过 sprix_sage.py 直接公开其实现，因此快速入门不需要额外执行软件包安装命令。

运行测试和基准测试

python -m unittest -v
python benchmark.py

单元测试用于验证路由器是否符合预期行为。基准测试会使用确定性随机种子运行外部非线性模拟器，以便将 SAGE 的静态版本和在线版本与更简单的路由策略进行比较。

创建第一个路由场景

第 1 步：导入核心类型

from sprix_sage import (
    Agent,
    ExecutionOutcome,
    Requirement,
    SAGERouter,
    Task,
)

此示例使用 Agent 描述候选智能体，使用 Requirement 和 Task 定义工作，使用 SAGERouter 选择路线，并使用 ExecutionOutcome 在执行后返回证据。

第 2 步：定义可用智能体

agents = [
    Agent(
        "planner",
        {"planning": 0.92, "coding": 0.55},
        cost=0.08,
        latency_ms=900,
    ),
    Agent(
        "coder",
        {"planning": 0.35, "coding": 0.96},
        cost=0.12,
        latency_ms=1200,
    ),
]

每个智能体都有标识符、能力向量、成本估算和延迟估算。在这里，planner 擅长规划，而 coder 更擅长编码。这种差异让 SAGE 有理由考虑协作或交接，而不是仅根据一个综合分数对智能体进行排名。

第 3 步：对需求和依赖关系建模

task = Task(
    "build-feature",
    requirements=(
        Requirement("planning", 0.4),
        Requirement(
            "coding",
            0.6,
            depends_on=("planning",),
        ),
    ),
    value=1.0,
    budget=0.30,
    deadline_ms=4000,
    progress=0.35,
)

该任务包含两项加权需求。编码需求依赖于规划，从而形成一个规划必须先于编码完成的小型 DAG。该任务还提供了约束路由所需的值：

• value 表示成功完成任务的价值。
• budget 限制可接受的成本。
• deadline_ms 限制调度延迟。
• progress 告知路由器任务已经完成 35%。

进度非常重要，因为在执行过程中变更任务所有者可能会造成上下文传输损失。因此，专家的能力优势并不会自动使 HANDOFF 成为最佳路线。

第 4 步：路由任务

router = SAGERouter(
    agents,
    incumbent_id="planner",
)

decision = router.route(task)

print(decision.mode)
print(decision.assignments)
print(decision.topology)

现任智能体是当前拥有该任务的智能体。调用 route 会比较可行的 SELF、COLLABORATE 和 HANDOFF 候选方案。可以通过以下字段检查返回的决策：

• decision.mode 表示所选执行模式。
• decision.assignments 表示需求到智能体的角色分配。
• decision.topology 表示由依赖关系形成的通信结构。

不要假设这个特定场景必然会产生某一种固定模式。结果取决于路由器经过校准的估算、约束条件、进度和已学习状态。

从执行结果中学习

路由只是工作流程的前半部分。所选智能体执行任务后，应将现有证据中可信度最高的信息返回给路由器：

router.record_outcome(
    decision,
    ExecutionOutcome(
        success=0.9,
        requirement_scores={
            "planning": 0.95,
            "coding": 0.86,
        },
        actual_cost=0.19,
        actual_latency_ms=1450,
    ),
)

总体成功值为路由器提供任务级结果。针对具体需求的评分可实现更精确的贡献归因，避免每个团队成员获得完全相同的评价。实际成本和延迟有助于校准未来预测及提供方报价。

反复执行路由并提供反馈，可让在线 SAGE 持续优化上下文信任、智能体配对效应、成功率预测和报价保真度。为实现有效学习，请持续记录结果，并在有实测证据可用时优先使用实测证据，而不是粗略的成功标签。

将 SAGE 与 A2A 系统集成

生产环境集成可将协议和市场信号转换为 SAGE 的内部表示：

• 将 AgentCard.skills 规范化为能力向量。
• 将安全要求用作严格的权限过滤条件。
• 在评分前检查支持的输入和输出模式是否兼容。
• 将任务状态、产物、已完成的 DAG 节点和失败记录映射到当前执行状态与传输损失估算。
• 将提供方报价表示为包含成本、延迟和置信度的竞价信息。
• 使用已完成任务的评估结果更新上下文信任、配对残差、成功率预测和报价保真度。

SAGE 返回决策后，A2A 客户端可以通过发送消息、流式传输、任务轮询或取消等协议操作执行该决策。应明确保持二者的边界：SAGE 负责选择并解释路线，而 A2A 层负责通信和执行。

高级使用技巧

将任务建模为有意义的 DAG

使用依赖关系表示真实的执行顺序，而不是将所有需求放在一个扁平列表中。准确的依赖关系可让 SAGE 区分串行工作与并行工作，并生成更有用的关键路径延迟估算。

优先使用互补的能力描述

使用需求级能力描述智能体。单一的宽泛分数会掩盖规划、编码、研究或其他技能之间的区别，并削弱上下文信任与角色分配的有效性。

将权限视为硬性约束

不要使用预测质量绕过访问规则。SAGE 的权限优先设计要求在候选方案排名前排除不符合条件的智能体。兼容性、身份验证和策略检查也应在执行前完成。

使用符合实际的预算和截止时间

SAGE 会在构建候选方案期间以及团队层面再次检查约束条件。请提供针对具体任务的预算和截止时间，避免选中某些单独来看很有吸引力，但组合成本或关键路径并不可行的智能体集合。

检查完整决策，而非只关注模式

检查任务分配、拓扑、覆盖率、预测成功率、成本、延迟、风险、效用和选择理由。审计记录可以揭示某条路线是由于真正的能力互补、保留现任智能体的上下文、较低的传输损失，还是约束可行性而被选中。

衡量报价保真度

在执行后返回实际成本和延迟。如果没有观测值，路由器就缺少证据来判断报价中的置信度、价格和时间是否可靠。

使用统一效用评估权衡

随附的合成基准测试报告显示，与所列基线相比，在线 SAGE 可获得更高的模拟质量和统一效用，但其支出也高于静态 SAGE。这种明确的权衡非常重要：路由不应使用只关注能力的分数来掩盖成本。

该基准测试为合成测试，不能证明其在现实环境中更优。实际部署评估需要具有代表性的执行过程、强有力的路由基线、校准分析、市场轨迹重放、异构智能体和对抗性条件。

了解研究预览版的局限性

SAGE 是早期研究预览版，并不提供生产级服务水平保证，其结果也未经同行评审。其轻量级学习策略无法替代基于真实轨迹的训练或因果离线策略评估。

在部署到生产环境前，README 建议配置经过校准的评估器、已验证身份、签名能力元数据、隐私与安全审查、持久化事件驱动恢复、监控及任务专项验证。用于 A2A 发现、执行、流式传输和取消的实际适配器仍属于路线图项目，分布式服务运行和人工审批关卡也是如此。

代码库资源

• sprix_sage.py 包含上下文路由器、DAG 调度器、束搜索和在线更新。
• demo.py 提供端到端示例。
• benchmark.py 包含外部模拟器和统一效用基线。
• test_sprix_sage.py 包含行为单元测试。
• ALGORITHM.md 记录了形式化目标、搜索流程、贡献归因和局限性。
• GitHub 代码库包含贡献、安全、治理和许可证信息。

总结

Sprix SAGE Router 为状态感知智能体匹配提供了实用的参考实现。通过将工作表示为需求 DAG、在共享约束下比较 SELF、COLLABORATE 和 HANDOFF，并将执行证据反馈给路由器，你可以试验可审计的多智能体决策，同时兼顾预测质量和运营成本。

从 demo.py 开始，复现最小示例，检查每个路由决策，并运行测试和合成基准测试。对于实际的 A2A 部署，请添加单独的执行适配器，并完成你的环境所需的身份验证、安全、可观测性、恢复和验证工作。