Conversation Filters

Conversation filters transform conversation history before it’s sent to the LLM, allowing you to manage context windows, remove sensitive data, and optimize prompts.

ConversationFilter Interface

from vanna.core.filter import ConversationFilter
from vanna.core.storage import Message
from typing import List

class ConversationFilter(ABC):
    @abstractmethod
    async def filter_messages(self, messages: List[Message]) -> List[Message]:
        """Filter and transform conversation messages"""
        return messages

Registering Filters

from vanna import Agent

agent = Agent(
    llm_service=llm,
    conversation_filters=[
        ContextWindowFilter(max_tokens=8000),
        SensitiveDataFilter(),
        DeduplicationFilter()
    ]
)

Built-in Filters

ContextWindowFilter

Manage conversation history to fit within model context limits:

from vanna.core.filter import ContextWindowFilter

filter = ContextWindowFilter(max_tokens=8000)

Custom Filter Examples

Example 1: Context Window Management

from vanna.core.filter import ConversationFilter

class ContextWindowFilter(ConversationFilter):
    def __init__(self, max_tokens: int = 8000):
        self.max_tokens = max_tokens
    
    async def filter_messages(self, messages: List[Message]) -> List[Message]:
        # Estimate tokens (rough approximation: 4 chars = 1 token)
        total_tokens = 0
        filtered = []
        
        # Always keep system message
        system_msg = next((m for m in messages if m.role == 'system'), None)
        if system_msg:
            filtered.append(system_msg)
            total_tokens += len(system_msg.content or "") // 4
        
        # Add recent messages within limit
        for msg in reversed([m for m in messages if m.role != 'system']):
            msg_tokens = len(msg.content or "") // 4
            
            if total_tokens + msg_tokens > self.max_tokens:
                break
            
            filtered.insert(1 if system_msg else 0, msg)
            total_tokens += msg_tokens
        
        return filtered

Example 2: Sensitive Data Removal

import re

class SensitiveDataFilter(ConversationFilter):
    PATTERNS = {
        'email': r'[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+.[A-Z|a-z]{2,}',
        'phone': r'd{3}[-.]?d{3}[-.]?d{4}',
        'ssn': r'd{3}-d{2}-d{4}',
        'credit_card': r'd{4}[s-]?d{4}[s-]?d{4}[s-]?d{4}'
    }
    
    async def filter_messages(self, messages: List[Message]) -> List[Message]:
        filtered = []
        
        for msg in messages:
            if msg.content:
                content = msg.content
                
                # Redact sensitive patterns
                for pattern_type, pattern in self.PATTERNS.items():
                    content = re.sub(pattern, f'[REDACTED_{pattern_type.upper()}]', content)
                
                # Create new message with sanitized content
                filtered.append(Message(
                    role=msg.role,
                    content=content,
                    metadata=msg.metadata
                ))
            else:
                filtered.append(msg)
        
        return filtered

Example 3: Deduplication Filter

class DeduplicationFilter(ConversationFilter):
    async def filter_messages(self, messages: List[Message]) -> List[Message]:
        seen = set()
        filtered = []
        
        for msg in messages:
            # Create hash of message content
            msg_hash = hash((msg.role, msg.content))
            
            if msg_hash not in seen:
                seen.add(msg_hash)
                filtered.append(msg)
        
        return filtered

Example 4: Summarization Filter

class SummarizationFilter(ConversationFilter):
    def __init__(self, summarizer_llm, max_messages=20):
        self.summarizer = summarizer_llm
        self.max_messages = max_messages
    
    async def filter_messages(self, messages: List[Message]) -> List[Message]:
        if len(messages) <= self.max_messages:
            return messages
        
        # Keep recent messages
        recent = messages[-self.max_messages:]
        
        # Summarize older messages
        old_messages = messages[:-self.max_messages]
        
        summary_text = await self._summarize_messages(old_messages)
        
        # Create summary message
        summary_msg = Message(
            role='system',
            content=f"Previous conversation summary:\n{summary_text}"
        )
        
        return [summary_msg] + recent
    
    async def _summarize_messages(self, messages: List[Message]) -> str:
        # Use LLM to summarize
        conversation_text = "\n".join(
            f"{msg.role}: {msg.content}"
            for msg in messages
        )
        
        summary = await self.summarizer.summarize(conversation_text)
        return summary

Example 5: Priority Message Filter

class PriorityFilter(ConversationFilter):
    async def filter_messages(self, messages: List[Message]) -> List[Message]:
        # Categorize messages
        system_msgs = []
        important_msgs = []
        regular_msgs = []
        
        for msg in messages:
            if msg.role == 'system':
                system_msgs.append(msg)
            elif msg.metadata and msg.metadata.get('important'):
                important_msgs.append(msg)
            else:
                regular_msgs.append(msg)
        
        # Return prioritized list: system -> important -> recent regular
        return system_msgs + important_msgs + regular_msgs[-10:]

Example 6: Tool Result Compression

class ToolResultCompressionFilter(ConversationFilter):
    async def filter_messages(self, messages: List[Message]) -> List[Message]:
        filtered = []
        
        for msg in messages:
            if msg.role == 'tool' and msg.metadata:
                # Compress large tool results
                if len(msg.content or "") > 1000:
                    # Keep metadata but truncate content
                    filtered.append(Message(
                        role=msg.role,
                        content=msg.content[:500] + "\n... [truncated] ...",
                        metadata=msg.metadata
                    ))
                else:
                    filtered.append(msg)
            else:
                filtered.append(msg)
        
        return filtered

Filter Execution Order

Filters are applied in the order they’re registered:

agent = Agent(
    conversation_filters=[
        DeduplicationFilter(),         # Remove duplicates first
        SensitiveDataFilter(),         # Then redact sensitive data
        ToolResultCompressionFilter(), # Compress tool results
        ContextWindowFilter()          # Finally trim to context window
    ]
)

Use Cases

Managing Long Conversations

conversation_filters=[
    SummarizationFilter(summarizer_llm, max_messages=50),
    ContextWindowFilter(max_tokens=8000)
]

Privacy Protection

conversation_filters=[
    SensitiveDataFilter(),
    PIIRemovalFilter(),
    CredentialRedactionFilter()
]

Performance Optimization

conversation_filters=[
    ToolResultCompressionFilter(),
    DeduplicationFilter(),
    ContextWindowFilter(max_tokens=4000)  # Use smaller context
]

Testing Filters

# Test your filter
filter = ContextWindowFilter(max_tokens=100)

test_messages = [
    Message(role='user', content='Hello'),
    Message(role='assistant', content='Hi there!'),
    Message(role='user', content='What is the weather?')
]

filtered = await filter.filter_messages(test_messages)
assert len(filtered) <= len(test_messages)

Best Practices

1. Order matters - Apply filters in logical sequence
2. Preserve system messages - Keep important context
3. Test with real data - Ensure filters work as expected
4. Monitor token usage - Verify context window management
5. Be careful with summarization - May lose important details
6. Log filter actions - For debugging and auditing
7. Consider performance - Filters run on every LLM call

See Also

• Context Enrichers
• LLM Middlewares
• System Prompts