Conversation Rooms

Deterministic, restart-safe turn-taking for a team of agents in one shared chat

A conversation room lets a roster of agents share one chat and reply on their own — you send a message, the room runs the scheduled turns in a deterministic, bounded, restart-safe order and hands you the replies.

Quick Start

Build a team of agents, name the roster, and send one message — the room runs every scheduled turn and returns the replies.

1. Simple room

   Three agents share one chat. on_message fans out to the whole roster and returns their replies.

   from praisonaiagents import Agent
   from praisonaiagents.rooms import Room, RoomConfig, RoomEvent
   
   researcher = Agent(name="researcher", instructions="Research the topic.")
   critic     = Agent(name="critic",     instructions="Critique proposals.")
   writer     = Agent(name="writer",     instructions="Write the final draft.")
   
   cfg = RoomConfig(name="standup", roster=["researcher", "critic", "writer"])
   room = Room(cfg, agents={
       "researcher": researcher.achat,
       "critic":     critic.achat,
       "writer":     writer.achat,
   })
   
   replies = await room.on_message(RoomEvent(speaker="user", content="plan the sprint @critic"))
   for turn in replies:
       print(f"{turn.speaker}: {turn.content}")

2. Configure caps

   Bound the conversation with max_rounds and max_messages so a room can never loop unbounded.

   from praisonaiagents.rooms import RoomConfig
   
   cfg = RoomConfig(
       name="standup",
       roster=["researcher", "critic", "writer"],
       max_rounds=2,
       max_messages=10,
   )

3. Load from YAML / dict

   Build the config from a parsed mapping. Either roster: or agents: names the roster.

   from praisonaiagents.rooms import RoomConfig
   
   cfg = RoomConfig.from_dict("standup", {
       "agents": ["researcher", "critic", "writer"],
       "max_rounds": 2,
   })

4. Advanced: drive the planner yourself

   If you’re building a wrapper that persists its own transcript, drive plan_next directly instead of using Room.

   from praisonaiagents.rooms import RoomEvent, RoundRobinRoomPlanner
   
   roster = ["planner", "coder", "reviewer"]
   transcript = [RoomEvent(speaker="user", content="build a login form", round=0)]
   
   planner = RoundRobinRoomPlanner()
   
   while (next_speaker := planner.plan_next(roster, transcript)) is not None:
       print("next up:", next_speaker)
       # Run the named agent, then append its reply below.
       transcript.append(RoomEvent(speaker=next_speaker, content="...", round=0))

User Interaction Flow

A real user builds a team and asks the room a question — no manual loop, no reset.

1. Create Agent objects.
2. Declare a RoomConfig naming the roster.
3. Build a Room(cfg, agents=...).
4. await room.on_message(RoomEvent(speaker="user", content="...")).
5. The room fans out to round 0, honours any @-mentions for later rounds, and returns the replies.
6. The next user message reopens a fresh activity automatically.

How a Room Runs

Room.on_message appends your event, drains the planner, runs each scheduled agent, and returns the replies.

The planner is pure, so persisting room.transcript and rebuilding the room is enough to resume — replaying a settled transcript returns no next speaker, so there is no double-execution.

Choosing a Configuration Surface

Pick where your room config lives.

Configuration Options

RoomConfig is a dataclass describing the roster and planner caps.

RoomConfig methods:

• build_planner() — constructs RoundRobinRoomPlanner(max_rounds=..., max_messages=...).
• to_dict() — plain dict for YAML round-tripping or diagnostics.
• RoomConfig.from_dict(name, data) — builds from a parsed mapping; accepts roster: or agents:. Rejects scalar values (bare string, bytes, mapping) with an actionable ValueError.

RoundRobinRoomPlanner is a dataclass with two hard caps.

RoomEvent is a frozen dataclass — immutable, hashable, and safe to persist or replay.

For the full Room and RoomConfig API surface, see the auto-generated SDK reference.

Turn-Taking Rules

Current Activity Semantics

The planner scopes accounting to the slice of the transcript from the last human (non-roster) message onward.

• A room only starts once a human (non-roster) message has been seen. Before that, plan_next returns None.
• max_messages is per activity, not lifetime. A settled room reliably restarts on the next human message, no matter how long the historical transcript is.
• @-mention admission is also scoped to the current activity — stale mentions from a prior settled activity do not leak into a new activity’s later rounds.

Mention Parsing

extract_mentions(text) returns the ordered, de-duplicated list of @name mentions found in text.

from praisonaiagents.rooms import extract_mentions

extract_mentions("@coder and @reviewer, then @coder again")
# ['coder', 'reviewer']  — first-seen order, de-duplicated

extract_mentions("email me at foo@bar.com")
# []  — a @ glued to a word character is not a mention

extract_mentions("")
# []

• First-seen order is preserved so addressing is deterministic.
• Case is preserved; matching against roster ids is case-sensitive.
• Name grammar: @ followed by [A-Za-z0-9][\w\-.]* — letters, digits, underscore, dash, dot; a name may not start with _, -, or ..
• Email-safe: a @ glued to a preceding word character does not match, so scanning human text is safe.

Common Patterns

from praisonaiagents.rooms import Room, RoomConfig, RoomEvent

async def reply(prompt: str) -> str:
    return "on it"

cfg = RoomConfig(name="r", roster=["ghost", "real"])
room = Room(cfg, agents={"real": reply})  # "ghost" has no callable

replies = await room.on_message(RoomEvent(speaker="user", content="hi"))
# "ghost" records passed=True; "real" replies.

from praisonaiagents.rooms import Room, RoomConfig

cfg = RoomConfig(name="r", roster=["a", "b"])
room = Room(
    cfg,
    agents={"a": a.achat, "b": b.achat},
    render=lambda transcript: "\n".join(f"[{e.speaker}] {e.content}" for e in transcript),
)

from typing import Optional, Sequence
from praisonaiagents.rooms import Room, RoomConfig, RoomEvent, RoomTurnPlannerProtocol

class FirstAgentPlanner:
    def plan_next(
        self,
        roster: Sequence[str],
        transcript: Sequence[RoomEvent],
    ) -> Optional[str]:
        return roster[0] if roster else None

assert isinstance(FirstAgentPlanner(), RoomTurnPlannerProtocol)

cfg = RoomConfig(name="r", roster=["a", "b"])
room = Room(cfg, agents={"a": a.achat, "b": b.achat}, planner=FirstAgentPlanner())

Best Practices

Make it Live

The core Room is in-memory and I/O-free by design. To run a room over a live platform (Telegram / Slack / Discord / webhook) with a durable transcript and restart-safe schedule, use GatewayRoom from praisonai_bot.bots — it wraps the same core Room, swaps the transcript in and out via BotSessionManager, and plugs in anywhere a single agent is bound today.

Related