Agent Communication

from typing import Literal
from pydantic import BaseModel

class Envelope(BaseModel):
    message_id: str
    task_id: str                    # ties every message to one task, across agents
    sender: str
    recipient: str
    type: Literal["request", "result", "question", "error", "status"]

class Result(BaseModel):
    envelope: Envelope
    status: Literal["complete", "partial", "failed"]
    findings: list[str]
    sources: list[str]              # every finding must be traceable
    not_done: str                   # what was asked and not achieved, and why
    tokens_used: int

def send(message: BaseModel) -> None:
    validate(message)               # reject malformed messages before anyone reads them
    log_message(message)            # every exchange is recorded for the trace
    inbox[message.envelope.recipient].put(message)

class Handoff(BaseModel):
    task_id: str
    from_agent: str
    to_agent: str
    goal: str                       # the user's request, in their words
    established: list[str]          # facts found, with ids
    actions_taken: list[str]        # e.g. "refund_delivery_fee(88213) -> ₹25 refunded"
    remaining: str
    constraints: list[str]          # e.g. "customer asked not to be phoned"

def hand_off(package: Handoff) -> None:
    set_owner(package.task_id, package.to_agent)        # exactly one owner, recorded
    send(package)
    if not acknowledged(package.task_id, within_seconds=30):
        set_owner(package.task_id, "human_queue")       # an unacknowledged task is not lost

def aggregate(results: list[Result]) -> dict:
    """Mechanical work in code. Only the final writing goes to a model."""
    failed = [r for r in results if r.status == "failed"]
    findings = dedupe([f for r in results for f in r.findings])
    conflicts = find_conflicts(findings)                 # e.g. two different revenue figures
    gaps = [r.not_done for r in results if r.not_done]

    answer = llm(SYNTHESIS_PROMPT.format(
        findings=findings, conflicts=conflicts, gaps=gaps,
        rule="Use only the findings listed. State every conflict and gap explicitly.",
    ))
    return {"answer": answer, "sources": dedupe([s for r in results for s in r.sources]),
            "incomplete": bool(failed or gaps)}

WORKER_RULES = """Return a report of at most 200 words.
- Findings as short bullets, each with its source.
- Do NOT include your search history, reasoning or the pages you read.
- If you produced something large (a table, a file), save it with save_artifact
  and return only its reference id."""

def save_artifact(content: str) -> str:
    """Bulk goes to storage. Contexts carry a reference, not the content."""
    ref = f"artifact-{uuid4().hex[:8]}"
    storage.put(ref, content)
    return ref                      # another agent reads it with read_artifact(ref) if needed

def record_action(task_id: str, agent: str, action: str, key: str) -> bool:
    """A shared ledger of actions. Returns False if someone already did this."""
    return db.insert_if_absent("actions", key=key,
                               row={"task_id": task_id, "agent": agent, "action": action})

def refund_once(task_id: str, order_id: int) -> str:
    """Before any write, every agent checks the ledger."""
    if not record_action(task_id, "billing", "refund_delivery_fee", key=f"{task_id}:refund_fee:{order_id}"):
        return "Already done by another agent. Do not repeat."
    return refund_delivery_fee(order_id)

DECISIONS = "Shared decisions so far (follow these; do not re-decide):\n{decisions}"
# e.g. "- Currency is INR throughout.\n- 'Late' means more than 30 minutes."

Q: How do agents communicate?

By message passing with private contexts, by sharing a conversation history, or by reading and writing structured shared state. Most systems use typed messages plus a small shared state; shared transcripts do not scale.

Q: Why structured messages?

So code can validate, route, log and count them, and so the receiver need not guess what is a finding, a caveat or a failure. An envelope with a task id lets a trace follow a task across agents.

Q: Delegation versus hand-off?

In delegation the sender keeps ownership and waits for a result. In a hand-off ownership moves and the receiver finishes the task. Hand-offs need an acknowledgement, a single recorded owner, and the list of actions already taken.

Q: What is context explosion and how do you prevent it?

Unbounded growth in what each agent reads, from shared transcripts, verbose reports and forwarding. Prevent it with private contexts, bounded structured reports, and references to stored artifacts instead of copies.

Q: How do you handle contradictory agents?

Detect conflicts in code before synthesis, resolve by evidence or a designated judge, state unresolved conflicts explicitly, and reduce them at source by settling shared definitions before work fans out.

Q: How do you prevent duplicated work?

Non-overlapping briefs with explicit scope, one owner per task, a shared ledger of actions checked before any write, and idempotency keys as the final safeguard.

Q: How do you aggregate results?

Merge, deduplicate and detect conflicts in code; carry each worker's stated gaps; have a model write only from the merged findings with an instruction to state conflicts and gaps.

Q: What is the cost of adding an agent?

Another prompt and tool set to pay for, more channels to define and log, more hand-offs where information is lost, and another trace to read. Overhead grows faster than headcount if agents talk freely.

Your task. Replace free-text messages between two agents with a typed protocol, a hand-off package and a shared action ledger, and show a duplicate action being prevented.

Steps

1. Start with a triage agent that hands off to a billing agent by passing the raw conversation.
2. Reproduce the bug: triage refunds a fee, hands off, and billing refunds it again.
3. Define Envelope, Handoff and Result as typed models, and validate every message before delivery.
4. Add actions_taken to the hand-off and a shared ledger checked before every write.
5. Require an acknowledgement within a time limit, with a fallback to a human queue.
6. Log every message with its task id and print the full exchange for one task.

You are done when

• The double refund no longer happens, and you can show which of your two safeguards stopped it.
• A malformed message is rejected before the receiving agent sees it.
• An unacknowledged hand-off ends up in the human queue, not lost.
• The printed exchange lets a stranger follow the task without reading either agent's full context.

Stretch. Add a third agent and count the channels. Then restructure so all messages go through one coordinator and count again.