FILTER_SYSTEM_IMPLEMENTATION.md

Semantic Kernel-Style Filter/Hook System Implementation

Overview

This implementation adds a comprehensive observability and middleware system to agentic-brain, inspired by Microsoft's Semantic Kernel. The filter/hook system enables:

• Pre/Post-Invoke Hooks: Intercept LLM calls before and after execution
• Filter Chain Pattern: Compose multiple filters for modular observability
• Built-in Filters: Ready-to-use implementations for common patterns
• Extensibility: Easy to create custom filters for domain-specific needs
• Integration: Works seamlessly with ProviderFallbackChain

Architecture

Components

1. FilterContext (src/agentic_brain/hooks/filters.py)

Core data structure that carries information through the filter chain:

• Request identification (request_id, correlation_id)
• Request details (model, prompt, temperature, max_tokens)
• Metadata (expandable for custom data)
• Result and error information
• Execution tracking (attempts, retry_count)

2. Filter Protocol (src/agentic_brain/hooks/filters.py)

Abstract base class defining the filter interface:

class Filter(ABC):
    async def pre_invoke(self, context: FilterContext) -> None
    async def post_invoke(self, context: FilterContext) -> None

3. FilterChain (src/agentic_brain/hooks/filters.py)

Orchestrates filter execution:

• Runs pre-invoke hooks in order
• Executes the LLM call (or returns cached result)
• Runs post-invoke hooks even if errors occur
• Supports sync/async and batch processing
• Tracks which filters have been applied

Built-in Filters

LoggingFilter (src/agentic_brain/hooks/builtin.py)

Logs all LLM requests and responses at configurable levels.

LoggingFilter(log_level=logging.INFO, log_full_response=False)

Features:

• Logs request details: model, prompt length, tokens
• Logs response details: duration, attempts, result length
• Separate logging for errors with full context

MetricsFilter (src/agentic_brain/hooks/builtin.py)

Tracks performance metrics across requests.

MetricsFilter()

Collects:

• Total requests and errors
• Average/total latency
• Token counts (input + output)
• Success rates
• Per-model request/error counts

CacheFilter (src/agentic_brain/hooks/builtin.py)

Caches responses to avoid redundant LLM calls.

CacheFilter(ttl_seconds=3600)

Features:

• Hash-based caching (prompt + model + temperature)
• TTL support for cache expiration
• Cache hit detection
• Prevents actual LLM calls on cache hits

RetryFilter (src/agentic_brain/hooks/builtin.py)

Implements exponential backoff retry logic.

RetryFilter(
    max_retries=3,
    initial_backoff_ms=100.0,
    max_backoff_ms=10000.0,
    exponential_base=2.0
)

Features:

• Exponential backoff calculation
• Configurable retry policies
• Selective error retry (can specify retryable error types)

CostFilter (src/agentic_brain/hooks/builtin.py)

Tracks and limits API costs.

CostFilter(
    budget_usd=100.0,
    pricing_models={...}  # Optional custom pricing
)

Features:

• Built-in pricing for common models (GPT-4, Claude, Llama, etc.)
• Token-based cost calculation
• Budget enforcement
• Cost summary reporting

CompositeFilter (src/agentic_brain/hooks/builtin.py)

Groups multiple filters under a single name.

observability = CompositeFilter(
    "observability",
    [LoggingFilter(), MetricsFilter()]
)

Usage Examples

Basic Usage

from agentic_brain.hooks import FilterChain, FilterContext
from agentic_brain.hooks.builtin import LoggingFilter, MetricsFilter

# Create chain
chain = FilterChain([
    LoggingFilter(),
    MetricsFilter(),
])

# Create context
context = FilterContext(
    model="gpt-4",
    prompt="What is AI?",
)

# Invoke through chain
result = await chain.invoke(context, llm_call_func)

Production Setup

chain = (
    FilterChain()
    .add_filter(LoggingFilter(log_level=logging.INFO))
    .add_filter(MetricsFilter())
    .add_filter(CacheFilter(ttl_seconds=3600))
    .add_filter(CostFilter(budget_usd=1000.0))
    .add_filter(RetryFilter(max_retries=3))
)

With ProviderFallbackChain

from agentic_brain.router.smart_fallback import ProviderFallbackChain

# Create with filters
fallback = ProviderFallbackChain(
    filters=[
        LoggingFilter(),
        MetricsFilter(),
        CacheFilter(),
    ]
)

# Or add later
fallback.add_filter(CostFilter(budget_usd=10.0))
fallback.remove_filter("CacheFilter")

Custom Filter

from agentic_brain.hooks import Filter, FilterContext

class CustomMetricsFilter(Filter):
    @property
    def name(self) -> str:
        return "CustomMetrics"

    async def pre_invoke(self, context: FilterContext) -> None:
        context.metadata["custom_start"] = time.time()

    async def post_invoke(self, context: FilterContext) -> None:
        elapsed = time.time() - context.metadata.get("custom_start", 0)
        print(f"Custom timing: {elapsed:.3f}s")

Files Created

Core Implementation

1. src/agentic_brain/hooks/filters.py (10 KB)

   • FilterContext dataclass
   • Filter protocol (ABC)
   • FilterChain orchestration
   • ~420 lines of production code

2. src/agentic_brain/hooks/builtin.py (16 KB)

   • LoggingFilter
   • MetricsFilter
   • CacheFilter
   • RetryFilter
   • CostFilter
   • CompositeFilter
   • ~520 lines of production code

Documentation

3. src/agentic_brain/hooks/FILTER_SYSTEM.md

   • Comprehensive guide with examples
   • Architecture overview
   • Best practices
   • Integration patterns

4. src/agentic_brain/hooks/__init__.py (updated)

   • Exports all filter classes
   • Backward compatible with existing hooks

Examples

5. examples/filter_system_examples.py (9 KB)
   • 6 complete working examples
   • Basic observability
   • Caching
   • Cost tracking
   • Integration with ProviderFallbackChain
   • Production setup
   • Custom filters

Tests

6. tests/test_hooks_filters.py (17 KB)
   • 28 comprehensive tests
   • 100% passing
   • Coverage:
     • FilterContext functionality
     • Filter protocol implementation
     • FilterChain execution and error handling
     • All built-in filters
     • Integration tests
     • Batch processing
     • Method chaining

Updates

7. src/agentic_brain/router/smart_fallback.py (updated)
   • Added filters parameter to __init__
   • Added methods:
     • add_filter()
     • remove_filter()
     • get_filter()
     • apply_filters_pre()
     • apply_filters_post()

Key Features

1. Composability

Filters can be combined in any order using the FilterChain pattern:

chain = FilterChain([filter1, filter2, filter3])

2. Error Handling

• Pre-invoke errors block execution (fail-fast)
• Post-invoke errors are isolated (won't mask original errors)
• Comprehensive error logging

3. Performance

• Cache filter eliminates redundant calls
• Metrics filter tracks performance
• Exponential backoff prevents thundering herd

4. Cost Control

• Built-in pricing models for major LLM providers
• Budget enforcement with warnings
• Per-model cost tracking

5. Extensibility

• Easy to implement custom filters
• Filters can modify context before/after invocation
• Metadata dictionary for storing custom data

6. Integration

• Works with async/sync code
• Batch processing support
• Seamless ProviderFallbackChain integration

Testing

All tests pass successfully:

pytest tests/test_hooks_filters.py -v
# 28 passed in 0.52s

Test coverage includes:

• FilterContext creation and utilities
• Filter protocol implementation
• FilterChain execution flow
• Error handling (both pre and post-invoke)
• All built-in filters
• Cache TTL expiration
• Cost calculation and tracking
• Metrics aggregation
• Batch processing
• Integration scenarios

Performance Impact

Without Filters

• Direct LLM call execution

With Filters (typical setup)

• LoggingFilter: ~1ms (JSON serialization)
• MetricsFilter: <1ms (counter updates)
• CacheFilter: <1ms on cache hit, eliminates expensive call
• CostFilter: <1ms (pricing lookup)

Caching typically provides 10-100x speedup by eliminating redundant API calls.

Best Practices

1. Order Matters

   # Good: Cache before expensive operations
   FilterChain([LoggingFilter(), CacheFilter(), MetricsFilter()])
   
   # Less optimal: Logging after cache
   FilterChain([CacheFilter(), LoggingFilter()])

2. Monitor Costs

   # Always track spending in production
   cost_filter = CostFilter(budget_usd=100.0)

3. Cache Strategically

   # Short TTL for frequently changing queries
   CacheFilter(ttl_seconds=300)
   
   # Longer TTL for stable queries
   CacheFilter(ttl_seconds=3600)

4. Custom Filters for Domain Logic

   # Create domain-specific filters
   class SecurityFilter(Filter):
       async def pre_invoke(self, context):
           # Validate prompt before LLM
           pass

Semantic Kernel Alignment

This implementation follows Semantic Kernel's design principles:

Feature	SK	agentic-brain
Pre-invoke hooks	✓	✓
Post-invoke hooks	✓	✓
Filter chain pattern	✓	✓
Error handling	✓	✓
Composability	✓	✓
Async support	✓	✓
Built-in filters	✓	✓
Batch processing	✓	✓

See Also

• FILTER_SYSTEM.md - Detailed guide and reference
• examples/filter_system_examples.py - Working examples
• tests/test_hooks_filters.py - Test patterns
• src/agentic_brain/hooks/ - Source code

Future Enhancements

Potential additions:

• Rate limiting filter
• Authentication/authorization filter
• Request validation filter
• Response transformation filter
• Distributed tracing integration
• Prometheus metrics export
• Custom filter registry/discovery