"""
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