"""
Pattern 114: Online Learning
Description:
Online Learning enables agents to continuously learn and adapt from incoming
data streams in real-time. Unlike batch learning, online learning updates
the agent's knowledge incrementally as new data arrives, allowing adaptation
to changing environments and concept drift.
This pattern is crucial for agents operating in dynamic environments where
patterns, user preferences, or optimal strategies change over time. It
combines incremental updates, drift detection, and continuous adaptation.
Key Components:
1. Stream Processor: Handles incoming data streams
2. Incremental Updater: Updates knowledge incrementally
3. Drift Detector: Identifies when patterns change
4. Performance Monitor: Tracks learning effectiveness
5. Model Adapter: Adjusts model based on feedback
6. Forgetting Mechanism: Manages outdated knowledge
Learning Strategies:
- Incremental Learning: Update with each new example
- Mini-Batch: Update with small batches
- Windowed: Learn from recent window
- Adaptive Rate: Adjust learning rate dynamically
- Ensemble: Maintain multiple learners
Use Cases:
- Real-time recommendation systems
- Adaptive chatbots learning from conversations
- Fraud detection with evolving patterns
- Personalization engines
- Market prediction systems
LangChain Implementation:
Uses conversation memory, feedback integration, and dynamic prompt
adjustment to enable continuous learning from interactions.
"""
import os
from typing import List, Dict, Any, Optional, Deque
from collections import deque
from dotenv import load_dotenv
from datetime import datetime
import json
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain.memory import ConversationBufferMemory
load_dotenv()
class ExampleBuffer:
"""Manages a buffer of learning examples"""
def __init__(self, max_size: int = 100):
self.max_size = max_size
self.examples: Deque[Dict[str, Any]] = deque(maxlen=max_size)
self.performance_history: List[float] = []
def add_example(self, input_text: str, output_text: str,
feedback_score: Optional[float] = None):
"""Add a new example to the buffer"""
example = {
"input": input_text,
"output": output_text,
"feedback": feedback_score,
"timestamp": datetime.now().isoformat()
}
self.examples.append(example)
if feedback_score is not None:
self.performance_history.append(feedback_score)
def get_recent_examples(self, n: int = 5) -> List[Dict[str, str]]:
"""Get n most recent examples"""
recent = list(self.examples)[-n:]
return [{"input": ex["input"], "output": ex["output"]} for ex in recent]
def get_best_examples(self, n: int = 5) -> List[Dict[str, str]]:
"""Get n best-rated examples"""
if not self.examples:
return []
# Filter examples with feedback
rated = [ex for ex in self.examples if ex["feedback"] is not None]
if not rated:
return self.get_recent_examples(n)
# Sort by feedback score
sorted_examples = sorted(rated, key=lambda x: x["feedback"], reverse=True)
top_n = sorted_examples[:n]
return [{"input": ex["input"], "output": ex["output"]} for ex in top_n]
def detect_drift(self, window_size: int = 20) -> bool:
"""Detect if performance is degrading (concept drift)"""
if len(self.performance_history) drift_threshold
def get_performance_trend(self) -> str:
"""Get description of performance trend"""
if len(self.performance_history) = 20 else []
recent_avg = sum(recent) / len(recent)
if not older:
return f"recent_avg: {recent_avg:.2f}"
older_avg = sum(older) / len(older)
if recent_avg > older_avg + 0.1:
return f"improving (from {older_avg:.2f} to {recent_avg:.2f})"
elif recent_avg ChatPromptTemplate:
"""Create prompt with dynamic examples from learning buffer"""
# Get examples from buffer
if use_best:
examples = self.example_buffer.get_best_examples(n=3)
else:
examples = self.example_buffer.get_recent_examples(n=3)
if not examples:
# No examples yet, use base prompt
return ChatPromptTemplate.from_messages([
("system", self.base_system_prompt),
("human", "{input}")
])
# Create few-shot prompt with learned examples
few_shot_prompt = FewShotChatMessagePromptTemplate(
example_prompt=self.example_prompt,
examples=examples
)
# Get performance trend
trend = self.example_buffer.get_performance_trend()
system_msg = f"""{self.base_system_prompt}
Learning Status: {len(self.example_buffer.examples)} examples learned
Performance Trend: {trend}
Learn from these example interactions:"""
return ChatPromptTemplate.from_messages([
("system", system_msg),
few_shot_prompt,
("human", "{input}")
])
def respond(self, user_input: str) -> str:
"""Generate response using current learned knowledge"""
# Create dynamic prompt with learned examples
prompt = self.create_dynamic_prompt(use_best=True)
# Generate response
chain = prompt | self.llm | StrOutputParser()
response = chain.invoke({"input": user_input})
self.total_interactions += 1
return response
def learn_from_feedback(self, user_input: str, response: str,
feedback_score: float):
"""Learn from user feedback on a response"""
print(f"\nLearning from feedback (score: {feedback_score})...")
# Add to example buffer
self.example_buffer.add_example(user_input, response, feedback_score)
# Check for drift
if self.example_buffer.detect_drift():
print("⚠️ Concept drift detected! Performance declining.")
print("Consider adjusting learning strategy or model.")
# Show learning progress
trend = self.example_buffer.get_performance_trend()
print(f"Performance trend: {trend}")
def interact_and_learn(self, user_input: str, feedback_score: Optional[float] = None) -> Dict[str, Any]:
"""Complete interaction cycle with learning"""
print(f"\n{'='*60}")
print(f"User: {user_input}")
print(f"{'='*60}")
# Generate response
response = self.respond(user_input)
print(f"\nAssistant: {response}")
# Learn from feedback if provided
if feedback_score is not None:
self.learn_from_feedback(user_input, response, feedback_score)
return {
"input": user_input,
"response": response,
"feedback": feedback_score,
"total_interactions": self.total_interactions,
"examples_learned": len(self.example_buffer.examples),
"performance_trend": self.example_buffer.get_performance_trend()
}
def get_learning_stats(self) -> Dict[str, Any]:
"""Get statistics about learning progress"""
return {
"total_interactions": self.total_interactions,
"examples_in_buffer": len(self.example_buffer.examples),
"performance_history_size": len(self.example_buffer.performance_history),
"performance_trend": self.example_buffer.get_performance_trend(),
"drift_detected": self.example_buffer.detect_drift()
}
def demonstrate_online_learning():
"""Demonstrate online learning pattern"""
print("\n" + "="*70)
print("ONLINE LEARNING DEMONSTRATION")
print("="*70)
# Create agent for customer support domain
agent = OnlineLearningAgent(domain="customer support")
# Example 1: Initial interactions (building knowledge)
print("\n" + "="*70)
print("Example 1: Initial Learning Phase")
print("="*70)
interactions = [
("How do I reset my password?", 0.9),
("Where can I find my order history?", 0.85),
("How do I contact support?", 0.95),
("What is your return policy?", 0.8),
("How long does shipping take?", 0.9),
]
for user_input, feedback in interactions:
agent.interact_and_learn(user_input, feedback)
# Show learning stats
print("\n" + "-"*60)
print("Learning Statistics After Initial Phase:")
stats = agent.get_learning_stats()
print(json.dumps(stats, indent=2))
# Example 2: Applying learned knowledge
print("\n" + "="*70)
print("Example 2: Applying Learned Knowledge")
print("="*70)
new_queries = [
"I forgot my password, what should I do?", # Similar to learned
"How can I see my past orders?", # Similar to learned
]
for query in new_queries:
result = agent.interact_and_learn(query, feedback_score=0.9)
# Example 3: Continuous learning with feedback
print("\n" + "="*70)
print("Example 3: Continuous Learning with Mixed Feedback")
print("="*70)
mixed_interactions = [
("Can I change my delivery address?", 0.7), # Medium feedback
("Do you offer gift wrapping?", 0.85), # Good feedback
("How do I cancel my subscription?", 0.6), # Lower feedback
("What payment methods do you accept?", 0.9), # High feedback
]
for user_input, feedback in mixed_interactions:
agent.interact_and_learn(user_input, feedback)
# Example 4: Detecting concept drift
print("\n" + "="*70)
print("Example 4: Simulating Performance Degradation (Drift)")
print("="*70)
# Simulate declining performance
drift_interactions = [
("Question about new feature X", 0.4), # New topics, poor performance
("Tell me about feature Y", 0.3),
("How does feature Z work?", 0.35),
("Feature A questions", 0.4),
("Feature B support", 0.3),
]
for user_input, feedback in drift_interactions:
agent.interact_and_learn(user_input, feedback)
# Final statistics
print("\n" + "="*70)
print("Final Learning Statistics:")
print("="*70)
final_stats = agent.get_learning_stats()
print(json.dumps(final_stats, indent=2))
# Summary
print("\n" + "="*70)
print("SUMMARY")
print("="*70)
print("""
Online Learning Pattern demonstrated:
Key Features:
1. Incremental Learning - Updates knowledge with each interaction
2. Example Buffer - Maintains recent and best examples
3. Performance Tracking - Monitors learning effectiveness
4. Drift Detection - Identifies when patterns change
5. Dynamic Adaptation - Adjusts prompts based on learned examples
Learning Mechanisms:
- Few-shot learning with best examples
- Continuous feedback integration
- Performance trend analysis
- Concept drift detection
- Adaptive example selection
Applications:
- Customer support bots improving from interactions
- Personalized assistants learning user preferences
- Recommendation systems adapting to trends
- Fraud detection with evolving patterns
- Adaptive tutoring systems
Benefits:
- Continuous improvement without retraining
- Adaptation to changing patterns
- Personalization based on interaction history
- Early warning for performance degradation
- Efficient learning from limited examples
Limitations:
- Requires consistent feedback
- May need periodic full retraining
- Can accumulate biases from feedback
- Limited by context window size
""")
if __name__ == "__main__":
demonstrate_online_learning()
Agentic Design Pattern: Online Learning
Pattern 114: Online Learning – Line-by-Line Explanation
Overview
This document provides a detailed line-by-line explanation of the Online Learning pattern implementation.
Pattern Description
Pattern 114: Online Learning
Description: Online Learning enables agents to continuously learn and adapt from incoming data streams in real-time. Unlike batch learning, online learning updates the agent’s knowledge incrementally as new data arrives, allowing adaptation to changing environments and concept drift.
This pattern is crucial for agents operating in dynamic environments where
patterns, user preferences, or optimal strategies change over time. It
combines incremental updates, drift detection, and continuous adaptation.
Key Components: 1. Stream Processor: Handles incoming data streams 2. Incremental Updater: Updates knowledge incrementally 3. Drift Detector: Identifies when patterns change 4. Performance Monitor: Tracks learning effectiveness 5. Model Adapter: Adjusts model based on feedback 6. Forgetting Mechanism: Manages outdated knowledge
Learning Strategies: – Incremental Learning: Update with each new example – Mini-Batch: Update with small batches – Windowed: Learn from recent window – Adaptive Rate: Adjust learning rate dynamically – Ensemble: Maintain multiple learners
Use Cases: – Real-time recommendation systems – Adaptive chatbots learning from conversations – Fraud detection with evolving patterns – Personalization engines – Market prediction systems
LangChain Implementation: Uses conversation memory, feedback integration, and dynamic prompt adjustment to enable continuous learning from interactions.
File Statistics
- Total Lines: 385
- Classes: 2
- Functions: 1
- Constants: 0
Module Documentation
"""
Pattern 114: Online Learning
Description:
Online Learning enables agents to continuously learn and adapt from incoming
data streams in real-time. Unlike batch learning, online learning updates
the agent's knowledge incrementally as new data arrives, allowing adaptation
to changing environments and concept drift.
This pattern is crucial for agents operating in dynamic environments where
patterns, user preferences, or optimal strategies change over time. It
combines incremental updates, drift detection, and continuous adaptation.
Key Components:
1. Stream Processor: Handles incoming data streams
2. Incremental Updater: Updates knowledge incrementally
3. Drift Detector: Identifies when patterns change
4. Performance Monitor: Tracks learning effectiveness
5. Model Adapter: Adjusts model based on feedback
6. Forgetting Mechanism: Manages outdated knowledge
Learning Strategies:
- Incremental Learning: Update with each new example
- Mini-Batch: Update with small batches
- Windowed: Learn from recent window
- Adaptive Rate: Adjust learning rate dynamically
- Ensemble: Maintain multiple learners
Use Cases:
- Real-time recommendation systems
- Adaptive chatbots learning from conversations
- Fraud detection with evolving patterns
- Personalization engines
- Market prediction systems
LangChain Implementation:
Uses conversation memory, feedback integration, and dynamic prompt
adjustment to enable continuous learning from interactions.
"""
Purpose: This module implements the Online Learning pattern, providing a structured approach to agent design and implementation.
Import Statements
Standard Library Imports
import collections.deque
import datetime.datetime
import json
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import langchain_core.prompts.FewShotChatMessagePromptTemplate
import os
import typing.Any
import typing.Deque
import typing.Dict
import typing.List
import typing.Optional
Purpose: Core Python functionality:
– collections.deque: Specialized container datatypes
– datetime.datetime: Date and time handling
– json: JSON serialization/deserialization
– langchain_core.output_parsers.StrOutputParser: Regular expression operations
– langchain_core.prompts.ChatPromptTemplate: Regular expression operations
– langchain_core.prompts.FewShotChatMessagePromptTemplate: Regular expression operations
– os: Environment variable access and system operations
– typing.Any: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.Deque: 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)
– typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional)
LangChain Imports
from langchain.memory.ConversationBufferMemory
from langchain_openai.ChatOpenAI
Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows – Memory: State management across interactions
Other Imports
from dotenv.load_dotenv
Purpose: Additional dependencies for specialized functionality.
Classes
Class: ExampleBuffer
Lines 56-135
Purpose:
Manages a buffer of learning examples
Methods:
__init__(self, max_size)
Lines 59-62
Purpose: Initialize the class instance with required parameters.
Parameters:
– max_size: Max size
add_example(self, input_text, output_text, feedback_score)
Lines 64-76
Purpose:
Add a new example to the buffer
Parameters:
– input_text: Input data
– output_text: Output text
– feedback_score: Feedback score
get_recent_examples(self, n)
Lines 78-81
Purpose:
Get n most recent examples
Parameters:
– n: N
Returns: <ast.Subscript object at 0x000001A06D88BE90>
get_best_examples(self, n)
Lines 83-97
Purpose:
Get n best-rated examples
Parameters:
– n: N
Returns: <ast.Subscript object at 0x000001A06D765F50>
detect_drift(self, window_size)
Lines 99-113
Purpose:
Detect if performance is degrading (concept drift)
Parameters:
– window_size: Window size
Returns: bool
get_performance_trend(self)
Lines 115-135
Purpose:
Get description of performance trend
Parameters:
Returns: str
Class: OnlineLearningAgent
Lines 138-254
Purpose:
Agent that learns continuously from interactions
Methods:
__init__(self, model_name, domain)
Lines 141-156
Purpose: Initialize the class instance with required parameters.
Parameters:
– model_name: Language model instance
– domain: Domain
create_dynamic_prompt(self, use_best)
Lines 158-192
Purpose:
Create prompt with dynamic examples from learning buffer
Parameters:
– use_best: Use best
Returns: ChatPromptTemplate
respond(self, user_input)
Lines 194-204
Purpose:
Generate response using current learned knowledge
Parameters:
– user_input: Input data
Returns: str
learn_from_feedback(self, user_input, response, feedback_score)
Lines 206-221
Purpose:
Learn from user feedback on a response
Parameters:
– user_input: Input data
– response: Response
– feedback_score: Feedback score
interact_and_learn(self, user_input, feedback_score)
Lines 223-244
Purpose:
Complete interaction cycle with learning
Parameters:
– user_input: Input data
– feedback_score: Feedback score
Returns: <ast.Subscript object at 0x000001A06D16DBD0>
get_learning_stats(self)
Lines 246-254
Purpose:
Get statistics about learning progress
Parameters:
Returns: <ast.Subscript object at 0x000001A06E094F10>
Functions
demonstrate_online_learning()
Lines 257-380
Purpose:
Demonstrate online learning pattern
Key Concepts Summary
Pattern Overview
The Online 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
- Memory: Conversation state management
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 113
- Pattern 115
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 Online 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: 114_online_learning.py
Total Lines: 385
Pattern Category: AI Agent Design Pattern #114
Generated from actual implementation code
For educational and reference purposes
Leave a Reply