"""
Pattern 21: Cooperative Agent Pattern

Agents work together toward shared goals, sharing resources and information
to achieve outcomes that benefit the group.
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


@dataclass
class SharedGoal:
    """Goal shared among cooperative agents."""
    goal_id: str
    description: str
    target: Any
    contributors: List[str] = field(default_factory=list)
    contributions: Dict[str, Any] = field(default_factory=dict)
    status: str = "active"  # active, achieved, failed
    progress: float = 0.0


@dataclass
class Contribution:
    """Contribution from an agent."""
    agent_name: str
    goal_id: str
    contribution_type: str
    data: Any
    timestamp: datetime = field(default_factory=datetime.now)


class CooperativeAgent:
    """Agent that cooperates with others toward shared goals."""
    
    def __init__(self, name: str, role: str, skills: List[str], model_name: str = "gpt-4"):
        self.name = name
        self.role = role
        self.skills = skills
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.pattern: Optional['CooperativeAgentPattern'] = None
    
    def contribute_to_goal(self, goal: SharedGoal) -> Contribution:
        """Contribute to shared goal based on skills."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, a {self.role} with skills in {', '.join(self.skills)}. "
                      f"Work cooperatively to achieve the shared goal. "
                      f"Provide your contribution based on your skills."),
            ("human", f"Shared Goal: {goal.description}\n"
                     f"Target: {goal.target}\n"
                     f"Other Contributors: {', '.join(goal.contributors)}\n"
                     f"Current Progress: {goal.progress:.0%}\n\n"
                     f"What is your contribution?")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        contribution = Contribution(
            agent_name=self.name,
            goal_id=goal.goal_id,
            contribution_type=self.skills[0] if self.skills else "general",
            data=str(response.content)
        )
        
        return contribution
    
    def share_resource(self, resource_name: str, resource_data: Any):
        """Share resource with team."""
        if self.pattern:
            self.pattern.shared_resources[resource_name] = {
                "owner": self.name,
                "data": resource_data,
                "timestamp": datetime.now()
            }
    
    def request_help(self, task: str) -> List[str]:
        """Request help from other agents."""
        if not self.pattern:
            return []
        
        responses = []
        for agent in self.pattern.agents.values():
            if agent.name != self.name:
                # Ask if agent can help
                prompt = ChatPromptTemplate.from_messages([
                    ("system", f"You are {agent.name}. {self.name} is requesting help. "
                              f"Based on your skills ({', '.join(agent.skills)}), can you help?"),
                    ("human", f"Request: {task}\n\nCan you help? Respond yes/no and briefly explain how.")
                ])
                
                chain = prompt | agent.llm
                response = chain.invoke({})
                responses.append(f"{agent.name}: {str(response.content)[:100]}")
        
        return responses


class CooperativeAgentPattern:
    """
    Cooperative Agent Pattern - Team collaboration for shared goals.
    
    Agents work together, share resources, and contribute their
    skills toward achieving common objectives.
    """
    
    def __init__(self):
        self.agents: Dict[str, CooperativeAgent] = {}
        self.goals: Dict[str, SharedGoal] = {}
        self.shared_resources: Dict[str, Dict[str, Any]] = {}
        self.contribution_history: List[Contribution] = []
    
    def register_agent(self, agent: CooperativeAgent):
        """Register cooperative agent."""
        self.agents[agent.name] = agent
        agent.pattern = self
    
    def add_shared_goal(self, goal: SharedGoal):
        """Add shared goal for team."""
        self.goals[goal.goal_id] = goal
    
    def work_on_goal(self, goal_id: str):
        """Have all agents contribute to goal."""
        if goal_id not in self.goals:
            return
        
        goal = self.goals[goal_id]
        print(f"\n   Working on goal: {goal.description}")
        
        for agent in self.agents.values():
            print(f"   - {agent.name} contributing...")
            
            # Agent contributes
            contribution = agent.contribute_to_goal(goal)
            self.contribution_history.append(contribution)
            
            # Record contribution
            if agent.name not in goal.contributors:
                goal.contributors.append(agent.name)
            goal.contributions[agent.name] = contribution.data
            
            # Update progress
            goal.progress = len(goal.contributors) / max(len(self.agents), 1)
        
        # Check if goal achieved
        if goal.progress >= 1.0:
            goal.status = "achieved"
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute cooperative work scenario."""
        return {
            "pattern": "Cooperative",
            "total_agents": len(self.agents),
            "shared_goals": len(self.goals),
            "achieved_goals": sum(1 for g in self.goals.values() if g.status == "achieved"),
            "total_contributions": len(self.contribution_history),
            "shared_resources": len(self.shared_resources),
            "scenario": scenario
        }


# Example usage
if __name__ == "__main__":
    print("=== Cooperative Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = CooperativeAgentPattern()
    
    # Create cooperative agents
    print("1. Creating cooperative team...")
    researcher = CooperativeAgent("Alice", "researcher", ["research", "analysis"])
    developer = CooperativeAgent("Bob", "developer", ["coding", "testing"])
    designer = CooperativeAgent("Carol", "designer", ["UI/UX", "graphics"])
    writer = CooperativeAgent("Dave", "writer", ["documentation", "communication"])
    
    # Register agents
    for agent in [researcher, developer, designer, writer]:
        pattern.register_agent(agent)
    
    # Add shared goal
    print("\n2. Setting shared goal...")
    goal = SharedGoal(
        goal_id="proj1",
        description="Create a user-friendly mobile app",
        target="Launch MVP in 3 months"
    )
    pattern.add_shared_goal(goal)
    
    # Work on goal cooperatively
    print("\n3. Agents working cooperatively...")
    pattern.work_on_goal("proj1")
    
    # Agent requests help
    print("\n4. Agent requesting help from team...")
    responses = developer.request_help("Need help optimizing database queries")
    for response in responses[:2]:  # Show first 2 responses
        print(f"   {response}...")
    
    # Share resource
    print("\n5. Agent sharing resource...")
    researcher.share_resource("market_research", {"users": 1000, "feedback": "positive"})
    
    # Results
    result = pattern.run("Team collaboration")
    print(f"\n6. Cooperation Results:")
    print(f"   Team Size: {result['total_agents']}")
    print(f"   Goals Achieved: {result['achieved_goals']}/{result['shared_goals']}")
    print(f"   Total Contributions: {result['total_contributions']}")
    print(f"   Shared Resources: {result['shared_resources']}")
    
    print("\n✓ Cooperative agents worked together toward shared goals!")

    

Agentic Architecture Pattern: Cooperative Agent

Pattern 21: 21 Cooperative Agent – Line-by-Line Explanation

Overview

Pattern 21: Cooperative Agent Pattern

Agents work together toward shared goals, sharing resources and information to achieve outcomes that benefit the group.

Code Structure Analysis

File Information

• Pattern Number: 21
• Pattern Name: 21 Cooperative Agent
• File Path: patterns\collaboration\21_cooperative_agent.py
• Total Lines: 216

Detailed Code Explanation

Module Docstring

"""
Pattern 21: Cooperative Agent Pattern

Agents work together toward shared goals, sharing resources and information
to achieve outcomes that benefit the group.
"""

Explanation: This module docstring describes the purpose and key features of the 21 Cooperative Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Classes

SharedGoal (Line 16)

Description: Goal shared among cooperative agents.

Contribution (Line 28)

Description: Contribution from an agent.

CooperativeAgent (Line 37)

Description: Agent that cooperates with others toward shared goals.

Methods:

__init__ (Line 40)

def __init__(self, name, role, skills, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – role: Parameter for role – skills: Parameter for skills – model_name: LLM model to use (e.g., “gpt-4”)

contribute_to_goal (Line 47)

def contribute_to_goal(self, goal):
    """Contribute to shared goal based on skills."""

Purpose: Contribute to shared goal based on skills.

Parameters: – self: Instance reference – goal: Parameter for goal

share_resource (Line 72)

def share_resource(self, resource_name, resource_data):
    """Share resource with team."""

Purpose: Share resource with team.

Parameters: – self: Instance reference – resource_name: Parameter for resource_name – resource_data: Parameter for resource_data

request_help (Line 81)

def request_help(self, task):
    """Request help from other agents."""

Purpose: Request help from other agents.

Parameters: – self: Instance reference – task: Parameter for task

CooperativeAgentPattern (Line 103)

Description: Cooperative Agent Pattern – Team collaboration for shared goals.

Agents work together, share resources, and contribute their skills toward achieving common objectives.

Methods:

__init__ (Line 111)

def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 117)

def register_agent(self, agent):
    """Register cooperative agent."""

Purpose: Register cooperative agent.

Parameters: – self: Instance reference – agent: Parameter for agent

add_shared_goal (Line 122)

def add_shared_goal(self, goal):
    """Add shared goal for team."""

Purpose: Add shared goal for team.

Parameters: – self: Instance reference – goal: Parameter for goal

work_on_goal (Line 126)

def work_on_goal(self, goal_id):
    """Have all agents contribute to goal."""

Purpose: Have all agents contribute to goal.

Parameters: – self: Instance reference – goal_id: Parameter for goal_id

run (Line 153)

def run(self, scenario):
    """Execute cooperative work scenario."""

Purpose: Execute cooperative work scenario.

Parameters: – self: Instance reference – scenario: Parameter for scenario

Design Patterns Used

• Strategy Pattern: Different execution strategies for various scenarios
• State Pattern: Managing different states in the workflow
• Chain of Responsibility: Sequential processing through workflow nodes
• Observer Pattern: Tracking and monitoring pattern execution
• Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = SharedGoal(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

• LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
• Workflow Orchestration: LangGraph manages execution flow through nodes
• State Management: TypedDict defines workflow state structure
• Error Handling: Comprehensive error handling and fallback mechanisms
• Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities