"""
Pattern 155: Human-Agent Collaboration Pattern
Purpose: Seamless collaboration between humans and AI agents
"""

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


class CollaborationMode(Enum):
    """Modes of human-agent collaboration."""
    HUMAN_LED = "human_led"
    AGENT_LED = "agent_led"
    EQUAL_PARTNERSHIP = "equal_partnership"
    HUMAN_OVERSIGHT = "human_oversight"


@dataclass
class CollaborationTask:
    """A collaborative task."""
    id: str
    description: str
    mode: CollaborationMode
    human_input: Optional[str] = None
    agent_output: Optional[str] = None
    human_feedback: Optional[str] = None
    iterations: int = 0
    completed: bool = False
    created_at: datetime = field(default_factory=datetime.now)


class HumanAgentCollaborationPattern:
    """
    Human-Agent Collaboration - Synergistic human-AI teamwork
    
    Combines human creativity, judgment, and domain expertise
    with AI's computational power, pattern recognition, and scalability.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.tasks: List[CollaborationTask] = []
        self.collaboration_history: List[Dict[str, Any]] = []
        
    def create_task(self, task_id: str, description: str, mode: CollaborationMode) -> CollaborationTask:
        """Create a collaborative task."""
        task = CollaborationTask(id=task_id, description=description, mode=mode)
        self.tasks.append(task)
        return task
    
    def agent_process(self, task: CollaborationTask, context: str = "") -> str:
        """Agent processes task."""
        mode_guidance = {
            CollaborationMode.HUMAN_LED: "Follow human guidance closely. Provide support.",
            CollaborationMode.AGENT_LED: "Take initiative. Present recommendations to human.",
            CollaborationMode.EQUAL_PARTNERSHIP: "Collaborate as equals. Share decision-making.",
            CollaborationMode.HUMAN_OVERSIGHT: "Work autonomously but seek human approval."
        }
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"Collaboration mode: {mode_guidance[task.mode]}"),
            ("human", "Task: {task}\nContext: {context}\nProvide your contribution:")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"task": task.description, "context": context})
        
        task.agent_output = str(response.content)
        task.iterations += 1
        
        return task.agent_output
    
    def incorporate_human_input(self, task: CollaborationTask, human_input: str):
        """Incorporate human input into task."""
        task.human_input = human_input
        
        # Agent responds to human input
        if task.mode in [CollaborationMode.EQUAL_PARTNERSHIP, CollaborationMode.AGENT_LED]:
            self.agent_process(task, context=f"Human input: {human_input}")
    
    def incorporate_human_feedback(self, task: CollaborationTask, feedback: str):
        """Incorporate human feedback."""
        task.human_feedback = feedback
        
        # Refine based on feedback
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Refine your work based on human feedback"),
            ("human", "Previous output: {output}\nFeedback: {feedback}\nProvide improved version:")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "output": task.agent_output,
            "feedback": feedback
        })
        
        task.agent_output = str(response.content)
        task.iterations += 1
    
    def complete_task(self, task: CollaborationTask):
        """Mark task as complete."""
        task.completed = True
        
        self.collaboration_history.append({
            "task_id": task.id,
            "mode": task.mode.value,
            "iterations": task.iterations,
            "human_involved": task.human_input is not None,
            "feedback_provided": task.human_feedback is not None,
            "completed_at": datetime.now()
        })
    
    def run(self, task_id: str, description: str, mode: CollaborationMode, 
            human_inputs: List[str] = None) -> Dict[str, Any]:
        """Execute collaborative workflow."""
        task = self.create_task(task_id, description, mode)
        
        # Initial agent processing
        agent_output = self.agent_process(task)
        
        # Incorporate human inputs if provided
        if human_inputs:
            for human_input in human_inputs:
                if "feedback:" in human_input.lower():
                    self.incorporate_human_feedback(task, human_input)
                else:
                    self.incorporate_human_input(task, human_input)
        
        self.complete_task(task)
        
        return {
            "pattern": "HumanAgentCollaboration",
            "task_id": task_id,
            "mode": mode.value,
            "iterations": task.iterations,
            "final_output": task.agent_output[:200] + "..." if len(task.agent_output) > 200 else task.agent_output,
            "human_involvement": len(human_inputs) if human_inputs else 0
        }
    
    def get_stats(self) -> Dict[str, Any]:
        """Get collaboration statistics."""
        if not self.collaboration_history:
            return {"no_data": True}
        
        return {
            "total_collaborations": len(self.collaboration_history),
            "avg_iterations": sum(h["iterations"] for h in self.collaboration_history) / len(self.collaboration_history),
            "human_involvement_rate": sum(1 for h in self.collaboration_history if h["human_involved"]) / len(self.collaboration_history),
            "feedback_rate": sum(1 for h in self.collaboration_history if h["feedback_provided"]) / len(self.collaboration_history)
        }


if __name__ == "__main__":
    print("=== Human-Agent Collaboration Pattern Demo ===\n")
    
    pattern = HumanAgentCollaborationPattern()
    
    # Demo 1: Equal Partnership
    print("Demo 1: Equal Partnership Mode")
    result1 = pattern.run(
        "collab1",
        "Design a new mobile app feature",
        CollaborationMode.EQUAL_PARTNERSHIP,
        human_inputs=[
            "I think we should focus on accessibility",
            "Feedback: Great ideas, but make it simpler"
        ]
    )
    print(f"Mode: {result1['mode']}")
    print(f"Iterations: {result1['iterations']}")
    print(f"Output: {result1['final_output']}\n")
    
    # Demo 2: Human Oversight
    print("Demo 2: Human Oversight Mode")
    result2 = pattern.run(
        "collab2",
        "Analyze quarterly financial data",
        CollaborationMode.HUMAN_OVERSIGHT,
        human_inputs=["Approved, proceed with recommendations"]
    )
    print(f"Mode: {result2['mode']}")
    print(f"Output: {result2['final_output'][:100]}...\n")
    
    print("="*70)
    stats = pattern.get_stats()
    print(f"Statistics: {stats}")

    

Agentic Architecture Pattern: Human Agent Collaboration

Pattern 155: 155 Human Agent Collaboration Pattern – Line-by-Line Explanation

Overview

Pattern 155: Human-Agent Collaboration Pattern Purpose: Seamless collaboration between humans and AI agents

Code Structure Analysis

File Information

• Pattern Number: 155
• Pattern Name: 155 Human Agent Collaboration Pattern
• File Path: patterns\hybrid_ai\155_human_agent_collaboration_pattern.py
• Total Lines: 191

Detailed Code Explanation

Module Docstring

"""
Pattern 155: Human-Agent Collaboration Pattern
Purpose: Seamless collaboration between humans and AI agents
"""

Explanation: This module docstring describes the purpose and key features of the 155 Human Agent Collaboration Pattern pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate  # 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 enum import Enum: Enumeration support for fixed sets of values – from langchain_openai import ChatOpenAI: LangChain framework integration – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration

Enumerations

CollaborationMode (Line 14)

class CollaborationMode(Enum):
    HUMAN_LED
    AGENT_LED
    EQUAL_PARTNERSHIP
    HUMAN_OVERSIGHT

Purpose: Modes of human-agent collaboration.

Values: – HUMAN_LED: Represents human led – AGENT_LED: Represents agent led – EQUAL_PARTNERSHIP: Represents equal partnership – HUMAN_OVERSIGHT: Represents human oversight

Classes

CollaborationMode (Line 14)

Description: Modes of human-agent collaboration.

CollaborationTask (Line 23)

Description: A collaborative task.

HumanAgentCollaborationPattern (Line 36)

Description: Human-Agent Collaboration – Synergistic human-AI teamwork

Combines human creativity, judgment, and domain expertise with AI’s computational power, pattern recognition, and scalability.

Methods:

__init__ (Line 44)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – model_name: LLM model to use (e.g., “gpt-4”)

create_task (Line 49)

def create_task(self, task_id, description, mode):
    """Create a collaborative task."""

Purpose: Create a collaborative task.

Parameters: – self: Instance reference – task_id: Parameter for task_id – description: Parameter for description – mode: Parameter for mode

agent_process (Line 55)

def agent_process(self, task, context):
    """Agent processes task."""

Purpose: Agent processes task.

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

incorporate_human_input (Line 77)

def incorporate_human_input(self, task, human_input):
    """Incorporate human input into task."""

Purpose: Incorporate human input into task.

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

incorporate_human_feedback (Line 85)

def incorporate_human_feedback(self, task, feedback):
    """Incorporate human feedback."""

Purpose: Incorporate human feedback.

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

complete_task (Line 104)

def complete_task(self, task):
    """Mark task as complete."""

Purpose: Mark task as complete.

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

run (Line 117)

def run(self, task_id, description, mode, human_inputs):
    """Execute collaborative workflow."""

Purpose: Execute collaborative workflow.

Parameters: – self: Instance reference – task_id: Parameter for task_id – description: Parameter for description – mode: Parameter for mode – human_inputs: Parameter for human_inputs

get_stats (Line 144)

def get_stats(self):
    """Get collaboration statistics."""

Purpose: Get collaboration statistics.

Parameters: – self: Instance reference

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 = CollaborationMode(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