Agent Pattern: Collaborative Agent (Shared Scratchpad)

[yasha-dev1]

Overview

The Collaborative Agent pattern enables multiple agents to work together transparently on a shared workspace (scratchpad). Unlike supervisor or swarm patterns where agents work in isolation and pass results, collaborative agents can see each other's work-in-progress, read and write to a common state, and build upon each other's contributions in real-time.

This pattern is particularly effective for complex problem-solving where agents benefit from observing peer reasoning and intermediate results.

How It Works

1. Shared State: A central scratchpad/workspace accessible to all agents
2. Transparent Contributions: Each agent can read what others have written
3. Incremental Building: Agents build on each other's work rather than starting fresh
4. State Deltas: Updates recorded as deltas (not full history) to manage context size
5. Coordination: Agents can see when others are working and what they're doing

Control Flow:

        Shared Scratchpad State
              ┌─────────────┐
              │ {           │
              │   research: │ ← Agent A writes
              │   analysis: │ ← Agent B reads A's work, adds analysis
              │   draft:    │ ← Agent C reads A+B, writes draft
              │   review:   │ ← Agent D reviews all work
              │ }           │
              └─────────────┘
                 ↑ ↑ ↑ ↑
                 │ │ │ │
              Agent A B C D (all read/write same state)

Reference Implementations

• Shared Scratchpad Collaboration Pattern - Official agentic design pattern
• LangGraph State Management - State machines for multi-agent workflows
• AWS Multi-Agent Systems with LangGraph - Shared state architecture
• Google ADK Context-Aware Framework - Context engineering for production agents
• Multi-Agent Orchestration 2025-2026 - Enterprise strategy guide

Proposed PyWorkflow Implementation

from pyworkflow_agents import CollaborativeAgent, SharedScratchpad
from pyworkflow_agents.providers import AnthropicProvider
from pyworkflow import workflow, get_context

# Create shared scratchpad
scratchpad = SharedScratchpad(
    channels={
        "research": {"type": "append"},      # Append-only channel
        "analysis": {"type": "replace"},     # Replace entire value
        "code": {"type": "merge"},           # Merge dict updates
        "review": {"type": "append"},
    },
    max_size_kb=100,  # Limit context size
)

# Define collaborative agents
researcher = CollaborativeAgent(
    name="researcher",
    provider=AnthropicProvider(model="claude-sonnet-4-5-20250929"),
    instructions="Research topics and write findings to 'research' channel.",
    scratchpad=scratchpad,
    read_channels=["research"],   # Can only read own channel
    write_channels=["research"],
)

analyst = CollaborativeAgent(
    name="analyst",
    provider=AnthropicProvider(model="claude-sonnet-4-5-20250929"),
    instructions="Read research findings and write analysis to 'analysis' channel.",
    scratchpad=scratchpad,
    read_channels=["research", "analysis"],   # Read research + own work
    write_channels=["analysis"],
)

coder = CollaborativeAgent(
    name="coder",
    provider=AnthropicProvider(model="claude-sonnet-4-5-20250929"),
    instructions="Implement code based on research and analysis.",
    scratchpad=scratchpad,
    read_channels=["research", "analysis", "code"],
    write_channels=["code"],
)

reviewer = CollaborativeAgent(
    name="reviewer",
    provider=AnthropicProvider(model="claude-sonnet-4-5-20250929"),
    instructions="Review all work and provide feedback.",
    scratchpad=scratchpad,
    read_channels=["research", "analysis", "code", "review"],  # Read everything
    write_channels=["review"],
)

# Collaborative workflow
@workflow(durable=True)
async def collaborative_workflow(task: str):
    """
    Agents collaborate via shared scratchpad stored in workflow context.
    """
    ctx = get_context()
    
    # Initialize scratchpad in workflow context
    ctx.scratchpad = scratchpad.initialize()
    
    # Sequential execution with shared state
    await researcher.run(task)
    # scratchpad.research now has researcher's findings
    
    await analyst.run("Analyze the research findings")
    # scratchpad.analysis now has analyst's insights (built on research)
    
    await coder.run("Implement the solution")
    # scratchpad.code now has implementation (built on research + analysis)
    
    await reviewer.run("Review all work")
    # scratchpad.review now has final review
    
    return ctx.scratchpad.to_dict()

# Alternative: Parallel collaboration
@workflow(durable=True)
async def parallel_collaborative_workflow(task: str):
    """
    Agents work in parallel, reading shared state as it updates.
    """
    ctx = get_context()
    ctx.scratchpad = scratchpad.initialize()
    
    # All agents run in parallel, reading/writing shared scratchpad
    results = await asyncio.gather(
        researcher.run(task),
        analyst.run(task),  # Polls scratchpad for research updates
        coder.run(task),    # Polls scratchpad for research + analysis
    )
    
    return ctx.scratchpad.to_dict()

Key Mapping to PyWorkflow Primitives:

• Scratchpad = Dict in WorkflowContext.scratchpad
• Channel updates = Recorded as SCRATCHPAD_UPDATE events (event-sourced)
• State replay = Scratchpad reconstructed from events during recovery
• Channel types = append, replace, merge (like LangGraph channels)
• Access control = Agent-level read/write permissions per channel

Event Types

New events for collaborative pattern:

class EventType(str, Enum):
    # Existing events...
    SCRATCHPAD_INIT = "scratchpad_init"           # Initialize scratchpad
    SCRATCHPAD_UPDATE = "scratchpad_update"       # Agent updates a channel
    SCRATCHPAD_READ = "scratchpad_read"           # Agent reads a channel (optional, for debugging)
    SCRATCHPAD_CONFLICT = "scratchpad_conflict"   # Concurrent write conflict

Event Data Schema:

# SCRATCHPAD_INIT
{
    "channels": ["research", "analysis", "code", "review"],
    "channel_types": {
        "research": "append",
        "analysis": "replace",
        "code": "merge",
        "review": "append"
    },
    "max_size_kb": 100
}

# SCRATCHPAD_UPDATE
{
    "agent_name": "researcher",
    "channel": "research",
    "operation": "append",  # or "replace", "merge"
    "delta": "Found 10 papers on multi-agent systems...",
    "timestamp": "2026-02-14T10:30:00Z",
    "scratchpad_size_kb": 12
}

# SCRATCHPAD_CONFLICT
{
    "agent_a": "analyst",
    "agent_b": "coder",
    "channel": "analysis",
    "resolution": "last_write_wins"  # or "merge", "reject"
}

Trade-offs

Pros

• Transparency: All agents see the full problem-solving process
• Incremental building: Agents build on each other's work instead of duplicating
• Coordination: Agents can observe what peers are doing
• Debugging: Scratchpad state visible in event log at every step
• Event sourcing: Full scratchpad history reconstructed during replay

Cons

• Context explosion: Scratchpad grows unbounded if not managed
• Coordination overhead: Agents must parse and understand peer contributions
• Concurrent writes: Need conflict resolution for parallel agents
• Mental model: Harder to reason about than supervisor delegation
• Error propagation: Mistakes in scratchpad affect all downstream agents

When to Use

• Complex problem-solving requiring multiple perspectives
• Tasks where seeing intermediate reasoning helps (mathematical proofs, code review)
• Collaborative workflows (multiple agents refining a document)
• Research tasks where cumulative knowledge is valuable

When to Avoid

• Simple linear workflows (supervisor is simpler)
• Independent parallel tasks (scatter-gather is better)
• Context-sensitive applications (scratchpad grows large)
• Tasks requiring strict isolation (agents shouldn't see each other's work)

Context Management Best Practices

Based on Google ADK research:

1. Scratchpad Hygiene: Summarize or compress observations before adding to scratchpad
2. Channel Scoping: Use channels to partition state (don't dump everything into one dict)
3. Delta Updates: Record only changes, not full state (append/merge, not replace)
4. Size Limits: Enforce max_size_kb to prevent context explosion
5. Error Handling: Don't dump raw error logs - summarize failures

Anti-pattern:

# BAD: Dumping raw error into scratchpad
scratchpad["errors"].append(full_stack_trace)  # 50KB of noise

Good pattern:

# GOOD: Summarize error for context purity
scratchpad["errors"].append({
    "step": "data_fetch",
    "error": "API rate limit exceeded",
    "action": "Retry in 60s"
})

Implementation Checklist

• Create SharedScratchpad class in pyworkflow_agents/scratchpad.py
• Implement channel types: append, replace, merge
• Add scratchpad to WorkflowContext
• Add SCRATCHPAD_* event types
• Implement event replay for scratchpad reconstruction
• Add size tracking and max_size_kb enforcement
• Implement conflict resolution strategies (last-write-wins, merge)
• Add agent-level read/write permissions per channel
• Create CollaborativeAgent class
• Add scratchpad visualization tool (show state over time)
• Create examples in examples/agents/collaborative_pattern.py
• Add tests for concurrent writes and conflict resolution
• Document scratchpad best practices and anti-patterns

Related Issues

• Agent Pattern: Supervisor (Manager + Workers) #154 - Supervisor Agent - Alternative orchestration pattern
• Agent Pattern: Swarm (Peer Handoffs) #157 - Swarm Agent - Handoff context vs shared state
• #TBD - Parallel Agent - Can combine with parallel scratchpad updates
• #TBD - Reflection/Self-Correction - Can use scratchpad for revision history

References

• Shared Scratchpad Collaboration Pattern
• LangGraph Multi-Agent Workflows
• AWS: Multi-Agent Systems with LangGraph
• Google ADK: Context-Aware Framework
• Multi-Agent Orchestration 2025-2026
• Multi-Agent Design Patterns
• Multi-Agent Systems Tutorial