Skip to content

fleet

fleet

Fleet config generator — produces concert-of-concerts from agent roster.

A fleet is a simplified YAML that launches and manages multiple agent scores as a unit. The fleet generator takes a compiler config and produces: 1. Individual agent scores (via the compilation pipeline) 2. A fleet config referencing those scores with group dependencies

Classes

FleetGenerator

Generates fleet configuration from a compiler config.

Takes the full compiler config with agent roster and produces a fleet YAML that references individual agent scores with group dependencies for startup ordering.

Functions
generate 
generate(config, scores_dir)

Generate a fleet config from the compiler config.

Parameters:

Name Type Description Default
config dict[str, Any]

Full compiler config with project, agents, groups.

 required
scores_dir Path

Directory where individual agent scores will be written.

 required

Returns:

Type Description
dict[str, Any]

Fleet config dict suitable for YAML serialization.
Source code in src/marianne/compose/fleet.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def generate(
    self,
    config: dict[str, Any],
    scores_dir: Path,
) -> dict[str, Any]:
    """Generate a fleet config from the compiler config.

    Args:
        config: Full compiler config with project, agents, groups.
        scores_dir: Directory where individual agent scores will be written.

    Returns:
        Fleet config dict suitable for YAML serialization.
    """
    project_name = config.get("project", {}).get("name", "unnamed")
    agents = config.get("agents", [])

    if not agents:
        raise ValueError("Fleet config requires at least one agent")

    # Build score entries
    scores: list[dict[str, str | None]] = []
    for agent in agents:
        name = agent.get("name", "")
        if not name:
            continue
        group = agent.get("group")
        entry: dict[str, str | None] = {
            "path": str(scores_dir / f"{name}.yaml"),
        }
        if group:
            entry["group"] = group
        scores.append(entry)

    # Build group definitions from config
    groups = self._build_groups(config)

    fleet: dict[str, Any] = {
        "name": f"{project_name}-fleet",
        "type": "fleet",
        "scores": scores,
    }

    if groups:
        fleet["groups"] = groups

    return fleet
write 
write(config, scores_dir, output_path)

Generate and write a fleet config to disk.

Parameters:

Name Type Description Default
config dict[str, Any]

Full compiler config.

 required
scores_dir Path

Directory where agent scores live.

 required
output_path Path

Path to write the fleet YAML.

 required

Returns:

Type Description
Path

Path to the written fleet config.
Source code in src/marianne/compose/fleet.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def write(
    self,
    config: dict[str, Any],
    scores_dir: Path,
    output_path: Path,
) -> Path:
    """Generate and write a fleet config to disk.

    Args:
        config: Full compiler config.
        scores_dir: Directory where agent scores live.
        output_path: Path to write the fleet YAML.

    Returns:
        Path to the written fleet config.
    """
    fleet = self.generate(config, scores_dir)

    output_path.parent.mkdir(parents=True, exist_ok=True)
    with open(output_path, "w") as f:
        yaml.dump(fleet, f, default_flow_style=False, sort_keys=False)

    _logger.info("Fleet config written to %s", output_path)
    return output_path