PRD 04: Inter-Agent Communication & Collaboration

1. Overview

Purpose

Enable agents to communicate, share knowledge, and collaborate on complex tasks through structured messaging, shared memory, and coordinated decision-making.

Vision Alignment

Agents form "organs" in the Context Engineering paradigm - specialized cells working together toward common goals through:

• Message passing

• Shared context

• Collaborative reasoning

• Consensus building

2. Problem Statement

Current system has no:

• Agent-to-agent messaging

• Shared knowledge base

• Coordination protocols

• Consensus mechanisms

• Collaborative problem solving

3. Success Criteria

• Agents can send/receive messages

• Shared memory accessible by agent teams

• Collaborative reasoning produces better results

• Consensus mechanisms prevent conflicts

• Performance scales with agent count

4. Functional Requirements

4.1 Communication Protocol

class AgentCommunicationProtocol:
    """
    Structured messaging between agents
    """
    
    async def send_message(
        self,
        from_agent: Agent,
        to_agent: Agent,
        message_type: MessageType,
        content: Any,
        priority: int = 5
    ) -> MessageResult:
        # Validate message structure
        # Add to message queue
        # Notify recipient
        # Track delivery
        
    async def broadcast(
        self,
        from_agent: Agent,
        team: List[Agent],
        message: Message
    ) -> BroadcastResult:
        # Send to all team members
        # Track acknowledgments
        # Handle failures

4.2 Message Types

class MessageType(Enum):
    TASK_REQUEST = "task_request"           # Request help with task
    KNOWLEDGE_SHARE = "knowledge_share"     # Share information
    CONSENSUS_REQUEST = "consensus_request" # Request agreement
    RESULT_SHARE = "result_share"          # Share results
    COORDINATION = "coordination"           # Coordinate actions
    FEEDBACK = "feedback"                  # Provide feedback

4.3 Shared Context Management

class SharedContextManager:
    """
    Manages shared knowledge between agents
    """
    
    async def create_shared_context(
        self,
        team: List[Agent],
        initial_context: Dict
    ) -> SharedContext:
        # Create shared memory space
        # Set access permissions
        # Initialize with context
        
    async def update_shared_context(
        self,
        context_id: str,
        agent: Agent,
        updates: Dict,
        merge_strategy: str = "consensus"
    ) -> UpdateResult:
        # Validate update permissions
        # Apply merge strategy
        # Propagate changes
        # Handle conflicts

4.4 Collaborative Reasoning

class CollaborativeReasoner:
    """
    Enables multi-agent problem solving
    """
    
    async def collaborative_solve(
        self,
        problem: Problem,
        agents: List[Agent],
        strategy: str = "ensemble"
    ) -> Solution:
        # Distribute problem to agents
        # Collect individual solutions
        # Apply collaboration strategy
        # Synthesize final solution
        
    async def consensus_building(
        self,
        proposals: List[Proposal],
        agents: List[Agent],
        method: str = "weighted_vote"
    ) -> Consensus:
        # Present proposals to agents
        # Collect votes/opinions
        # Apply consensus method
        # Return agreed solution

5. Technical Architecture

5.1 Communication Infrastructure

┌─────────────────────────────────────────────────────────┐
│                Communication Layer                       │
├─────────────────────────────────────────────────────────┤
│                                                          │
│  ┌──────────┐      Message Bus       ┌──────────┐       │
│  │ Agent A  │─────────────────────────│ Agent B  │       │
│  └──────────┘           ↓             └──────────┘       │
│       ↓            ┌──────────┐            ↓            │
│  ┌──────────┐      │  Router  │      ┌──────────┐       │
│  │  Memory  │      └──────────┘      │  Memory  │       │
│  └──────────┘           ↓             └──────────┘       │
│       ↓         ┌──────────────┐          ↓             │
│  ┌──────────────│ Shared Context│──────────────┐         │
│  └──────────────└──────────────┘──────────────┘         │
│                                                          │
└─────────────────────────────────────────────────────────┘

5.2 Collaboration Strategies

class CollaborationStrategies:
    ENSEMBLE = {
        "description": "All agents contribute equally",
        "aggregation": "weighted_average",
        "conflict_resolution": "voting"
    }
    
    HIERARCHICAL = {
        "description": "Lead agent coordinates others",
        "aggregation": "leader_decision",
        "conflict_resolution": "leader_override"
    }
    
    SPECIALIZED = {
        "description": "Each agent handles their expertise",
        "aggregation": "domain_based",
        "conflict_resolution": "expert_priority"
    }
    
    CONSENSUS = {
        "description": "All agents must agree",
        "aggregation": "unanimous",
        "conflict_resolution": "negotiation"
    }

6. Implementation Details

6.1 Message Queue Implementation

class AgentMessageQueue:
    """
    Redis-backed message queue for agents
    """
    
    def __init__(self):
        self.redis_client = redis.Redis(...)
        self.pubsub = self.redis_client.pubsub()
    
    async def publish(
        self,
        channel: str,
        message: Message
    ):
        # Serialize message
        message_json = json.dumps({
            "id": str(uuid4()),
            "from": message.from_agent_id,
            "to": message.to_agent_id,
            "type": message.type,
            "content": message.content,
            "timestamp": datetime.now().isoformat(),
            "priority": message.priority
        })
        
        # Publish to Redis channel
        await self.redis_client.publish(
            f"agent:{message.to_agent_id}",
            message_json
        )
        
        # Store in message history
        await self.store_message(message)

6.2 Collaborative Problem Solving

async def solve_collaboratively(
    self,
    problem: Problem,
    team: List[Agent]
) -> Solution:
    """
    Multi-agent collaborative problem solving
    """
    # Phase 1: Problem Distribution
    shared_context = await self.create_shared_context(
        team=team,
        initial_context={
            "problem": problem.description,
            "constraints": problem.constraints,
            "objectives": problem.objectives
        }
    )
    
    # Phase 2: Individual Analysis
    individual_analyses = []
    for agent in team:
        analysis = await agent.analyze_problem(
            problem=problem,
            shared_context=shared_context
        )
        individual_analyses.append(analysis)
        
        # Share insights
        await self.update_shared_context(
            context_id=shared_context.id,
            agent=agent,
            updates={"insights": analysis.insights}
        )
    
    # Phase 3: Solution Generation
    proposed_solutions = []
    for agent in team:
        # Agent sees all insights
        enriched_context = await self.get_shared_context(shared_context.id)
        
        solution = await agent.generate_solution(
            problem=problem,
            context=enriched_context,
            other_insights=individual_analyses
        )
        proposed_solutions.append(solution)
    
    # Phase 4: Consensus Building
    consensus = await self.build_consensus(
        proposals=proposed_solutions,
        agents=team,
        method="weighted_vote"
    )
    
    # Phase 5: Solution Synthesis
    final_solution = await self.synthesize_solution(
        consensus=consensus,
        all_proposals=proposed_solutions,
        shared_context=enriched_context
    )
    
    return final_solution

7. Database Schema

-- Agent messages
CREATE TABLE agent_messages (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    from_agent_id INTEGER REFERENCES agents(id),
    to_agent_id INTEGER REFERENCES agents(id),
    message_type VARCHAR(50),
    content JSONB,
    priority INTEGER,
    status VARCHAR(50), -- sent, delivered, read
    created_at TIMESTAMP DEFAULT NOW(),
    delivered_at TIMESTAMP,
    read_at TIMESTAMP
);

-- Shared contexts
CREATE TABLE shared_contexts (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    name VARCHAR(255),
    team_id UUID,
    context_data JSONB,
    version INTEGER DEFAULT 1,
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

-- Context access permissions
CREATE TABLE context_permissions (
    id SERIAL PRIMARY KEY,
    context_id UUID REFERENCES shared_contexts(id),
    agent_id INTEGER REFERENCES agents(id),
    permission_level VARCHAR(50), -- read, write, admin
    granted_at TIMESTAMP DEFAULT NOW()
);

-- Collaboration sessions
CREATE TABLE collaboration_sessions (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    problem_id INTEGER REFERENCES tasks(id),
    team_agents INTEGER[],
    strategy VARCHAR(50),
    shared_context_id UUID REFERENCES shared_contexts(id),
    status VARCHAR(50),
    result JSONB,
    started_at TIMESTAMP DEFAULT NOW(),
    completed_at TIMESTAMP
);

8. API Endpoints

# Send message between agents
POST /api/agents/communicate
{
    "from_agent_id": "...",
    "to_agent_id": "...",
    "message_type": "knowledge_share",
    "content": {...},
    "priority": 5
}

# Create collaboration session
POST /api/collaboration/create
{
    "problem": {...},
    "agent_ids": [...],
    "strategy": "ensemble",
    "timeout": 300
}

# Update shared context
PUT /api/collaboration/context/{context_id}
{
    "agent_id": "...",
    "updates": {...},
    "merge_strategy": "consensus"
}

# Get collaboration result
GET /api/collaboration/session/{session_id}/result

9. Success Metrics

• Message delivery rate: > 99.9%

• Collaboration success rate: > 85%

• Consensus achievement time: < 30s

• Solution quality improvement: > 30%

• Agent coordination efficiency: > 80%