"""
Pattern 141: Lazy Evaluation
Description:
The Lazy Evaluation pattern defers computation until results are actually needed,
optimizing resource usage by avoiding unnecessary work. Instead of eagerly computing
all results upfront, lazy evaluation creates computation plans that execute only
when values are requested. This pattern is particularly valuable for expensive
operations, large datasets, and scenarios where not all computed values may be used.
Lazy evaluation improves performance by reducing wasted computation, lowering
memory usage, enabling infinite sequences, and allowing for optimization opportunities
through computation graph analysis. It's especially useful in AI agents that might
explore multiple paths but only need results from successful branches, or when
processing large datasets where early termination is possible.
This implementation provides lazy evaluation capabilities for LangChain agents,
including deferred chain execution, lazy data loading, conditional computation,
result streaming, and computation graph optimization. It supports both simple
delayed execution and complex multi-stage lazy pipelines.
Components:
- Deferred Execution: Delay computation until needed
- Lazy Data Loading: Load data on-demand
- Conditional Computation: Skip unnecessary branches
- Result Streaming: Process results incrementally
- Graph Optimization: Optimize computation plans
- Caching Integration: Cache lazy results
Use Cases:
- Large dataset processing
- Multi-path exploration
- Conditional workflows
- Resource-constrained environments
- Expensive API calls
- Database query optimization
- Streaming data processing
- Early termination scenarios
LangChain Implementation:
This implementation uses:
- Python generators for lazy iteration
- Deferred chain execution with LCEL
- Lazy data loading with document loaders
- Conditional branching in chains
- Streaming for incremental results
- Custom lazy wrappers for expensive operations
Benefits:
- Reduces unnecessary computation
- Lowers memory usage
- Enables early termination
- Improves responsiveness
- Supports infinite sequences
- Allows optimization opportunities
- Reduces API costs
Trade-offs:
- Can make debugging harder
- May cause unexpected delays
- Complexity in error handling
- Potential memory leaks if not careful
- Harder to reason about execution flow
- May hide performance issues
Production Considerations:
- Profile to identify expensive operations
- Use for truly expensive computations
- Combine with caching
- Handle errors in lazy contexts
- Set timeouts for lazy operations
- Monitor actual execution patterns
- Document lazy behavior clearly
- Test edge cases thoroughly
- Consider memory management
- Balance with eager evaluation
- Use generators appropriately
- Profile actual vs potential savings
"""
import os
import time
from typing import List, Dict, Any, Optional, Iterator, Callable, Generator
from datetime import datetime
from dataclasses import dataclass
from functools import wraps
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()
@dataclass
class LazyResult:
"""Wrapper for lazy computation result."""
computation: Callable[[], Any]
_cached_value: Optional[Any] = None
_computed: bool = False
_computation_time: Optional[float] = None
def compute(self) -> Any:
"""Execute the lazy computation."""
if not self._computed:
start = time.time()
self._cached_value = self.computation()
self._computation_time = time.time() - start
self._computed = True
return self._cached_value
@property
def value(self) -> Any:
"""Get the computed value."""
return self.compute()
@property
def is_computed(self) -> bool:
"""Check if value has been computed."""
return self._computed
@property
def computation_time(self) -> Optional[float]:
"""Get computation time in seconds."""
return self._computation_time
def lazy(func: Callable) -> Callable:
"""Decorator to make a function lazy."""
@wraps(func)
def wrapper(*args, **kwargs):
def computation():
return func(*args, **kwargs)
return LazyResult(computation=computation)
return wrapper
class LazyChain:
"""Lazy evaluation wrapper for LangChain chains."""
def __init__(self, chain):
"""Initialize lazy chain."""
self.chain = chain
self._result = None
self._computed = False
def compute(self, input_data: Dict[str, Any]) -> Any:
"""Execute the chain."""
if not self._computed:
self._result = self.chain.invoke(input_data)
self._computed = True
return self._result
def stream(self, input_data: Dict[str, Any]) -> Iterator[Any]:
"""Stream results lazily."""
for chunk in self.chain.stream(input_data):
yield chunk
class LazyDataLoader:
"""Lazy data loader that loads data on demand."""
def __init__(self, data_source: str):
"""Initialize lazy data loader."""
self.data_source = data_source
self._data = None
self._loaded = False
self._load_time = None
def load(self) -> List[Any]:
"""Load data (simulated)."""
if not self._loaded:
start = time.time()
# Simulate expensive data loading
time.sleep(0.1)
self._data = [f"Item {i} from {self.data_source}" for i in range(100)]
self._load_time = time.time() - start
self._loaded = True
return self._data
def iter_lazy(self, chunk_size: int = 10) -> Generator[List[Any], None, None]:
"""Iterate over data in chunks."""
data = self.load()
for i in range(0, len(data), chunk_size):
yield data[i:i + chunk_size]
@property
def is_loaded(self) -> bool:
"""Check if data is loaded."""
return self._loaded
class LazyEvaluationAgent:
"""
Agent that uses lazy evaluation for efficient processing.
This agent defers expensive computations until results are needed,
optimizing resource usage and enabling early termination.
"""
def __init__(self, temperature: float = 0.3):
"""Initialize the lazy evaluation agent."""
self.llm = ChatOpenAI(temperature=temperature, model="gpt-3.5-turbo")
self.computation_stats = {
"deferred": 0,
"executed": 0,
"saved": 0
}
# Create lazy chains
self.analysis_chain = LazyChain(
ChatPromptTemplate.from_template(
"Analyze this text: {text}\n\nProvide a detailed analysis."
) | self.llm | StrOutputParser()
)
@lazy
def expensive_computation(self, value: int) -> int:
"""Simulate expensive computation."""
time.sleep(0.5) # Simulate work
return value ** 2
def lazy_filter(
self,
items: List[Any],
condition: Callable[[Any], bool]
) -> Generator[Any, None, None]:
"""Lazy filter that yields items matching condition."""
for item in items:
if condition(item):
yield item
def lazy_map(
self,
items: List[Any],
transform: Callable[[Any], Any]
) -> Generator[Any, None, None]:
"""Lazy map that transforms items on demand."""
for item in items:
yield transform(item)
def conditional_computation(
self,
condition: bool,
true_computation: Callable[[], Any],
false_computation: Callable[[], Any]
) -> LazyResult:
"""Conditionally defer computation."""
if condition:
return LazyResult(computation=true_computation)
else:
return LazyResult(computation=false_computation)
def lazy_pipeline(
self,
data: List[Any],
operations: List[Callable[[Any], Any]],
early_exit: Optional[Callable[[Any], bool]] = None
) -> Generator[Any, None, None]:
"""
Execute a lazy pipeline with early exit capability.
Args:
data: Input data
operations: List of operations to apply
early_exit: Optional function to check if we should stop early
"""
for item in data:
result = item
# Apply operations lazily
for op in operations:
result = op(result)
# Check early exit condition
if early_exit and early_exit(result):
yield result
return # Early termination
yield result
def parallel_lazy_evaluation(
self,
tasks: List[Callable[[], Any]],
max_results: Optional[int] = None
) -> Generator[Any, None, None]:
"""
Evaluate multiple lazy tasks, stopping when enough results obtained.
Args:
tasks: List of lazy computations
max_results: Maximum number of results to compute
"""
count = 0
for task in tasks:
if max_results and count >= max_results:
break
result = LazyResult(computation=task)
yield result
count += 1
def lazy_llm_batch(
self,
prompts: List[str],
process_until: Optional[Callable[[str], bool]] = None
) -> Generator[str, None, None]:
"""
Process prompts lazily, with optional early termination.
Args:
prompts: List of prompts to process
process_until: Optional function to check if we should stop
"""
for prompt in prompts:
result = self.llm.invoke(prompt).content
yield result
if process_until and process_until(result):
break # Early termination
def get_stats(self) -> Dict[str, Any]:
"""Get computation statistics."""
total = self.computation_stats["executed"] + self.computation_stats["saved"]
return {
"deferred": self.computation_stats["deferred"],
"executed": self.computation_stats["executed"],
"saved": self.computation_stats["saved"],
"efficiency": (
self.computation_stats["saved"] / total * 100
if total > 0 else 0
)
}
def demonstrate_lazy_evaluation():
"""Demonstrate the lazy evaluation pattern."""
print("=" * 80)
print("Lazy Evaluation Pattern Demonstration")
print("=" * 80)
agent = LazyEvaluationAgent()
# Demonstration 1: Basic Lazy Computation
print("\n" + "=" * 80)
print("Demonstration 1: Basic Lazy Computation")
print("=" * 80)
print("\nCreating lazy computation...")
lazy_result = agent.expensive_computation(10)
print(f"Computation created (not executed yet)")
print(f"Is computed: {lazy_result.is_computed}")
print("\nAccessing value (triggers computation)...")
start = time.time()
value = lazy_result.value
elapsed = time.time() - start
print(f"Result: {value}")
print(f"Time: {elapsed:.3f}s")
print(f"Is computed: {lazy_result.is_computed}")
print("\nAccessing value again (uses cached)...")
start = time.time()
value2 = lazy_result.value
elapsed = time.time() - start
print(f"Result: {value2}")
print(f"Time: {elapsed:.3f}s (cached)")
# Demonstration 2: Lazy Filter
print("\n" + "=" * 80)
print("Demonstration 2: Lazy Filter (Early Termination)")
print("=" * 80)
numbers = list(range(1, 101))
print(f"Processing {len(numbers)} numbers...")
# Eager evaluation
start = time.time()
eager_result = [x for x in numbers if x % 7 == 0][:5] # Get first 5
eager_time = time.time() - start
print(f"\nEager evaluation: {eager_result}")
print(f"Time: {eager_time:.4f}s (processed all {len(numbers)} items)")
# Lazy evaluation with early termination
start = time.time()
lazy_filtered = agent.lazy_filter(numbers, lambda x: x % 7 == 0)
lazy_result = []
for item in lazy_filtered:
lazy_result.append(item)
if len(lazy_result) >= 5:
break # Early termination
lazy_time = time.time() - start
print(f"\nLazy evaluation: {lazy_result}")
print(f"Time: {lazy_time:.4f}s (stopped after finding 5)")
# Demonstration 3: Lazy Data Loading
print("\n" + "=" * 80)
print("Demonstration 3: Lazy Data Loading")
print("=" * 80)
loader = LazyDataLoader("database")
print(f"Data loader created")
print(f"Is loaded: {loader.is_loaded}")
print("\nProcessing first chunk only...")
for chunk in loader.iter_lazy(chunk_size=10):
print(f"Processing chunk of {len(chunk)} items")
# Only process first chunk
break
print(f"Is loaded: {loader.is_loaded} (only loaded what was needed)")
# Demonstration 4: Conditional Lazy Computation
print("\n" + "=" * 80)
print("Demonstration 4: Conditional Lazy Computation")
print("=" * 80)
def expensive_path_a():
time.sleep(0.3)
return "Result from expensive path A"
def expensive_path_b():
time.sleep(0.3)
return "Result from expensive path B"
condition = True
print(f"Condition: {condition}")
print("Creating conditional lazy computation...")
result = agent.conditional_computation(
condition,
expensive_path_a,
expensive_path_b
)
print(f"Computation deferred, is computed: {result.is_computed}")
print("\nNow computing...")
value = result.value
print(f"Result: {value}")
print(f"Computation time: {result.computation_time:.3f}s")
# Demonstration 5: Lazy Pipeline with Early Exit
print("\n" + "=" * 80)
print("Demonstration 5: Lazy Pipeline with Early Exit")
print("=" * 80)
data = list(range(1, 21))
operations = [
lambda x: x * 2,
lambda x: x + 10,
lambda x: x ** 2
]
early_exit = lambda x: x > 1000
print(f"Processing {len(data)} items through pipeline...")
print("Early exit when result > 1000")
results = list(agent.lazy_pipeline(data, operations, early_exit))
print(f"\nProcessed {len(results)} items before early exit")
print(f"Results: {results}")
# Demonstration 6: Lazy Map Transformation
print("\n" + "=" * 80)
print("Demonstration 6: Lazy Map (Process Only What's Needed)")
print("=" * 80)
items = ["apple", "banana", "cherry", "date", "elderberry"]
def expensive_transform(item: str) -> str:
"""Simulate expensive transformation."""
time.sleep(0.1)
return item.upper()
print(f"Items to transform: {items}")
print("Transforming lazily, taking only first 3...")
start = time.time()
transformed = agent.lazy_map(items, expensive_transform)
result = []
for item in transformed:
result.append(item)
if len(result) >= 3:
break
elapsed = time.time() - start
print(f"Transformed: {result}")
print(f"Time: {elapsed:.2f}s (only transformed 3 items, not all {len(items)})")
# Demonstration 7: Parallel Lazy Tasks with Limit
print("\n" + "=" * 80)
print("Demonstration 7: Parallel Lazy Tasks (Limited Execution)")
print("=" * 80)
tasks = [
lambda: f"Task 1 result",
lambda: f"Task 2 result",
lambda: f"Task 3 result",
lambda: f"Task 4 result",
lambda: f"Task 5 result",
]
print(f"Created {len(tasks)} tasks")
print("Executing only first 2 tasks...")
lazy_results = agent.parallel_lazy_evaluation(tasks, max_results=2)
executed = []
for lazy_result in lazy_results:
executed.append(lazy_result.value)
print(f"Executed: {len(executed)} tasks")
print(f"Results: {executed}")
print(f"Saved: {len(tasks) - len(executed)} task executions")
# Demonstration 8: Lazy LLM Batch Processing
print("\n" + "=" * 80)
print("Demonstration 8: Lazy LLM Batch (Early Termination)")
print("=" * 80)
prompts = [
"Count to 3",
"Count to 5",
"Count to 10",
"Count to 20"
]
print(f"Created {len(prompts)} prompts")
print("Processing until we get a result with '10' in it...")
processed = 0
found_result = None
for result in agent.lazy_llm_batch(
prompts,
process_until=lambda r: "10" in r
):
processed += 1
if "10" in result:
found_result = result
break
print(f"\nProcessed: {processed}/{len(prompts)} prompts")
print(f"Found result: {found_result[:50]}...")
print(f"Saved: {len(prompts) - processed} LLM calls")
# Summary
print("\n" + "=" * 80)
print("Summary: Lazy Evaluation Pattern")
print("=" * 80)
print("""
The Lazy Evaluation pattern optimizes resource usage by deferring computation:
Key Features Demonstrated:
1. Basic Lazy Computation - Defer expensive operations until needed
2. Lazy Filter - Filter with early termination
3. Lazy Data Loading - Load data on-demand
4. Conditional Computation - Only compute chosen branch
5. Lazy Pipeline - Process with early exit
6. Lazy Map - Transform only what's needed
7. Limited Execution - Execute only required tasks
8. LLM Batch Processing - Stop when condition met
Benefits:
- Reduces unnecessary computation
- Lowers memory usage
- Enables early termination
- Improves responsiveness
- Reduces API costs
- Supports infinite sequences
- Allows optimization
Best Practices:
- Use for expensive operations
- Combine with caching
- Handle errors properly
- Set timeouts
- Monitor execution patterns
- Document lazy behavior
- Test edge cases
- Balance with eager evaluation
- Use generators appropriately
- Profile actual savings
Common Use Cases:
- Large dataset processing
- Multi-path exploration
- Conditional workflows
- Resource-constrained environments
- Expensive API calls
- Database queries
- Streaming data
- Early termination scenarios
This pattern is essential for efficient AI agents that need to optimize
resource usage and avoid unnecessary computation.
""")
if __name__ == "__main__":
demonstrate_lazy_evaluation()
Agentic Design Pattern: Lazy Evaluation
Pattern 141: Lazy Evaluation – Line-by-Line Explanation
Overview
This document provides a detailed line-by-line explanation of the Lazy Evaluation pattern implementation.
Pattern Description
Pattern 141: Lazy Evaluation
Description: The Lazy Evaluation pattern defers computation until results are actually needed, optimizing resource usage by avoiding unnecessary work. Instead of eagerly computing all results upfront, lazy evaluation creates computation plans that execute only when values are requested. This pattern is particularly valuable for expensive operations, large datasets, and scenarios where not all computed values may be used.
Lazy evaluation improves performance by reducing wasted computation, lowering
memory usage, enabling infinite sequences, and allowing for optimization opportunities
through computation graph analysis. It's especially useful in AI agents that might
explore multiple paths but only need results from successful branches, or when
processing large datasets where early termination is possible.
This implementation provides lazy evaluation capabilities for LangChain agents,
including deferred chain execution, lazy data loading, conditional computation,
result streaming, and computation graph optimization. It supports both simple
delayed execution and complex multi-stage lazy pipelines.
Components: – Deferred Execution: Delay computation until needed – Lazy Data Loading: Load data on-demand – Conditional Computation: Skip unnecessary branches – Result Streaming: Process results incrementally – Graph Optimization: Optimize computation plans – Caching Integration: Cache lazy results
Use Cases: – Large dataset processing – Multi-path exploration – Conditional workflows – Resource-constrained environments – Expensive API calls – Database query optimization – Streaming data processing – Early termination scenarios
LangChain Implementation: This implementation uses: – Python generators for lazy iteration – Deferred chain execution with LCEL – Lazy data loading with document loaders – Conditional branching in chains – Streaming for incremental results – Custom lazy wrappers for expensive operations
Benefits: – Reduces unnecessary computation – Lowers memory usage – Enables early termination – Improves responsiveness – Supports infinite sequences – Allows optimization opportunities – Reduces API costs
Trade-offs: – Can make debugging harder – May cause unexpected delays – Complexity in error handling – Potential memory leaks if not careful – Harder to reason about execution flow – May hide performance issues
Production Considerations: – Profile to identify expensive operations – Use for truly expensive computations – Combine with caching – Handle errors in lazy contexts – Set timeouts for lazy operations – Monitor actual execution patterns – Document lazy behavior clearly – Test edge cases thoroughly – Consider memory management – Balance with eager evaluation – Use generators appropriately – Profile actual vs potential savings
File Statistics
- Total Lines: 603
- Classes: 4
- Functions: 2
- Constants: 0
Module Documentation
"""
Pattern 141: Lazy Evaluation
Description:
The Lazy Evaluation pattern defers computation until results are actually needed,
optimizing resource usage by avoiding unnecessary work. Instead of eagerly computing
all results upfront, lazy evaluation creates computation plans that execute only
when values are requested. This pattern is particularly valuable for expensive
operations, large datasets, and scenarios where not all computed values may be used.
Lazy evaluation improves performance by reducing wasted computation, lowering
memory usage, enabling infinite sequences, and allowing for optimization opportunities
through computation graph analysis. It's especially useful in AI agents that might
explore multiple paths but only need results from successful branches, or when
processing large datasets where early termination is possible.
This implementation provides lazy evaluation capabilities for LangChain agents,
including deferred chain execution, lazy data loading, conditional computation,
result streaming, and computation graph optimization. It supports both simple
delayed execution and complex multi-stage lazy pipelines.
Components:
- Deferred Execution: Delay computation until needed
- Lazy Data Loading: Load data on-demand
- Conditional Computation: Skip unnecessary branches
- Result Streaming: Process results incrementally
- Graph Optimization: Optimize computation plans
- Caching Integration: Cache lazy results
Use Cases:
- Large dataset processing
- Multi-path exploration
- Conditional workflows
- Resource-constrained environments
- Expensive API calls
- Database query optimization
- Streaming data processing
- Early termination scenarios
LangChain Implementation:
This implementation uses:
- Python generators for lazy iteration
- Deferred chain execution with LCEL
- Lazy data loading with document loaders
- Conditional branching in chains
- Streaming for incremental results
- Custom lazy wrappers for expensive operations
Benefits:
- Reduces unnecessary computation
- Lowers memory usage
- Enables early termination
- Improves responsiveness
- Supports infinite sequences
- Allows optimization opportunities
- Reduces API costs
Trade-offs:
- Can make debugging harder
- May cause unexpected delays
- Complexity in error handling
- Potential memory leaks if not careful
- Harder to reason about execution flow
- May hide performance issues
Production Considerations:
- Profile to identify expensive operations
- Use for truly expensive computations
- Combine with caching
- Handle errors in lazy contexts
- Set timeouts for lazy operations
- Monitor actual execution patterns
- Document lazy behavior clearly
- Test edge cases thoroughly
- Consider memory management
- Balance with eager evaluation
- Use generators appropriately
- Profile actual vs potential savings
"""
Purpose: This module implements the Lazy Evaluation pattern, providing a structured approach to agent design and implementation.
Import Statements
Standard Library Imports
import dataclasses.dataclass
import datetime.datetime
import functools.wraps
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import time
import typing.Any
import typing.Callable
import typing.Dict
import typing.Generator
import typing.Iterator
import typing.List
import typing.Optional
Purpose: Core Python functionality:
– dataclasses.dataclass: Simplified class definitions with @dataclass
– datetime.datetime: Date and time handling
– functools.wraps: Higher-order functions and operations on callable objects
– langchain_core.output_parsers.StrOutputParser: Regular expression operations
– langchain_core.prompts.ChatPromptTemplate: Regular expression operations
– os: Environment variable access and system operations
– time: Time-related functions
– typing.Any: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.Callable: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.Generator: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.Iterator: 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_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: LazyResult
Lines 97-126
Purpose:
Wrapper for lazy computation result.
Methods:
compute(self)
Lines 104-111
Purpose:
Execute the lazy computation.
Parameters:
Returns: Any
value(self)
Lines 114-116
Decorators: property
Purpose:
Get the computed value.
Parameters:
Returns: Any
is_computed(self)
Lines 119-121
Decorators: property
Purpose:
Check if value has been computed.
Parameters:
Returns: bool
computation_time(self)
Lines 124-126
Decorators: property
Purpose:
Get computation time in seconds.
Parameters:
Returns: <ast.Subscript object at 0x000001A06D169C10>
Class: LazyChain
Lines 139-158
Purpose:
Lazy evaluation wrapper for LangChain chains.
Methods:
__init__(self, chain)
Lines 142-146
Purpose:
Initialize lazy chain.
Parameters:
– chain: Chain
compute(self, input_data)
Lines 148-153
Purpose:
Execute the chain.
Parameters:
– input_data: Input data
Returns: Any
stream(self, input_data)
Lines 155-158
Purpose:
Stream results lazily.
Parameters:
– input_data: Input data
Returns: <ast.Subscript object at 0x000001A06D7B1150>
Class: LazyDataLoader
Lines 161-191
Purpose:
Lazy data loader that loads data on demand.
Methods:
__init__(self, data_source)
Lines 164-169
Purpose:
Initialize lazy data loader.
Parameters:
– data_source: Data source
load(self)
Lines 171-180
Purpose:
Load data (simulated).
Parameters:
Returns: <ast.Subscript object at 0x000001A06D72DDD0>
iter_lazy(self, chunk_size)
Lines 182-186
Purpose:
Iterate over data in chunks.
Parameters:
– chunk_size: Chunk size
Returns: <ast.Subscript object at 0x000001A06D72ED90>
is_loaded(self)
Lines 189-191
Decorators: property
Purpose:
Check if data is loaded.
Parameters:
Returns: bool
Class: LazyEvaluationAgent
Lines 194-334
Purpose:
Agent that uses lazy evaluation for efficient processing.
This agent defers expensive computations until results are needed,
optimizing resource usage and enabling early termination.
Methods:
__init__(self, temperature)
Lines 202-216
Purpose:
Initialize the lazy evaluation agent.
Parameters:
– temperature: Sampling temperature (0-1)
expensive_computation(self, value)
Lines 219-222
Decorators: lazy
Purpose:
Simulate expensive computation.
Parameters:
– value: Value
Returns: int
lazy_filter(self, items, condition)
Lines 224-232
Purpose:
Lazy filter that yields items matching condition.
Parameters:
– items: Items
– condition: Condition
Returns: <ast.Subscript object at 0x000001A06D88E250>
lazy_map(self, items, transform)
Lines 234-241
Purpose:
Lazy map that transforms items on demand.
Parameters:
– items: Items
– transform: Transform
Returns: <ast.Subscript object at 0x000001A06D622A50>
conditional_computation(self, condition, true_computation, false_computation)
Lines 243-253
Purpose:
Conditionally defer computation.
Parameters:
– condition: Condition
– true_computation: True computation
– false_computation: False computation
Returns: LazyResult
lazy_pipeline(self, data, operations, early_exit)
Lines 255-281
Purpose:
Execute a lazy pipeline with early exit capability.
Args:
data: Input data
operations: List of operations to apply
early_exit: Optional function to check if we should stop early
Parameters:
– data: Data
– operations: Operations
– early_exit: Early exit
Returns: <ast.Subscript object at 0x000001A06D77C0D0>
parallel_lazy_evaluation(self, tasks, max_results)
Lines 283-302
Purpose:
Evaluate multiple lazy tasks, stopping when enough results obtained.
Args:
tasks: List of lazy computations
max_results: Maximum number of results to compute
Parameters:
– tasks: Tasks
– max_results: Max results
Returns: <ast.Subscript object at 0x000001A06D865010>
lazy_llm_batch(self, prompts, process_until)
Lines 304-321
Purpose:
Process prompts lazily, with optional early termination.
Args:
prompts: List of prompts to process
process_until: Optional function to check if we should stop
Parameters:
– prompts: Prompt template or text
– process_until: Process until
Returns: <ast.Subscript object at 0x000001A06D864E50>
get_stats(self)
Lines 323-334
Purpose:
Get computation statistics.
Parameters:
Returns: <ast.Subscript object at 0x000001A06D811510>
Functions
lazy(func)
Lines 129-136
Purpose:
Decorator to make a function lazy.
Parameters:
– func: Func
Returns: Callable
demonstrate_lazy_evaluation()
Lines 337-598
Purpose:
Demonstrate the lazy evaluation pattern.
Key Concepts Summary
Pattern Overview
The Lazy Evaluation 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 140
- Pattern 142
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 Lazy Evaluation 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: 141_lazy_evaluation.py
Total Lines: 603
Pattern Category: AI Agent Design Pattern #141
Generated from actual implementation code
For educational and reference purposes
Leave a Reply