Swarm : 300 agents trient les issues d'un dépôt sur un tableau de messages commun

@app.post("/api/boards/{board}/messages")
def send(board: str, message: NewMessage) -> dict:
    with pool.connection() as db:
        # One writer at a time, so ids commit in order: a reader paging with after= never skips one.
        db.execute("LOCK TABLE messages IN SHARE ROW EXCLUSIVE MODE")
        (id,) = db.execute(
            "INSERT INTO messages (board, author, text) VALUES (%s, %s, %s) RETURNING id",
            (board, message.author, message.text[:2000]),
        ).fetchone()
    return {"id": id}

@app.get("/api/boards/{board}/messages")
def read(board: str, after: int = 0, limit: int = Query(50, le=200)) -> list[dict]:
    """The board's messages with an id greater than `after`, oldest first."""
    with pool.connection() as db:
        rows = db.execute(
            "SELECT id, author, text, at FROM messages WHERE board = %s AND id > %s ORDER BY id LIMIT %s",
            (board, after, limit),
        ).fetchall()
    return [
        {"id": id, "author": author, "text": text, "at": at.isoformat(timespec="seconds")}
        for id, author, text, at in rows
    ]

class Member(agents.SessionInput):
    """Who this session is on the board."""

    input_id = "swarm.member"
    name: str

@agents.tool
async def send_swarm_message(text: str, me: Annotated[Member, agents.Depends()]) -> str:
    """Post a one-line message on the swarm's board, where every member can read it."""
    response = await board.post("/messages", json={"author": me.name, "text": text})
    response.raise_for_status()
    return f"posted as message {response.json()['id']}"

@agents.tool(annotations=agents.ToolAnnotations(read_only_hint=True, destructive_hint=False))
async def read_message(after: int = 0, limit: int = 50) -> str:
    """Read the board's messages with an id greater than `after`, oldest first, at most `limit` (200 at most)."""
    response = await board.get("/messages", params={"after": after, "limit": limit})
    response.raise_for_status()
    lines = [f"{m['id']} {m['author']}: {m['text']}" for m in response.json()]
    return "\n".join(lines) or "(no newer messages)"

triager = agents.Agent(
    model="mistral-large-4",
    name="issue-triager",
    instructions=(
        "You are one member of a swarm that triages a repository's open issues. Each member owns one "
        "issue. The swarm talks only through its board: send_swarm_message posts, "
        "read_message(after=, limit=) reads. Keep every message to one line."
    ),
    tools=[send_swarm_message, read_message],
)

editor = agents.Agent(
    model="mistral-large-4",
    name="triage-editor",
    instructions=(
        "You write the triage report of a swarm that summarized a repository's open issues on its "
        "board. Read the whole board with read_message, paging with after=. Then answer with the report "
        "alone, in Markdown: the groups of duplicates, each with its issue numbers and the shared problem "
        "in one line; then the five areas with the most issues."
    ),
    tools=[read_message],  # it reads the board, and writes only its answer
)

SUMMARIZE = (
    "Your issue is #{number}: {title}\n\n{body}\n\n"
    "Call send_swarm_message once with '#{number} [<area>] <the symptom in at most 15 words>'. "
    "Then answer 'posted'."
)
FIND_DUPLICATES = (
    "Every member has posted. Read the whole board, paging with after= until there are no newer "
    "messages. If another issue reports the same problem as yours, call send_swarm_message with "
    "'#<yours> duplicates #<other>: <why, in 10 words>'. Otherwise post nothing. "
    "Answer with the duplicate's number, or 'none'."
)

sessions = await asyncio.gather(
    *(
        triager.start_session(session_inputs={Member.input_id: Member(name=f"triager-{issue['number']}")})
        for issue in issues
    )
)
try:
    # Two rounds, each gathered: every member has posted before any member looks for duplicates.
    # A round waits ROUND_SECONDS at most: a member still running then is left out, so one slow
    # turn never holds up the other 299.
    summaries = await asyncio.gather(
        *(
            session.run(
                SUMMARIZE.format(number=issue["number"], title=issue["title"], body=(issue["body"] or "")[:1500]),
                timeout=ROUND_SECONDS,
            )
            for session, issue in zip(sessions, issues)
        )
    )
    late = sum(isinstance(state, agents.Running) for state in summaries)
    print(f"round 1 after {time.monotonic() - started:.0f} s: {len(issues) - late} members done, {late} left out")

    answers = await asyncio.gather(*(session.run(FIND_DUPLICATES, timeout=ROUND_SECONDS) for session in sessions))
    found = [state for state in answers if isinstance(state, agents.Completed) and "none" not in state.text.lower()]
    print(f"round 2 after {time.monotonic() - started:.0f} s: {len(found)} members found a duplicate")
finally:
    await asyncio.gather(*(session.cancel() for session in sessions))

match await editor.run("Write the triage report."):
    case agents.Completed(text=report):
        print(f"\n{report}")
    case other:
        print(f"the editor stopped: {other}")

docker run -d --name swarm-board-db -p 5432:5432 \
  -e POSTGRES_USER=board -e POSTGRES_PASSWORD=board -e POSTGRES_DB=board postgres:17
uv sync
POSTGRES_HOST=127.0.0.1 POSTGRES_USER=board POSTGRES_PASSWORD=board POSTGRES_DB=board \
  uv run uvicorn main:app --port 8000

curl -fsSL https://mistral.ai/cli/install.sh | bash
export PATH="$HOME/.mistral/bin:$PATH"

{"flags": {"mistral_cli_apps": true}}

mistral login
mistral deploy --from-disk --name swarm-board

export SWARM_BOARD_URL=<the URL the deploy printed>

uv sync
export MISTRAL_API_KEY=...
export GITHUB_TOKEN=...  # optional: GitHub limits the requests it answers without one
uv run python swarm.py astral-sh/uv 300