"""
Pattern 116: Multi-Task Learning
Description:
Agent learns multiple related tasks simultaneously using shared representations
with task-specific heads for specialization.
Components:
- Shared representations
- Task-specific heads
- Multi-task optimization
Use Cases:
- Related task sets
- Efficient training
- Knowledge sharing
LangChain Implementation:
Uses multi-task prompt templates and shared context for learning across tasks.
"""
import os
from typing import List, Dict, Any
from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
load_dotenv()
class MultiTaskLearningAgent:
"""Agent that learns multiple related tasks simultaneously."""
def __init__(self, model_name: str = "gpt-4"):
self.llm = ChatOpenAI(model=model_name, temperature=0.7)
self.tasks = {}
self.shared_knowledge = {}
self.task_performance = {}
def register_task(self, task_name: str, task_description: str, examples: List[Dict[str, str]]):
"""Register a new task for multi-task learning."""
self.tasks[task_name] = {
"description": task_description,
"examples": examples
}
self.task_performance[task_name] = []
def learn_from_shared_representations(self):
"""Extract shared knowledge from all registered tasks."""
shared_prompt = ChatPromptTemplate.from_messages([
("system", "Analyze these tasks and identify shared patterns, concepts, and skills that apply across them."),
("user", "Tasks:\n{tasks}\n\nIdentify shared knowledge:")
])
tasks_description = "\n\n".join([
f"{name}: {info['description']}\nExamples: {info['examples']}"
for name, info in self.tasks.items()
])
chain = shared_prompt | self.llm | StrOutputParser()
self.shared_knowledge["patterns"] = chain.invoke({"tasks": tasks_description})
print("Shared Knowledge Extracted:")
print(self.shared_knowledge["patterns"])
print()
def execute_task(self, task_name: str, input_data: str) -> str:
"""Execute a specific task using shared knowledge and task-specific head."""
if task_name not in self.tasks:
return f"Task '{task_name}' not registered"
task_info = self.tasks[task_name]
# Create task-specific prompt with shared knowledge
task_prompt = ChatPromptTemplate.from_messages([
("system", """You are a multi-task learning agent.
Shared Knowledge (applies to all tasks):
{shared_knowledge}
Current Task: {task_name}
Description: {task_description}
Examples:
{examples}
Use both shared knowledge and task-specific patterns."""),
("user", "{input}")
])
examples_str = "\n".join([
f"Input: {ex['input']}\nOutput: {ex['output']}"
for ex in task_info['examples']
])
chain = task_prompt | self.llm | StrOutputParser()
result = chain.invoke({
"shared_knowledge": self.shared_knowledge.get("patterns", "None extracted yet"),
"task_name": task_name,
"task_description": task_info['description'],
"examples": examples_str,
"input": input_data
})
return result
def evaluate_task(self, task_name: str, test_cases: List[Dict[str, str]]):
"""Evaluate performance on a specific task."""
if task_name not in self.tasks:
print(f"Task '{task_name}' not registered")
return
print(f"\n=== Evaluating Task: {task_name} ===")
correct = 0
total = len(test_cases)
for case in test_cases:
result = self.execute_task(task_name, case['input'])
is_correct = case['expected'].lower() in result.lower()
if is_correct:
correct += 1
print(f"Input: {case['input']}")
print(f"Expected: {case['expected']}")
print(f"Got: {result}")
print(f"✓" if is_correct else "✗")
print()
accuracy = correct / total if total > 0 else 0
self.task_performance[task_name].append(accuracy)
print(f"Accuracy: {accuracy:.2%}\n")
def get_task_correlations(self) -> str:
"""Analyze how tasks benefit from shared learning."""
correlation_prompt = ChatPromptTemplate.from_messages([
("system", "Analyze how these tasks benefit from shared multi-task learning."),
("user", """Tasks and Performance:
{task_info}
Shared Knowledge:
{shared_knowledge}
Explain how tasks benefit from each other and which skills transfer across tasks.""")
])
task_info = "\n".join([
f"{name}: {self.tasks[name]['description']} - Performance: {self.task_performance[name]}"
for name in self.tasks
])
chain = correlation_prompt | self.llm | StrOutputParser()
return chain.invoke({
"task_info": task_info,
"shared_knowledge": self.shared_knowledge.get("patterns", "Not extracted")
})
def demonstrate_multi_task_learning():
"""Demonstrate multi-task learning pattern."""
print("=== Multi-Task Learning Pattern ===\n")
agent = MultiTaskLearningAgent()
# Register related NLP tasks
print("1. Registering Multiple Related Tasks")
print("-" * 50)
# Task 1: Sentiment Analysis
agent.register_task(
"sentiment_analysis",
"Classify the sentiment of text as positive, negative, or neutral",
[
{"input": "I love this product!", "output": "positive"},
{"input": "This is terrible.", "output": "negative"},
{"input": "It's okay, nothing special.", "output": "neutral"}
]
)
# Task 2: Text Classification
agent.register_task(
"text_classification",
"Classify text into categories: technology, sports, politics, entertainment",
[
{"input": "New AI breakthrough announced", "output": "technology"},
{"input": "Championship game tonight", "output": "sports"},
{"input": "Election results are in", "output": "politics"}
]
)
# Task 3: Intent Detection
agent.register_task(
"intent_detection",
"Detect user intent: question, command, statement, request",
[
{"input": "What time is it?", "output": "question"},
{"input": "Please send the report.", "output": "command"},
{"input": "I think it will rain.", "output": "statement"}
]
)
print("✓ Registered 3 related NLP tasks\n")
# Extract shared knowledge
print("2. Learning Shared Representations")
print("-" * 50)
agent.learn_from_shared_representations()
# Test on sentiment analysis
print("3. Testing on Sentiment Analysis Task")
print("-" * 50)
test_cases = [
{"input": "This is amazing and wonderful!", "expected": "positive"},
{"input": "Worst experience ever", "expected": "negative"},
]
agent.evaluate_task("sentiment_analysis", test_cases)
# Test on text classification
print("4. Testing on Text Classification Task")
print("-" * 50)
test_cases = [
{"input": "Latest smartphone release announced", "expected": "technology"},
{"input": "Team wins the finals", "expected": "sports"},
]
agent.evaluate_task("text_classification", test_cases)
# Analyze task correlations
print("5. Analyzing Task Correlations and Knowledge Transfer")
print("-" * 50)
correlations = agent.get_task_correlations()
print(correlations)
print()
print("=== Summary ===")
print("Multi-task learning demonstrated with:")
print("- Multiple related NLP tasks")
print("- Shared knowledge extraction")
print("- Task-specific execution with shared representations")
print("- Knowledge transfer analysis")
if __name__ == "__main__":
demonstrate_multi_task_learning()
Agentic Design Pattern: Multi Task Learning
Pattern 116: Multi Task Learning – Line-by-Line Explanation
Overview
This document provides a detailed line-by-line explanation of the Multi Task Learning pattern implementation.
Pattern Description
Pattern 116: Multi-Task Learning
Description: Agent learns multiple related tasks simultaneously using shared representations with task-specific heads for specialization.
Components: – Shared representations – Task-specific heads – Multi-task optimization
Use Cases: – Related task sets – Efficient training – Knowledge sharing
LangChain Implementation: Uses multi-task prompt templates and shared context for learning across tasks.
File Statistics
- Total Lines: 245
- Classes: 1
- Functions: 1
- Constants: 0
Module Documentation
"""
Pattern 116: Multi-Task Learning
Description:
Agent learns multiple related tasks simultaneously using shared representations
with task-specific heads for specialization.
Components:
- Shared representations
- Task-specific heads
- Multi-task optimization
Use Cases:
- Related task sets
- Efficient training
- Knowledge sharing
LangChain Implementation:
Uses multi-task prompt templates and shared context for learning across tasks.
"""
Purpose: This module implements the Multi Task Learning pattern, providing a structured approach to agent design and implementation.
Import Statements
Standard Library Imports
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
Purpose: Core Python functionality:
– langchain_core.output_parsers.StrOutputParser: Regular expression operations
– langchain_core.prompts.ChatPromptTemplate: Regular expression operations
– os: Environment variable access and system operations
– typing.Any: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.List: Type hints for better code documentation (List, Dict, Any, Optional)
LangChain Imports
from langchain_openai.ChatOpenAI
Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows
Other Imports
from dotenv.load_dotenv
Purpose: Additional dependencies for specialized functionality.
Classes
Class: MultiTaskLearningAgent
Lines 33-157
Purpose:
Agent that learns multiple related tasks simultaneously.
Methods:
__init__(self, model_name)
Lines 36-40
Purpose: Initialize the class instance with required parameters.
Parameters:
– model_name: Language model instance
register_task(self, task_name, task_description, examples)
Lines 42-48
Purpose:
Register a new task for multi-task learning.
Parameters:
– task_name: Task name
– task_description: Task description
– examples: Examples
learn_from_shared_representations(self)
Lines 50-67
Purpose:
Extract shared knowledge from all registered tasks.
Parameters:
execute_task(self, task_name, input_data)
Lines 69-108
Purpose:
Execute a specific task using shared knowledge and task-specific head.
Parameters:
– task_name: Task name
– input_data: Input data
Returns: str
evaluate_task(self, task_name, test_cases)
Lines 110-133
Purpose:
Evaluate performance on a specific task.
Parameters:
– task_name: Task name
– test_cases: Test cases
get_task_correlations(self)
Lines 135-157
Purpose:
Analyze how tasks benefit from shared learning.
Parameters:
Returns: str
Functions
demonstrate_multi_task_learning()
Lines 160-240
Purpose:
Demonstrate multi-task learning pattern.
Key Concepts Summary
Pattern Overview
The Multi Task Learning pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.
Core Components
- Agent Class: Main implementation of the pattern logic
- Prompt Engineering: Carefully crafted prompts to guide LLM behavior
- Tool Integration: Optional tools for extended functionality
- Memory Management: State handling across interactions
- Output Parsing: Structured response processing
When to Use
✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability
❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)
Best Practices Demonstrated
1. Environment Configuration
- ✅ Use
.envfiles for API keys - ✅ Never hardcode credentials
- ✅ Validate environment variables on startup
2. Error Handling
- ✅ Graceful degradation on failures
- ✅ Informative error messages
- ✅ Logging for debugging
3. Prompt Engineering
- ✅ Clear, specific instructions
- ✅ Structured output formats
- ✅ Few-shot examples when beneficial
4. Code Organization
- ✅ Separation of concerns (class-based design)
- ✅ Type hints for clarity
- ✅ Docstrings for documentation
5. Performance Optimization
- ✅ Appropriate temperature settings
- ✅ Token budget management
- ✅ Caching where applicable
LangChain Components Used
- ChatOpenAI: OpenAI model interface
- Chains: Sequential component composition
Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development
Execution Flow
1. Initialize agent with configuration
↓
2. Load environment and dependencies
↓
3. Prepare prompt with input data
↓
4. Invoke LLM through LangChain
↓
5. Parse and validate response
↓
6. Return formatted result
Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation
Performance Considerations
Computational Cost
- LLM Calls: Primary cost factor (based on tokens)
- Latency: Depends on model and input size
- Memory: Minimal for most patterns
Optimization Strategies
- Prompt Optimization: Minimize tokens while maintaining clarity
- Caching: Cache repeated queries/responses
- Model Selection: Use smaller models when appropriate
- Batch Processing: Group similar requests
- Streaming: For long responses, stream tokens
Cost Estimates (Approximate)
- GPT-3.5-turbo: ~$0.002 per request
- GPT-4: ~$0.03 per request
- GPT-4-turbo: ~$0.01 per request
Actual costs vary based on input/output token counts
Common Pitfalls to Avoid
1. Prompt Design Issues
❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts
2. Error Handling
❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation
3. Token Management
❌ Exceeding model token limits
✅ Truncate inputs, chunk processing
4. Security
❌ Exposing API keys in code
✅ Use environment variables
5. Testing
❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs
6. Cost Management
❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring
Further Improvements
Production Enhancements
- Monitoring: Add logging, metrics, tracing
- Validation: Input/output schema validation
- Rate Limiting: Implement request throttling
- Caching: Redis or in-memory cache for responses
- Async Support: Non-blocking operations for scalability
Feature Extensions
- Multi-Model Support: Switch between providers
- Fallback Models: Graceful degradation strategy
- Custom Tools: Domain-specific function calling
- Advanced Memory: Vector stores for long-term memory
- Evaluation: Automated testing and benchmarking
Code Quality
- Testing: Unit tests, integration tests
- Documentation: API docs, usage examples
- Type Safety: Stricter type hints
- Configuration: Externalized config files
Related Patterns
- Pattern 115
- Pattern 117
Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…
Summary
The Multi Task Learning pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:
✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns
🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability
📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/
File: 116_multi_task_learning.py
Total Lines: 245
Pattern Category: AI Agent Design Pattern #116
Generated from actual implementation code
For educational and reference purposes
Leave a Reply