Protocol patterns

Agent systems keep re-deriving common topologies. ap.patterns ships six projected shapes as functions. Each returns a Shape: a closed protocol plus the message classes it declares, so shape.Ask is a real ap.Message subclass. Patterns bind no participant, choose no model, and spend no allowance.

Use a pattern when its topology is the product requirement. Write a custom protocol when you need a message or branch the pattern does not express.

What a Shape contains

A shape has protocol and generated message classes. Render it before binding participants; then pass the shape directly to Session.

import agentsparty as ap

Reader, Writer = ap.roles('Reader', 'Writer')
shape = ap.patterns.request_reply(Reader, Writer)
print(ap.render(shape))

Generated classes are part of the shape's convenience, not a second public message hierarchy. shape.Ask and shape.Answer are the exact classes used by the protocol, so they can be passed to Session.ask, keyed in a Machine, or looked up in a Transcript. If a pattern's generated field is too small for your domain, pass an override where the pattern accepts one or write a custom protocol with your own class.

Choosing a pattern

Patterns are intentionally opinionated. pipeline does not report back to its first role; triage sends a Skip to desks it did not choose; swarm needs at least two peers and has no implicit timeout; best_of collects candidates but does not tell a judge how to rank them. Those choices are visible in the rendered protocol and can be replaced with a custom topology when they do not fit.

request_reply

One caller asks and one responder answers. It is the shortest route to Session.ask.

import agentsparty as ap

Reader, Writer = ap.roles('Reader', 'Writer')
shape = ap.patterns.request_reply(Reader, Writer)
session = ap.Session(
    shape,
    {
        Writer: ap.Agent(
            ap.replies([shape.Answer('ok')]),
            'answer briefly',
        )
    },
)
answer = session.ask(Reader, shape.Ask('question'), expect=shape.Answer)

review_loop

A requester delegates to a worker; the worker drafts; a reviewer approves or rejects. Rejection sends a revision notice and returns to the loop.

import agentsparty as ap

Requester, Worker, Reviewer = ap.roles('Requester', 'Worker', 'Reviewer')
shape = ap.patterns.review_loop(Requester, Worker, Reviewer)
print(ap.render(shape))

pipeline

A relay chain sends one typed class from the first role through each handoff. The mapping order is the pipeline order, and the chain does not silently join back to its first role.

import agentsparty as ap

First, Researcher, Analyst = ap.roles('First', 'Researcher', 'Analyst')


class Dossier(ap.Message):
    """A dossier."""

    text: str


shape = ap.patterns.pipeline(First, {Researcher: Dossier, Analyst: Dossier})

triage

A router chooses one specialist and sends every other desk a Skip. The chosen desk answers and the router relays that answer to the client.

import agentsparty as ap

Client, Router, Billing, Technical = ap.roles(
    'Client', 'Router', 'Billing', 'Technical'
)
shape = ap.patterns.triage(
    Client,
    Router,
    {
        Billing: 'billing questions',
        Technical: 'technical questions',
    },
)

swarm

A client gives a task to the first peer. Each peer either answers or hands off to the next peer in a ring; the last handoff loops back to the first.

import agentsparty as ap

Client, Searcher, Writer = ap.roles('Client', 'Searcher', 'Writer')
shape = ap.patterns.swarm(Client, [Searcher, Writer])
session = ap.Session(
    shape,
    {
        Searcher: ap.Agent(ap.replies([]), ''),
        Writer: ap.Agent(ap.replies([]), ''),
    },
)

best_of

A requester sends a task to several workers. Each sends a candidate to the judge, which returns the selected answer. Omit judge when the requester collects candidates itself.

import agentsparty as ap

Requester, First, Second, Judge = ap.roles(
    'Requester', 'First', 'Second', 'Judge'
)
shape = ap.patterns.best_of(Requester, [First, Second], Judge)
session = ap.Session(
    shape,
    {
        First: ap.Agent(ap.replies([]), ''),
        Second: ap.Agent(ap.replies([]), ''),
        Judge: ap.Agent(ap.replies([]), ''),
    },
)

Boundaries

Patterns return declarations, not a distributed scheduler. They do not bind roles, choose models, enforce an allowance, or add a join after par; use protocol for those explicit choices.

Extending a pattern safely

Treat the returned shape as a protocol value. You can pass it directly to a session, render it in a review, or use its classes in a cast. If the application needs an extra audit message, add a custom protocol around a pattern only when the added edge preserves the roles and sequencing that the pattern assumes. The safer alternative is often to copy the small pattern topology into a named protocol and make the new message explicit; hiding an extra side effect in a participant does not change the declared contract.

Related

• agentsparty.patterns.Shape: pattern shape contract.
• Protocol: compose a topology that is not a shipped pattern.
• Participants and roles: bind each shape's roles.
• Sessions and runs: execute a shape with run or ask.
• Test sessions offline: use replies to test a shape offline.