"""
Pattern 54: Transfer Learning

Purpose: Transfer knowledge from source tasks/domains to target tasks/domains.
Enables learning in new domains with limited data by leveraging prior knowledge.
"""

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


class TransferStrategy(Enum):
    """Strategies for transferring knowledge."""
    FINE_TUNING = "fine_tuning"
    FEATURE_EXTRACTION = "feature_extraction"
    DOMAIN_ADAPTATION = "domain_adaptation"
    MULTI_TASK = "multi_task"


@dataclass
class Task:
    """A learning task."""
    id: str
    name: str
    domain: str
    description: str
    examples: List[Dict[str, Any]] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class Knowledge:
    """Transferable knowledge."""
    id: str
    source_task: str
    knowledge_type: str  # patterns, features, strategies
    content: Dict[str, Any]
    confidence: float
    applicability: Dict[str, float] = field(default_factory=dict)  # task_id -> score


@dataclass
class TransferResult:
    """Result of knowledge transfer."""
    source_task: str
    target_task: str
    strategy: TransferStrategy
    knowledge_transferred: List[str]
    performance_before: float
    performance_after: float
    improvement: float
    timestamp: datetime = field(default_factory=datetime.now)


class TransferLearningAgent:
    """
    Transfer Learning Agent - Transfer knowledge across tasks and domains.
    
    Learns from source tasks and applies that knowledge to new target tasks
    with limited training data. Uses LLM to identify transferable patterns.
    """
    
    def __init__(self, 
                 name: str,
                 strategy: TransferStrategy = TransferStrategy.FINE_TUNING):
        self.name = name
        self.strategy = strategy
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
        
        self.source_tasks: Dict[str, Task] = {}
        self.target_tasks: Dict[str, Task] = {}
        self.knowledge_base: List[Knowledge] = []
        self.transfer_results: List[TransferResult] = []
        
        self.trained_tasks: Set[str] = set()
        self.statistics: Dict[str, Any] = {
            "tasks_learned": 0,
            "transfers_performed": 0,
            "avg_improvement": 0.0
        }
    
    def learn_source_task(self, task: Task):
        """Learn from a source task to build transferable knowledge."""
        print(f"\n   {self.name}: Learning source task '{task.name}'...")
        
        self.source_tasks[task.id] = task
        
        # Extract knowledge using LLM
        knowledge = self._extract_knowledge(task)
        self.knowledge_base.extend(knowledge)
        
        self.trained_tasks.add(task.id)
        self.statistics["tasks_learned"] += 1
        
        print(f"   Extracted {len(knowledge)} knowledge items from {task.name}")
    
    def transfer_to_target(self, target_task: Task) -> TransferResult:
        """Transfer knowledge from source tasks to target task."""
        print(f"\n   {self.name}: Transferring knowledge to '{target_task.name}'...")
        
        self.target_tasks[target_task.id] = target_task
        
        # Find relevant knowledge
        relevant_knowledge = self._find_relevant_knowledge(target_task)
        
        # Simulate performance before transfer
        performance_before = random.uniform(0.3, 0.5)
        
        # Apply transfer based on strategy
        if self.strategy == TransferStrategy.FINE_TUNING:
            performance_after = self._fine_tune_transfer(target_task, relevant_knowledge)
        elif self.strategy == TransferStrategy.FEATURE_EXTRACTION:
            performance_after = self._feature_extraction_transfer(target_task, relevant_knowledge)
        elif self.strategy == TransferStrategy.DOMAIN_ADAPTATION:
            performance_after = self._domain_adaptation_transfer(target_task, relevant_knowledge)
        else:  # MULTI_TASK
            performance_after = self._multi_task_transfer(target_task, relevant_knowledge)
        
        improvement = performance_after - performance_before
        
        result = TransferResult(
            source_task=",".join([k.source_task for k in relevant_knowledge]),
            target_task=target_task.id,
            strategy=self.strategy,
            knowledge_transferred=[k.id for k in relevant_knowledge],
            performance_before=performance_before,
            performance_after=performance_after,
            improvement=improvement
        )
        
        self.transfer_results.append(result)
        self.statistics["transfers_performed"] += 1
        
        # Update average improvement
        total_improvement = sum(r.improvement for r in self.transfer_results)
        self.statistics["avg_improvement"] = total_improvement / len(self.transfer_results)
        
        print(f"   Transfer complete: {performance_before:.2%} → {performance_after:.2%} (+{improvement:.2%})")
        
        return result
    
    def _extract_knowledge(self, task: Task) -> List[Knowledge]:
        """Extract transferable knowledge from task."""
        knowledge_items = []
        
        # Use LLM to identify patterns
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an expert at identifying transferable patterns in learning tasks. "
                      "Extract key patterns, features, and strategies that could apply to similar tasks."),
            ("human", f"Task: {task.name}\nDomain: {task.domain}\nDescription: {task.description}\n\n"
                     f"Identify 3 key transferable patterns. For each pattern, provide:\n"
                     f"1. Pattern name\n2. Why it's transferable\n3. Domains it could apply to\n\n"
                     f"Format: PATTERN|REASON|DOMAINS (comma-separated)")
        ])
        
        chain = prompt | self.llm
        try:
            response = chain.invoke({})
            content = response.content if isinstance(response.content, str) else str(response.content)
            lines = content.strip().split('\n')
            
            for i, line in enumerate(lines[:3]):
                try:
                    parts = line.split('|')
                    if len(parts) >= 3:
                        pattern_name = parts[0].strip()
                        reason = parts[1].strip()
                        domains = [d.strip() for d in parts[2].split(',')]
                        
                        knowledge = Knowledge(
                            id=f"k_{task.id}_{i}",
                            source_task=task.id,
                            knowledge_type="pattern",
                            content={
                                "pattern": pattern_name,
                                "reason": reason,
                                "domains": domains
                            },
                            confidence=0.8
                        )
                        knowledge_items.append(knowledge)
                except:
                    continue
        except:
            pass
        
        # Add generic knowledge from examples
        if task.examples:
            knowledge_items.append(Knowledge(
                id=f"k_{task.id}_examples",
                source_task=task.id,
                knowledge_type="examples",
                content={"examples": task.examples[:5]},
                confidence=0.6
            ))
        
        return knowledge_items
    
    def _find_relevant_knowledge(self, target_task: Task) -> List[Knowledge]:
        """Find knowledge relevant to target task."""
        relevant = []
        
        for knowledge in self.knowledge_base:
            # Check domain match
            if knowledge.knowledge_type == "pattern":
                domains = knowledge.content.get("domains", [])
                if target_task.domain in domains or "general" in domains:
                    relevant.append(knowledge)
            else:
                # Generic knowledge is always somewhat relevant
                relevant.append(knowledge)
        
        return relevant[:5]  # Top 5 most relevant
    
    def _fine_tune_transfer(self, task: Task, knowledge: List[Knowledge]) -> float:
        """Fine-tuning transfer strategy."""
        base_performance = 0.5
        
        # Benefit from each knowledge item
        for k in knowledge:
            base_performance += k.confidence * 0.1
        
        # Benefit from similar domain
        if any(task.domain in k.content.get("domains", []) for k in knowledge):
            base_performance += 0.15
        
        return min(0.95, base_performance)
    
    def _feature_extraction_transfer(self, task: Task, knowledge: List[Knowledge]) -> float:
        """Feature extraction transfer strategy."""
        base_performance = 0.45
        
        # Extract features from source
        feature_count = len(knowledge)
        base_performance += feature_count * 0.08
        
        return min(0.90, base_performance)
    
    def _domain_adaptation_transfer(self, task: Task, knowledge: List[Knowledge]) -> float:
        """Domain adaptation transfer strategy."""
        base_performance = 0.40
        
        # Adapt knowledge to new domain
        adapted_count = sum(1 for k in knowledge 
                          if task.domain in k.content.get("domains", []))
        base_performance += adapted_count * 0.12
        
        return min(0.92, base_performance)
    
    def _multi_task_transfer(self, task: Task, knowledge: List[Knowledge]) -> float:
        """Multi-task transfer strategy."""
        base_performance = 0.55
        
        # Benefit from multiple source tasks
        unique_sources = len(set(k.source_task for k in knowledge))
        base_performance += unique_sources * 0.08
        
        return min(0.93, base_performance)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get transfer learning statistics."""
        return {
            "name": self.name,
            "strategy": self.strategy.value,
            "source_tasks": len(self.source_tasks),
            "target_tasks": len(self.target_tasks),
            "knowledge_items": len(self.knowledge_base),
            **self.statistics
        }


class TransferLearningPattern:
    """
    Transfer Learning Pattern - Enable knowledge transfer across tasks.
    
    Coordinates multiple agents to learn from source tasks and apply
    knowledge to new target tasks with limited data.
    """
    
    def __init__(self):
        self.agents: Dict[str, TransferLearningAgent] = {}
        self.tasks: Dict[str, Task] = {}
        self.transfer_history: List[TransferResult] = []
    
    def register_agent(self, agent: TransferLearningAgent):
        """Register transfer learning agent."""
        self.agents[agent.name] = agent
        print(f"   Registered agent: {agent.name} ({agent.strategy.value})")
    
    def add_task(self, task: Task):
        """Add a task to the pattern."""
        self.tasks[task.id] = task
    
    def train_source_tasks(self, task_ids: List[str]):
        """Train all agents on source tasks."""
        print("\n   Training on source tasks...")
        
        for task_id in task_ids:
            task = self.tasks.get(task_id)
            if task:
                for agent in self.agents.values():
                    agent.learn_source_task(task)
    
    def transfer_to_targets(self, target_task_ids: List[str]) -> Dict[str, List[TransferResult]]:
        """Transfer knowledge to target tasks."""
        print("\n   Transferring to target tasks...")
        
        results = {}
        for task_id in target_task_ids:
            task = self.tasks.get(task_id)
            if task:
                results[task_id] = []
                for agent in self.agents.values():
                    result = agent.transfer_to_target(task)
                    results[task_id].append(result)
                    self.transfer_history.append(result)
        
        return results
    
    def run(self, 
            source_task_ids: List[str],
            target_task_ids: List[str]) -> Dict[str, Any]:
        """Run transfer learning pipeline."""
        print("=== Transfer Learning Pattern Demo ===\n")
        
        # Train on source tasks
        print("1. Learning from source tasks...")
        self.train_source_tasks(source_task_ids)
        
        # Transfer to target tasks
        print("\n2. Transferring to target tasks...")
        transfer_results = self.transfer_to_targets(target_task_ids)
        
        # Get statistics
        stats = {name: agent.get_statistics() 
                for name, agent in self.agents.items()}
        
        return {
            "pattern": "TransferLearning",
            "agents": len(self.agents),
            "source_tasks": len(source_task_ids),
            "target_tasks": len(target_task_ids),
            "transfer_results": transfer_results,
            "statistics": stats
        }


# Example usage
if __name__ == "__main__":
    print("=== Transfer Learning Pattern Demo ===\n")
    
    # Create pattern
    pattern = TransferLearningPattern()
    
    # Create agents with different strategies
    print("1. Creating agents...")
    fine_tune_agent = TransferLearningAgent(
        name="FineTuneAgent",
        strategy=TransferStrategy.FINE_TUNING
    )
    pattern.register_agent(fine_tune_agent)
    
    feature_agent = TransferLearningAgent(
        name="FeatureAgent",
        strategy=TransferStrategy.FEATURE_EXTRACTION
    )
    pattern.register_agent(feature_agent)
    
    # Create source tasks
    print("\n2. Creating source tasks...")
    sentiment_task = Task(
        id="sentiment",
        name="Sentiment Analysis",
        domain="text_classification",
        description="Classify text sentiment as positive/negative",
        examples=[
            {"text": "Great product!", "label": "positive"},
            {"text": "Terrible service", "label": "negative"},
        ]
    )
    pattern.add_task(sentiment_task)
    
    spam_task = Task(
        id="spam",
        name="Spam Detection",
        domain="text_classification",
        description="Classify emails as spam or not spam",
        examples=[
            {"text": "Win money now!", "label": "spam"},
            {"text": "Meeting tomorrow at 3pm", "label": "not_spam"},
        ]
    )
    pattern.add_task(spam_task)
    
    # Create target task
    print("\n3. Creating target task...")
    toxicity_task = Task(
        id="toxicity",
        name="Toxicity Detection",
        domain="text_classification",
        description="Classify text as toxic or not toxic",
        examples=[
            {"text": "You're an idiot", "label": "toxic"},
        ]
    )
    pattern.add_task(toxicity_task)
    
    # Run transfer learning
    print("\n4. Running transfer learning...")
    summary = pattern.run(
        source_task_ids=["sentiment", "spam"],
        target_task_ids=["toxicity"]
    )
    
    print(f"\n5. Results:")
    print(f"   Agents: {summary['agents']}")
    print(f"   Source Tasks: {summary['source_tasks']}")
    print(f"   Target Tasks: {summary['target_tasks']}")
    
    print(f"\n6. Transfer Results:")
    for task_id, results in summary['transfer_results'].items():
        print(f"\n   Target: {task_id}")
        for result in results:
            print(f"     {result.strategy.value}: {result.performance_before:.2%} → "
                  f"{result.performance_after:.2%} (+{result.improvement:.2%})")
    
    print(f"\n7. Agent Statistics:")
    for agent_name, stats in summary['statistics'].items():
        print(f"\n   {agent_name}:")
        for key, value in stats.items():
            if key != "name":
                print(f"     {key}: {value}")
    
    print("\n✓ Transfer learning enabled knowledge reuse across tasks!")

    

Agentic Architecture Pattern: Transfer Learning

Pattern 54: 54 Transfer Learning – Line-by-Line Explanation

Overview

Pattern 54: Transfer Learning

Purpose: Transfer knowledge from source tasks/domains to target tasks/domains. Enables learning in new domains with limited data by leveraging prior knowledge.

Code Structure Analysis

File Information

• Pattern Number: 54
• Pattern Name: 54 Transfer Learning
• File Path: patterns\learning_adaptation\54_transfer_learning.py
• Total Lines: 440

Detailed Code Explanation

Module Docstring

"""
Pattern 54: Transfer Learning

Purpose: Transfer knowledge from source tasks/domains to target tasks/domains.
Enables learning in new domains with limited data by leveraging prior knowledge.
"""

Explanation: This module docstring describes the purpose and key features of the 54 Transfer Learning pattern.

Imports Section

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import random
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, Set: 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_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

TransferStrategy (Line 17)

class TransferStrategy(Enum):
    FINE_TUNING
    FEATURE_EXTRACTION
    DOMAIN_ADAPTATION
    MULTI_TASK

Purpose: Strategies for transferring knowledge.

Values: – FINE_TUNING: Represents fine tuning – FEATURE_EXTRACTION: Represents feature extraction – DOMAIN_ADAPTATION: Represents domain adaptation – MULTI_TASK: Represents multi task

Classes

TransferStrategy (Line 17)

Description: Strategies for transferring knowledge.

Task (Line 26)

Description: A learning task.

Knowledge (Line 37)

Description: Transferable knowledge.

TransferResult (Line 48)

Description: Result of knowledge transfer.

TransferLearningAgent (Line 60)

Description: Transfer Learning Agent – Transfer knowledge across tasks and domains.

Learns from source tasks and applies that knowledge to new target tasks with limited training data. Uses LLM to identify transferable patterns.

Methods:

__init__ (Line 68)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – strategy: Parameter for strategy

learn_source_task (Line 87)

def learn_source_task(self, task):
    """Learn from a source task to build transferable knowledge."""

Purpose: Learn from a source task to build transferable knowledge.

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

transfer_to_target (Line 102)

def transfer_to_target(self, target_task):
    """Transfer knowledge from source tasks to target task."""

Purpose: Transfer knowledge from source tasks to target task.

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

_extract_knowledge (Line 147)

def _extract_knowledge(self, task):
    """Extract transferable knowledge from task."""

Purpose: Extract transferable knowledge from task.

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

_find_relevant_knowledge (Line 204)

def _find_relevant_knowledge(self, target_task):
    """Find knowledge relevant to target task."""

Purpose: Find knowledge relevant to target task.

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

_fine_tune_transfer (Line 220)

def _fine_tune_transfer(self, task, knowledge):
    """Fine-tuning transfer strategy."""

Purpose: Fine-tuning transfer strategy.

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

_feature_extraction_transfer (Line 234)

def _feature_extraction_transfer(self, task, knowledge):
    """Feature extraction transfer strategy."""

Purpose: Feature extraction transfer strategy.

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

_domain_adaptation_transfer (Line 244)

def _domain_adaptation_transfer(self, task, knowledge):
    """Domain adaptation transfer strategy."""

Purpose: Domain adaptation transfer strategy.

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

_multi_task_transfer (Line 255)

def _multi_task_transfer(self, task, knowledge):
    """Multi-task transfer strategy."""

Purpose: Multi-task transfer strategy.

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

get_statistics (Line 265)

def get_statistics(self):
    """Get transfer learning statistics."""

Purpose: Get transfer learning statistics.

Parameters: – self: Instance reference

TransferLearningPattern (Line 277)

Description: Transfer Learning Pattern – Enable knowledge transfer across tasks.

Coordinates multiple agents to learn from source tasks and apply knowledge to new target tasks with limited data.

Methods:

__init__ (Line 285)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 290)

def register_agent(self, agent):
    """Register transfer learning agent."""

Purpose: Register transfer learning agent.

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

add_task (Line 295)

def add_task(self, task):
    """Add a task to the pattern."""

Purpose: Add a task to the pattern.

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

train_source_tasks (Line 299)

def train_source_tasks(self, task_ids):
    """Train all agents on source tasks."""

Purpose: Train all agents on source tasks.

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

transfer_to_targets (Line 309)

def transfer_to_targets(self, target_task_ids):
    """Transfer knowledge to target tasks."""

Purpose: Transfer knowledge to target tasks.

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

run (Line 325)

def run(self, source_task_ids, target_task_ids):
    """Run transfer learning pipeline."""

Purpose: Run transfer learning pipeline.

Parameters: – self: Instance reference – source_task_ids: Parameter for source_task_ids – target_task_ids: Parameter for target_task_ids

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