From 9e0ea5cee6b42010f0b07b0ac191977cbfdf37ae Mon Sep 17 00:00:00 2001 From: zinkerme-bit Date: Fri, 23 Oct 2026 08:34:32 +0200 Subject: [PATCH] docs(python): document coordinator and agent-tool APIs Explain strategy models and stages, strategy-server requests and registry propagation, agent-tool serialization, and the coordinator/agent-tool nodes. Add Google-style docstrings within the scope requested in issue #20. Assisted-by: OpenAI Codex --- sdk/python/jamjet/agent_tool.py | 57 +++++++-- .../jamjet/coordinator/default_strategy.py | 72 +++++++++++- sdk/python/jamjet/coordinator/server.py | 109 +++++++++++++++++- sdk/python/jamjet/coordinator/strategy.py | 95 +++++++++++++-- sdk/python/jamjet/workflow/nodes.py | 39 ++++++- 5 files changed, 351 insertions(+), 21 deletions(-) diff --git a/sdk/python/jamjet/agent_tool.py b/sdk/python/jamjet/agent_tool.py index 606b316d..cfb96a9e 100644 --- a/sdk/python/jamjet/agent_tool.py +++ b/sdk/python/jamjet/agent_tool.py @@ -1,3 +1,5 @@ +"""Agent-tool configuration objects and their IR serialization.""" + from __future__ import annotations from dataclasses import dataclass @@ -6,7 +8,19 @@ @dataclass class AgentToolDef: - """A wrapped agent that can be used as a tool by other agents.""" + """Configuration for an agent-tool invocation, without executing the agent. + + Attributes: + agent_uri: Explicit agent URI, or ``"auto"`` for an automatic target. + mode: Invocation mode: ``"sync"``, ``"streaming"``, or ``"conversational"``. + description: Human-readable description included in the IR fragment. + budget: Optional ``max_cost_usd`` and ``max_tokens`` limits, passed through + without validation. An empty mapping is omitted from the IR. + max_turns: Conversational turn limit. Serialization uses five when this + is ``None`` or zero; other modes do not serialize this field. + timeout_ms: Optional timeout in milliseconds. ``None`` and zero are + omitted from the IR. + """ agent_uri: str mode: str @@ -16,7 +30,17 @@ class AgentToolDef: timeout_ms: int | None = None def to_ir_kind(self) -> dict[str, Any]: - """Compile to IR node kind definition.""" + """Serialize the configuration as an agent-tool node-kind fragment. + + The target is encoded as ``{"auto": True}`` or ``{"explicit": uri}``. + Conversational mode becomes a mapping containing its turn limit; sync + and streaming modes remain strings. No agent is invoked or limits + validated by this method. + + Returns: + A dictionary containing the node type, target, mode, description, + and any nonempty budget or nonzero timeout. + """ agent_target: dict[str, Any] = {"auto": True} if self.agent_uri == "auto" else {"explicit": self.agent_uri} ir: dict[str, Any] = { "type": "agent_tool", @@ -41,15 +65,32 @@ def agent_tool( max_turns: int | None = None, timeout_ms: int | None = None, ) -> AgentToolDef: - """Wrap an agent as a callable tool. + """Create an agent-tool definition for later IR serialization. + + This stores configuration without calling an agent or validating the URI, + invocation mode, or limits. Args: - agent: Agent URI (e.g., "jamjet://org/classifier") or "auto" for coordinator routing. + agent: Agent URI, such as ``"jamjet://org/classifier"``, or ``"auto"`` + for an automatic target. description: Human-readable description of what this tool does. - mode: Invocation mode — "sync", "streaming", or "conversational". - budget: Budget constraints (max_cost_usd, max_tokens). - max_turns: Max turns for conversational mode. - timeout_ms: Timeout in milliseconds. + mode: Invocation mode: ``"sync"`` (default), ``"streaming"``, or + ``"conversational"``. + budget: Optional limits with keys ``max_cost_usd`` and ``max_tokens``. + An empty mapping is omitted during serialization. + max_turns: Turn limit for conversational mode only. ``None`` or zero + serializes as five turns. + timeout_ms: Optional timeout in milliseconds. ``None`` or zero is + omitted during serialization. + + Returns: + An ``AgentToolDef`` whose ``to_ir_kind()`` method produces a node-kind + dictionary. The definition itself is not a callable agent. + + Examples: + >>> tool = agent_tool("jamjet://org/classifier", "Classifies text") + >>> tool.to_ir_kind()["agent"] + {'explicit': 'jamjet://org/classifier'} """ return AgentToolDef( agent_uri=agent, diff --git a/sdk/python/jamjet/coordinator/default_strategy.py b/sdk/python/jamjet/coordinator/default_strategy.py index 26481dd8..779d3d08 100644 --- a/sdk/python/jamjet/coordinator/default_strategy.py +++ b/sdk/python/jamjet/coordinator/default_strategy.py @@ -1,3 +1,5 @@ +"""Registry filtering and heuristic scoring with an optional LLM tiebreaker.""" + from __future__ import annotations import json @@ -18,9 +20,15 @@ class DefaultCoordinatorStrategy(CoordinatorStrategy): - """Built-in coordinator strategy: structured scoring with optional LLM tiebreaker.""" + """Route using advertised agent metadata and optional model-based selection.""" def __init__(self, registry: Any = None) -> None: + """Initialize the strategy with an optional agent registry. + + Args: + registry: Object exposing an async list_agents() method that returns + agent mappings. Discovery returns no candidates when omitted. + """ self._registry = registry async def discover( @@ -31,6 +39,19 @@ async def discover( trust_domain: str | None, context: dict[str, Any], ) -> tuple[list[AgentCandidate], list[dict[str, str]]]: + """Filter registry agents by required skills and trust domain. + + Args: + task: Task description; unused by this implementation. + required_skills: Skills every returned agent must advertise. + preferred_skills: Optional skills; unused by this implementation. + trust_domain: Domain agents must match when this value is nonempty. + context: Additional inputs; unused by this implementation. + + Returns: + Eligible candidates and excluded agents with URI and reason mappings. + Both lists are empty when no registry is configured. + """ if self._registry is None: return [], [] @@ -80,6 +101,21 @@ async def score( weights: dict[str, float], context: dict[str, Any], ) -> tuple[list[ScoringResult], float]: + """Rank candidates using skill counts and advertised metadata. + + Args: + task: Task description passed to the capability scoring hook. The + default hook does not inspect its text. + candidates: Agents to score. + weights: Dimension weights passed to DimensionScores.composite(). + An empty mapping selects equal weights. + context: May contain preferred_reasoning_modes, a list of mode names + used to add a capability bonus for matching candidates. + + Returns: + Rankings sorted by descending composite score and the difference + between the first two scores, or 1.0 with fewer than two candidates. + """ results = [] for c in candidates: reasoning_bonus = self._score_reasoning_modes(c, context) @@ -111,6 +147,21 @@ async def decide( tiebreaker_model: str, context: dict[str, Any], ) -> Decision: + """Select the top candidate, using a model when the top scores are close. + + Args: + task: Task description supplied to the tiebreaker prompt. + top_candidates: Candidates in descending composite-score order. + threshold: Maximum difference between the top two scores that + enables a tiebreaker when a model is configured. + tiebreaker_model: Model identifier, or an empty string to disable it. + context: May contain agent_card_summaries, a mapping from agent URI + to a description included in the tiebreaker prompt. + + Returns: + A decision with no selection for an empty list, the top-ranked agent + for structured selection, or the tiebreaker's result. + """ if not top_candidates: return Decision(selected_uri=None, method="no_candidates", confidence=0.0) @@ -137,7 +188,20 @@ async def _llm_tiebreaker( context: dict[str, Any], max_candidates: int = 3, ) -> Decision: - """Call an LLM to break a tie between closely-scored candidates.""" + """Ask a model to select among the leading candidates. + + Args: + task: Task description included in the prompt. + candidates: Nonempty rankings in descending composite-score order. + model: Model identifier passed to call_llm(). + context: Optional agent_card_summaries keyed by agent URI. + max_candidates: Maximum number of leading candidates shown. + + Returns: + A model-backed decision with token counts. An invalid selected URI + falls back to the first candidate. A failed call or response parsing + returns a tiebreaker_failed decision selecting that same candidate. + """ tied = candidates[:max_candidates] # Build prompt with Agent Card summaries @@ -207,19 +271,23 @@ async def _llm_tiebreaker( ) def _score_capability(self, task: str, candidate: AgentCandidate) -> float: + """Return 0.3 without skills, otherwise skill count / 5 capped at 1.0.""" if not candidate.skills: return 0.3 return min(len(candidate.skills) / 5.0, 1.0) def _score_cost(self, candidate: AgentCandidate) -> float: + """Map low, medium, and high cost to scores; unknown cost scores 0.5.""" mapping = {"low": 1.0, "medium": 0.7, "high": 0.4} return mapping.get(candidate.cost_class or "", 0.5) def _score_latency(self, candidate: AgentCandidate) -> float: + """Map low, medium, and high latency to scores; unknown latency scores 0.5.""" mapping = {"low": 1.0, "medium": 0.7, "high": 0.4} return mapping.get(candidate.latency_class or "", 0.5) def _score_reasoning_modes(self, candidate: AgentCandidate, context: dict[str, Any]) -> float: + """Add up to 0.2 for overlap with context's preferred_reasoning_modes.""" preferred = context.get("preferred_reasoning_modes", []) if not preferred or not candidate.reasoning_modes: return 0.0 diff --git a/sdk/python/jamjet/coordinator/server.py b/sdk/python/jamjet/coordinator/server.py index 8aa4a83b..50abe2b9 100644 --- a/sdk/python/jamjet/coordinator/server.py +++ b/sdk/python/jamjet/coordinator/server.py @@ -1,3 +1,5 @@ +"""HTTP adapter for coordinator strategies; requires ``jamjet[sidecar]``.""" + from __future__ import annotations from typing import Any @@ -12,9 +14,25 @@ class StrategyServer: - """REST server that serves coordinator strategies to the Rust runtime.""" + """Expose discovery, scoring, and selection strategies over HTTP. + + Requests select a registered strategy with ``strategy_name``, defaulting to + ``"default"``. The built-in default discovers no agents until a registry is + supplied with ``set_registry()``. Construction configures routes; ``run()`` + starts the server. + + Attributes: + host: Interface on which ``run()`` binds the server. + port: TCP port used by ``run()``. + """ def __init__(self, host: str = "127.0.0.1", port: int = 4270) -> None: + """Configure strategy and health routes without starting a listener. + + Args: + host: Bind address, defaulting to the loopback interface. + port: Bind port, defaulting to 4270. + """ self.host = host self.port = port self._registry = None @@ -31,23 +49,64 @@ def __init__(self, host: str = "127.0.0.1", port: int = 4270) -> None: ) def register_strategy(self, name: str, strategy: CoordinatorStrategy) -> None: + """Register or replace a strategy under the given request name. + + Args: + name: Value clients send as ``strategy_name``. + strategy: Strategy implementation. A non-None shared registry is + assigned to the strategy's existing ``_registry`` attribute, + when present. + """ self._strategies[name] = strategy if self._registry is not None and hasattr(strategy, "_registry"): strategy._registry = self._registry def set_registry(self, registry: Any) -> None: + """Set the shared registry on existing strategies. + + Future registrations inherit it when it is not ``None``. Only strategies + exposing a ``_registry`` attribute receive the object. + + Args: + registry: Registry object. The default strategy expects an async + ``list_agents()`` method returning agent mappings. Passing + ``None`` disables discovery for that strategy. + """ self._registry = registry for strategy in self._strategies.values(): if hasattr(strategy, "_registry"): strategy._registry = registry def _get_strategy(self, data: dict) -> CoordinatorStrategy: + """Resolve ``strategy_name`` from a request, defaulting to ``"default"``. + + Args: + data: Decoded request body. + + Returns: + The registered strategy instance. + + Raises: + ValueError: The requested strategy name is not registered. + """ name = data.get("strategy_name", "default") if name not in self._strategies: raise ValueError(f"Unknown strategy: {name}") return self._strategies[name] async def _handle_discover(self, request: Request) -> JSONResponse: + """Handle discovery requests and serialize candidates and exclusions. + + Args: + request: JSON request containing ``task`` and optional + ``strategy_name``, ``required_skills``, ``preferred_skills``, + ``trust_domain``, and ``context`` values. + + Returns: + JSON with ``candidates`` and ``filtered_out``. Missing required fields or + ``ValueError``/``KeyError`` from strategy lookup or discovery produce + a 400 response containing ``error``. JSON decoding errors propagate. + """ data = await request.json() try: strategy = self._get_strategy(data) @@ -68,6 +127,18 @@ async def _handle_discover(self, request: Request) -> JSONResponse: return JSONResponse({"error": str(exc)}, status_code=400) async def _handle_score(self, request: Request) -> JSONResponse: + """Handle scoring requests and serialize rankings and score spread. + + Args: + request: JSON request containing ``task`` and optional + ``strategy_name``, ``candidates``, ``weights``, and ``context``. + Candidate mappings require ``uri``; other fields use defaults. + + Returns: + JSON with ``rankings`` and ``spread``. Missing required fields or + ``ValueError``/``KeyError`` during conversion or scoring produce a + 400 response containing ``error``. JSON decoding errors propagate. + """ data = await request.json() try: strategy = self._get_strategy(data) @@ -91,6 +162,21 @@ async def _handle_score(self, request: Request) -> JSONResponse: return JSONResponse({"error": str(exc)}, status_code=400) async def _handle_decide(self, request: Request) -> JSONResponse: + """Handle final selection requests and serialize the strategy decision. + + Args: + request: JSON request containing ``task`` and optional + ``strategy_name``, ``top_candidates``, ``threshold`` (0.1), + ``tiebreaker_model`` (empty string), and ``context``. Each + candidate requires ``uri``; missing dimension scores default to + 0.5 and missing composites to zero. + + Returns: + JSON with the selected URI, method, reasoning, confidence, rejected + candidates, and tiebreaker usage. Missing required fields or + ``ValueError``/``KeyError`` during conversion or selection produce a + 400 response containing ``error``. JSON decoding errors propagate. + """ data = await request.json() try: strategy = self._get_strategy(data) @@ -127,15 +213,24 @@ async def _handle_decide(self, request: Request) -> JSONResponse: return JSONResponse({"error": str(exc)}, status_code=400) async def _handle_health(self, request: Request) -> JSONResponse: + """Return server liveness and registered names without calling strategies.""" return JSONResponse({"status": "ok", "strategies": list(self._strategies.keys())}) def run(self) -> None: + """Run the HTTP server with Uvicorn, blocking until it stops. + + Install the ``jamjet[sidecar]`` extra for Starlette and Uvicorn. + + Raises: + ImportError: Uvicorn is unavailable in the current environment. + """ import uvicorn uvicorn.run(self._app, host=self.host, port=self.port) def _candidate_to_dict(c: AgentCandidate) -> dict: + """Serialize all candidate fields used by the strategy HTTP API.""" return { "uri": c.uri, "agent_card": c.agent_card, @@ -148,6 +243,17 @@ def _candidate_to_dict(c: AgentCandidate) -> dict: def _dict_to_candidate(d: dict) -> AgentCandidate: + """Read a candidate mapping, using empty or ``None`` optional defaults. + + Args: + d: Candidate mapping containing ``uri`` and optional metadata. + + Returns: + The corresponding candidate without validating its field values. + + Raises: + KeyError: The mapping has no ``uri`` field. + """ return AgentCandidate( uri=d["uri"], agent_card=d.get("agent_card", {}), @@ -160,6 +266,7 @@ def _dict_to_candidate(d: dict) -> AgentCandidate: def _scores_to_dict(s: DimensionScores) -> dict: + """Serialize the five individual scores without computing a composite.""" return { "capability_fit": s.capability_fit, "cost_fit": s.cost_fit, diff --git a/sdk/python/jamjet/coordinator/strategy.py b/sdk/python/jamjet/coordinator/strategy.py index 2deaf0d4..5b49c73b 100644 --- a/sdk/python/jamjet/coordinator/strategy.py +++ b/sdk/python/jamjet/coordinator/strategy.py @@ -1,3 +1,5 @@ +"""Data models and the discovery, scoring, and selection strategy interface.""" + from __future__ import annotations from abc import ABC, abstractmethod @@ -7,7 +9,17 @@ @dataclass class AgentCandidate: - """An agent discovered during the discovery phase.""" + """An agent available for coordinator routing. + + Attributes: + uri: Agent identifier used in rankings and routing decisions. + agent_card: Agent Card metadata available to custom strategies. + skills: Skills advertised by the agent. + latency_class: Advertised latency category, if known. + cost_class: Advertised cost category, if known. + trust_domain: Agent's trust domain, if specified. + reasoning_modes: Reasoning modes advertised by the agent. + """ uri: str agent_card: dict[str, Any] @@ -20,7 +32,15 @@ class AgentCandidate: @dataclass class DimensionScores: - """Per-dimension scoring for a single candidate.""" + """Per-dimension scores for one candidate. + + Attributes: + capability_fit: Suitability of the agent's capabilities. + cost_fit: Suitability of the agent's cost. + latency_fit: Suitability of the agent's latency. + trust_compatibility: Compatibility with trust requirements. + historical_performance: Score based on previous performance. + """ capability_fit: float = 0.5 cost_fit: float = 0.5 @@ -29,6 +49,16 @@ class DimensionScores: historical_performance: float = 0.5 def composite(self, weights: dict[str, float] | None = None) -> float: + """Combine the dimension scores using the supplied weights. + + Args: + weights: Weights keyed by dimension name. Missing dimensions use a + multiplier of 1.0. None or an empty mapping uses equal weights. + + Returns: + Weighted score sum divided by the sum of supplied weights, or 0.0 + when that sum is zero. + """ w = weights or { "capability_fit": 1.0, "cost_fit": 1.0, @@ -56,7 +86,13 @@ def composite(self, weights: dict[str, float] | None = None) -> float: @dataclass class ScoringResult: - """Scoring result for a single candidate.""" + """Scores and ranking value for one candidate. + + Attributes: + agent_uri: Identifier of the scored agent. + scores: Individual dimension scores. + composite: Combined score used to rank the candidate. + """ agent_uri: str scores: DimensionScores @@ -65,7 +101,17 @@ class ScoringResult: @dataclass class Decision: - """The coordinator's final routing decision.""" + """The coordinator's final routing decision. + + Attributes: + selected_uri: Selected agent, or None if no agent was selected. + method: Identifier describing how the selection was made. + reasoning: Optional explanation of the selection. + confidence: Confidence reported by the strategy. + rejected: Rejected agents, represented by URI and reason mappings. + tiebreaker_tokens: Optional input and output token counts for a tiebreaker. + tiebreaker_cost: Optional reported cost of the tiebreaker call. + """ selected_uri: str | None method: str @@ -77,7 +123,7 @@ class Decision: class CoordinatorStrategy(ABC): - """Base class for coordinator routing strategies.""" + """Define the discovery, scoring, and selection stages of agent routing.""" @abstractmethod async def discover( @@ -88,7 +134,18 @@ async def discover( trust_domain: str | None, context: dict[str, Any], ) -> tuple[list[AgentCandidate], list[dict[str, str]]]: - """Discover candidate agents. Returns (candidates, filtered_out).""" + """Discover agents eligible to handle a task. + + Args: + task: Description of the task to route. + required_skills: Skills required of an eligible agent. + preferred_skills: Skills preferred but not required. + trust_domain: Requested trust domain, or None for no restriction. + context: Additional strategy-specific discovery inputs. + + Returns: + Eligible candidates and filtered agents with URI and reason mappings. + """ ... @abstractmethod @@ -99,7 +156,18 @@ async def score( weights: dict[str, float], context: dict[str, Any], ) -> tuple[list[ScoringResult], float]: - """Score candidates. Returns (rankings sorted desc, spread).""" + """Score and rank discovered candidates. + + Args: + task: Description of the task to route. + candidates: Eligible agents returned by discovery. + weights: Weights keyed by scoring dimension name. + context: Additional strategy-specific scoring inputs. + + Returns: + Rankings in descending composite-score order and the spread between + the top two scores, as defined by the strategy. + """ ... @abstractmethod @@ -111,5 +179,16 @@ async def decide( tiebreaker_model: str, context: dict[str, Any], ) -> Decision: - """Make final selection. Called only when spread <= threshold.""" + """Select an agent from the ranked candidates. + + Args: + task: Description of the task to route. + top_candidates: Candidates in descending composite-score order. + threshold: Maximum score spread that warrants a tiebreaker. + tiebreaker_model: Model identifier to use for a tiebreaker. + context: Additional strategy-specific selection inputs. + + Returns: + The selected agent, selection method, and supporting metadata. + """ ... diff --git a/sdk/python/jamjet/workflow/nodes.py b/sdk/python/jamjet/workflow/nodes.py index 27e9322d..6cc84237 100644 --- a/sdk/python/jamjet/workflow/nodes.py +++ b/sdk/python/jamjet/workflow/nodes.py @@ -86,7 +86,20 @@ def to_ir_kind(self) -> dict[str, Any]: @dataclass class CoordinatorNode: - """Dynamic agent routing with structured scoring + LLM tiebreaker.""" + """Configure dynamic agent selection for a workflow graph. + + Attributes: + task: Task description supplied to the routing strategy. + required_skills: Skills that candidate agents must provide. + output_key: Output key for the selected agent's URI. Defaults to "result". + preferred_skills: Skills a routing strategy may use as preferences. + trust_domain: Optional trust domain used to filter candidates. + budget: Optional budget configuration forwarded to the runtime. + tiebreaker: Optional model and threshold configuration for close scores. + strategy: Registered strategy name. Defaults to "default". + weights: Scoring dimension weights, or None for the strategy defaults. + input_mapping: Workflow input mapping forwarded to the runtime. + """ task: str = "" required_skills: list[str] = field(default_factory=list) @@ -100,6 +113,12 @@ class CoordinatorNode: input_mapping: dict | None = None def to_ir_kind(self) -> dict[str, Any]: + """Serialize the coordinator configuration as an IR node kind. + + Returns: + A dictionary with empty mappings for unset weights and input mapping. + Trust domain, budget, and tiebreaker are included only when truthy. + """ ir = { "type": "coordinator", "task": self.task, @@ -121,7 +140,16 @@ def to_ir_kind(self) -> dict[str, Any]: @dataclass class AgentToolNode: - """Invoke a registered agent as a callable tool.""" + """Configure an agent invocation for a workflow graph. + + Attributes: + agent: Explicit agent URI, or "auto" for coordinator routing. + output_key: Output key for the agent's result. Defaults to "result". + mode: Invocation mode, such as "sync", "streaming", or "conversational". + input_mapping: Workflow input mapping forwarded to the runtime. + timeout_ms: Optional request timeout in milliseconds. + budget: Optional budget configuration forwarded to the runtime. + """ agent: str = "" output_key: str = "result" @@ -131,6 +159,13 @@ class AgentToolNode: budget: dict | None = None def to_ir_kind(self) -> dict[str, Any]: + """Serialize the agent invocation as an IR node kind. + + Returns: + A dictionary encoding the target as automatic or explicit. Unset + input mapping becomes an empty dictionary; timeout and budget are + included only when truthy. The mode is forwarded without validation. + """ agent_target: dict[str, Any] = {"auto": True} if self.agent == "auto" else {"explicit": self.agent} ir: dict[str, Any] = { "type": "agent_tool",