"""Output pipeline for chaining content filters.
Provides pipeline infrastructure for composing multiple ContentFilter
implementations in sequence. Supports per-skill filter configuration.
"""
from __future__ import annotations
from typing import List, Optional, TYPE_CHECKING
from .roleplay_filter import ContentFilter, RoleplayFilter
from .filter import TTSFilter
if TYPE_CHECKING:
from ..skills.skill import Skill
class OutputPipeline:
"""Chain multiple content filters for output processing.
Filters are applied in order. Each filter can optionally check
should_process() before filtering, allowing filters to self-select.
Example:
>>> pipeline = OutputPipeline([RoleplayFilter(), tts_filter])
>>> pipeline.process('*sighs* "Hello there" with code `foo()`')
'Hello there'
"""
def __init__(
self,
filters: Optional[List[ContentFilter]] = None,
respect_should_process: bool = True,
):
"""Initialize the pipeline with a list of filters.
Args:
filters: List of ContentFilter implementations to chain.
Applied in order. If None, empty pipeline (passthrough).
respect_should_process: If True, only apply filters where
should_process() returns True.
"""
self.filters: List[ContentFilter] = filters or []
self.respect_should_process = respect_should_process
def process(self, text: str) -> str:
"""Process text through all filters in sequence.
Args:
text: Input text to filter
Returns:
Text after all applicable filters have been applied
"""
if not text:
return ""
result = text
for filter_obj in self.filters:
# Skip filter if respect_should_process and filter says skip
if self.respect_should_process:
if hasattr(filter_obj, "should_process"):
if not filter_obj.should_process(result):
continue
# Apply the filter
if hasattr(filter_obj, "filter"):
result = filter_obj.filter(result)
elif hasattr(filter_obj, "filter_for_speech"):
# TTSFilter uses filter_for_speech method
result = filter_obj.filter_for_speech(result)
return result
def add_filter(self, filter_obj: ContentFilter) -> "OutputPipeline":
"""Add a filter to the end of the pipeline.
Args:
filter_obj: Filter to add
Returns:
Self for chaining
"""
self.filters.append(filter_obj)
return self
def insert_filter(self, index: int, filter_obj: ContentFilter) -> "OutputPipeline":
"""Insert a filter at a specific position.
Args:
index: Position to insert at
filter_obj: Filter to insert
Returns:
Self for chaining
"""
self.filters.insert(index, filter_obj)
return self
def remove_filter(self, filter_obj: ContentFilter) -> "OutputPipeline":
"""Remove a filter from the pipeline.
Args:
filter_obj: Filter to remove
Returns:
Self for chaining
"""
if filter_obj in self.filters:
self.filters.remove(filter_obj)
return self
def clear(self) -> "OutputPipeline":
"""Remove all filters from the pipeline.
Returns:
Self for chaining
"""
self.filters.clear()
return self
def __len__(self) -> int:
"""Return number of filters in pipeline."""
return len(self.filters)
def create_pipeline_for_skill(skill: Optional["Skill"] = None) -> OutputPipeline:
"""Factory function to create an OutputPipeline based on skill configuration.
Creates a pipeline with filters configured based on the skill's
output_filters list. If no skill or no output_filters defined,
returns a default pipeline with TTSFilter only.
Supported filter names:
- "roleplay": RoleplayFilter (extracts dialogue from quotes)
- "tts": TTSFilter (removes code, URLs, JSON, etc.)
Args:
skill: Optional Skill with output_filters configuration.
If None or no output_filters, uses default pipeline.
Returns:
Configured OutputPipeline instance
"""
# Default filters when no skill or no config
if skill is None:
return OutputPipeline([TTSFilter()])
# Get filter names from skill config
filter_names: List[str] = getattr(skill, "output_filters", []) or []
# If no filters configured, use default (TTS only)
if not filter_names:
return OutputPipeline([TTSFilter()])
# Build filter list based on configuration
filters: List[ContentFilter] = []
for name in filter_names:
name_lower = name.lower()
if name_lower == "roleplay":
filters.append(RoleplayFilter())
elif name_lower == "tts":
filters.append(TTSFilter())
# Unknown filters are silently ignored
# (could add logging here if desired)
# If all filters were unknown, fall back to default
if not filters:
return OutputPipeline([TTSFilter()])
return OutputPipeline(filters)
