协调器节点

协调器在运行时发现代理，从多个维度对它们进行评分，并路由到最佳匹配。当评分接近时，可选的 LLM 决胜机制会做出最终决策。

三个阶段

1. 发现

根据所需技能和信任域过滤代理注册表。缺少所需技能或信任域不匹配的代理会被拒绝并说明原因。

graph.add_coordinator("route",
    task="Route support ticket to best agent",
    required_skills=["support", "billing"],
    trust_domain="internal",
)

2. 评分

每个候选代理从五个维度进行评分:

维度	衡量内容	默认权重
`capability_fit`	技能覆盖度 + 推理模式匹配	1.0
`cost_fit`	代理成本等级(低=1.0，中=0.7，高=0.4)	1.0
`latency_fit`	代理延迟等级(相同映射)	1.0
`trust_compatibility`	信任域匹配(匹配=1.0，否则=0.5)	1.0
`historical_performance`	历史成功率(默认 0.5)	1.0

加权综合得分决定排名。可以覆盖权重来优先考虑重要因素:

graph.add_coordinator("route",
    task="Route latency-sensitive request",
    required_skills=["inference"],
    weights={"latency_fit": 3.0, "cost_fit": 0.5},
)

3. 决策

如果首选候选者的领先优势明显(差距 > 阈值)，则通过结构化评分胜出。如果分数接近，则启用 LLM 决胜机制。

LLM 决胜机制

当顶级候选者之间的分数差距在阈值范围内时，协调器会调用 LLM，结合任务上下文和代理卡片摘要来做出最终选择。

graph.add_coordinator("route",
    task="Route complex research task",
    required_skills=["research"],
    tiebreaker={"model": "claude-sonnet-4-6", "threshold": 0.1},
)

工作原理:

顶级候选者(最多 3 个)被格式化为结构化提示
LLM 返回 JSON: {"selected_uri": "...", "reasoning": "..."}
决策包含 method="llm_tiebreaker" 和 token 使用量以便跟踪成本
如果 LLM 调用失败，则回退到结构化选择并标记 method="tiebreaker_failed"

决胜机制使用异步 SDK 客户端(AsyncAnthropic、AsyncOpenAI)——优先尝试 Anthropic，失败后回退到 OpenAI。

推理模式

Agent Card 可以声明推理能力：

AgentCandidate(
    uri="jamjet://org/planner",
    agent_card={"name": "Planning Agent"},
    skills=["task-decomposition"],
    reasoning_modes=["plan-and-execute", "react"],
    cost_class="medium",
    latency_class="medium",
)

当协调器上下文包含 preferred_reasoning_modes 时，匹配的 agent 会获得能力分数加成（最多 +0.2）：

context = {"preferred_reasoning_modes": ["plan-and-execute"]}

如果未设置偏好，推理模式不会影响评分。

自定义策略

默认策略可以处理大多数情况，但你可以通过继承 CoordinatorStrategy 来实现自己的策略：

from jamjet.coordinator.strategy import CoordinatorStrategy, Decision

class MyStrategy(CoordinatorStrategy):
    async def discover(self, task, required_skills, preferred_skills, trust_domain, context):
        # 自定义发现逻辑
        ...

    async def score(self, task, candidates, weights, context):
        # 自定义评分逻辑
        ...

    async def decide(self, task, top_candidates, threshold, tiebreaker_model, context):
        # 自定义决策逻辑
        ...

在策略服务器中注册它：

from jamjet.coordinator.server import StrategyServer

server = StrategyServer()
server.register("my-strategy", MyStrategy())
server.run()

决策输出

协调器返回一个完全透明的 Decision：

Decision(
    selected_uri="jamjet://org/agent-a",
    method="llm_tiebreaker",        # 或 "structured"、"tiebreaker_failed"
    reasoning="最适合研究分解任务",
    confidence=0.82,
    rejected=[{"uri": "jamjet://org/agent-b", "reason": "not selected by tiebreaker"}],
    tiebreaker_tokens={"input": 150, "output": 30},
)

每个路由决策都会记录在事件日志中，以便观测和回放。

示例

参考 coordinator-routing 示例查看包含多个 agent 和评分的完整工作配置。

协调器节点

On this page