"""
Pattern 218: A/B Testing Pattern
Tests multiple approaches and selects best performer
Implements feedback and improvement mechanisms using LangChain/LangGraph.
"""
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict
class FeedbackType(Enum):
"""Types of feedback."""
POSITIVE = "positive"
NEGATIVE = "negative"
CONSTRUCTIVE = "constructive"
NEUTRAL = "neutral"
@dataclass
class FeedbackItem:
"""Represents a feedback item."""
source: str
content: str
feedback_type: FeedbackType
score: float = 0.0 # 0.0 to 1.0
timestamp: datetime = field(default_factory=datetime.now)
@dataclass
class EvaluationResult:
"""Result of evaluation."""
score: float
feedback: List[FeedbackItem]
improvements: List[str]
timestamp: datetime = field(default_factory=datetime.now)
class A/btestingPattern:
"""
A/B Testing Pattern - Tests multiple approaches and selects best performer
Provides a/b_testing capabilities for continuous improvement
and quality enhancement in agentic systems.
"""
def __init__(self, model_name: str = "gpt-4"):
self.llm = ChatOpenAI(model=model_name, temperature=0.3)
self.history: List[EvaluationResult] = []
self.feedback_log: List[FeedbackItem] = []
def evaluate(self, output: str, criteria: List[str]) -> EvaluationResult:
"""Evaluate output against criteria."""
prompt = ChatPromptTemplate.from_messages([
("system", """You are an expert evaluator for A/B Testing.
Evaluate the output against the criteria and provide:
1. Overall score (0.0 to 1.0)
2. Specific feedback points
3. Improvement suggestions"""),
("human", """Output to evaluate:
{output}
Criteria:
{criteria}
Provide evaluation.""")
])
chain = prompt | self.llm
response = chain.invoke({
"output": output,
"criteria": "\n".join(f"- {c}" for c in criteria)
})
content = str(response.content)
# Parse response
score = 0.75 # Default score
if "score:" in content.lower():
try:
score_text = content.lower().split("score:")[1].split("\n")[0].strip()
score = float(score_text.split()[0].replace("/1.0", "").replace(":", ""))
except:
pass
feedback_items = [
FeedbackItem(
source=self.__class__.__name__,
content=content,
feedback_type=FeedbackType.CONSTRUCTIVE,
score=score
)
]
improvements = [
line.strip("- ").strip()
for line in content.split("\n")
if line.strip().startswith("-") or "improve" in line.lower()
][:3]
result = EvaluationResult(
score=score,
feedback=feedback_items,
improvements=improvements
)
self.history.append(result)
self.feedback_log.extend(feedback_items)
return result
def provide_feedback(self, output: str) -> Dict[str, Any]:
"""Provide feedback on output."""
prompt = ChatPromptTemplate.from_messages([
("system", "You are providing constructive feedback using A/B Testing pattern."),
("human", "{output}")
])
chain = prompt | self.llm
response = chain.invoke({"output": output})
feedback = FeedbackItem(
source="feedback_provider",
content=str(response.content),
feedback_type=FeedbackType.CONSTRUCTIVE,
score=0.8
)
self.feedback_log.append(feedback)
return {
"feedback": feedback.content,
"score": feedback.score,
"type": feedback.feedback_type.value
}
def run(self, output: str, criteria: Optional[List[str]] = None) -> Dict[str, Any]:
"""Main execution method."""
if criteria is None:
criteria = ["clarity", "accuracy", "completeness"]
result = self.evaluate(output, criteria)
return {
"pattern": "A/B Testing",
"score": result.score,
"feedback_count": len(result.feedback),
"improvements": result.improvements,
"total_evaluations": len(self.history)
}
# Example usage
if __name__ == "__main__":
print("=== A/B Testing Pattern Demo ===\n")
pattern = A/btestingPattern()
# Example evaluation
test_output = "This is a sample output that demonstrates the a/b_testing functionality."
result = pattern.run(test_output, ["clarity", "usefulness", "accuracy"])
print(f"Pattern: {result['pattern']}")
print(f"Score: {result['score']:.2f}")
print(f"Improvements suggested: {len(result['improvements'])}")
if result['improvements']:
print("\nSuggestions:")
for i, imp in enumerate(result['improvements'][:3], 1):
print(f" {i}. {imp}")
print(f"\nTotal evaluations: {result['total_evaluations']}")
print(f"Total feedback items: {len(pattern.feedback_log)}")
Agentic Architecture Pattern: Ab Testing
Pattern 218: 218 Ab Testing – Line-by-Line Explanation
Overview
Pattern 218: A/B Testing Pattern
Tests multiple approaches and selects best performer Implements feedback and improvement mechanisms using LangChain/LangGraph.
Code Structure Analysis
File Information
- Pattern Number: 218
- Pattern Name: 218 Ab Testing
- File Path:
patterns\feedback_improvement\218_ab_testing.py - Total Lines: 179
Detailed Code Explanation
Module Docstring
"""
Pattern 218: A/B Testing Pattern
Tests multiple approaches and selects best performer
Implements feedback and improvement mechanisms using LangChain/LangGraph.
"""
Explanation: This module docstring describes the purpose and key features of the 218 Ab Testing 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_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict # 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_core.prompts import ChatPromptTemplate: LangChain framework integration
– from langchain_openai import ChatOpenAI: LangChain framework integration
– from langgraph.graph import StateGraph, END: LangGraph for workflow orchestration
– from typing_extensions import TypedDict: Type hints for better code documentation
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
# Usage example not available
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
Leave a Reply