plexus.cli.procedure.conversation_filter module

Token-aware conversation filtering for procedure AI execution.

This module provides specialized conversation filtering that considers both message content and token limits for optimal model context management.

class plexus.cli.procedure.conversation_filter.ConversationFilter

Bases: ABC

Abstract base class for conversation filtering strategies.

abstractmethod filter_conversation(conversation_history: List, max_tokens: int | None = None) → List

Filter conversation history according to specific strategy.

class plexus.cli.procedure.conversation_filter.ConversationFilterBase(model: str = 'gpt-4o')

Bases: object

Base class for conversation filtering with token measurement capabilities.

This class provides token counting and measurement functionality without performing any actual filtering. Other filters can inherit from this class to get token measurement capabilities and then implement their own filtering logic.

Initialize the base conversation filter.

Args:

model: Model name for token encoding (default: “gpt-4o”)

__init__(model: str = 'gpt-4o')

Initialize the base conversation filter.

Args:

model: Model name for token encoding (default: “gpt-4o”)

analyze_conversation_tokens(conversation_history: List, max_tokens: int | None = None) → Dict[str, Any]

Analyze token usage in conversation history without modifying it.

This method provides pure measurement and analysis functionality. It reports on token usage, message breakdown, and whether limits are exceeded, but does not perform any filtering or modification of the conversation.

Args:

conversation_history: List of conversation messages max_tokens: Optional maximum token limit for analysis

Returns:

Dictionary with token analysis results

class plexus.cli.procedure.conversation_filter.ManagerAgentConversationFilter(model: str = 'gpt-4o')

Bases: ConversationFilterBase, ConversationFilter

Specialized conversation filter for manager/orchestration agents that provides high-level oversight.

The manager agent needs a strategic overview of the conversation rather than detailed technical execution. This filter creates a filtered message list optimized for orchestration decisions, with proper system message setup for manager-style responses.

This filter is designed for SOP (Standard Operating Procedure) agents and other manager-level agents that need to understand conversation flow and guide the overall process.

Initialize the manager agent filter.

Args:

model: Model name for token encoding (default: “gpt-4o”)

__init__(model: str = 'gpt-4o')

Initialize the manager agent filter.

Args:

model: Model name for token encoding (default: “gpt-4o”)

filter_conversation(conversation_history: List, max_tokens: int | None = None, manager_system_prompt: str | None = None) → List

Filter conversation history for manager agent consumption.

This creates a filtered message list that: 1. Replaces the original system message with a manager-specific system prompt 2. Preserves recent messages in full (last 3-5 messages) 3. Summarizes older messages for context 4. Adds a final system message instructing manager-style response

Args:

conversation_history: Complete conversation history max_tokens: Optional maximum token limit (affects summary length) manager_system_prompt: System prompt specifically for the manager agent

Returns:

Filtered list of messages optimized for manager agent oversight

class plexus.cli.procedure.conversation_filter.SOPAgentConversationFilter(model: str = 'gpt-4o')

Bases: ConversationFilterBase, ConversationFilter

Specialized conversation filter for SOP (Standard Operating Procedure) agents that need minimal conversation context to make orchestration decisions.

This filter is designed to prevent the SOP agent from getting overwhelmed by detailed conversation history while still providing enough context to make informed guidance decisions.

Key behaviors: 1. Preserves only the most recent message (last assistant or user message) 2. Truncates all other messages to brief summaries 3. Maintains conversation flow indicators for stage awareness 4. Optimized for orchestration/guidance rather than detailed technical work

Initialize the SOP agent conversation filter.

Args:

model: Model name for token encoding (default: “gpt-4o”)

__init__(model: str = 'gpt-4o')

Initialize the SOP agent conversation filter.

Args:

model: Model name for token encoding (default: “gpt-4o”)

filter_conversation(conversation_history: List, max_tokens: int | None = None) → List

Filter conversation history for SOP agent consumption.

This creates a minimal context that preserves: 1. The most recent message in full (for immediate context) 2. Brief summaries of earlier messages (for flow awareness) 3. Stage transition indicators

Args:

conversation_history: Complete conversation history max_tokens: Optional maximum token limit

Returns:

Filtered conversation history with minimal context for SOP oversight

class plexus.cli.procedure.conversation_filter.TokenAwareConversationFilter(model: str = 'gpt-4o')

Bases: ConversationFilterBase, ConversationFilter

Token-aware conversation filter that combines character-based and token-based filtering.

This filter implements intelligent conversation management by: 1. Identifying tool result messages (ToolMessage or SystemMessage with tool patterns) 2. Counting tokens per message for precise context management (inherited from base) 3. Applying filtering rules based on both content size and token limits 4. Preserving recent tool results while truncating older ones

Initialize the token-aware filter.

Args:

model: Model name for token encoding (default: “gpt-4o”)

__init__(model: str = 'gpt-4o')

Initialize the token-aware filter.

Args:

model: Model name for token encoding (default: “gpt-4o”)

filter_conversation(conversation_history: List, max_tokens: int | None = None) → List

Filter conversation history with token awareness.

Behavior Driven Design: GIVEN a conversation history with tool result messages WHEN the conversation exceeds token limits THEN filter by keeping recent tool results full and truncating older ones AND provide detailed per-message token reporting

Args:

conversation_history: Complete conversation history max_tokens: Optional maximum token limit

Returns:

Filtered conversation history

class plexus.cli.procedure.conversation_filter.WorkerAgentConversationFilter(model: str = 'gpt-4o')

Bases: TokenAwareConversationFilter

Specialized conversation filter for worker agents that performs specific filtering:

1. Filters tool result messages (keeps most recent 2, truncates older ones)

2. Filters empty AI messages that contain only tool calls (no text content)

This filter is designed for coding assistant workers that make tool calls but may produce empty reasoning messages that just contain tool calls.

Initialize the token-aware filter.

Args:

model: Model name for token encoding (default: “gpt-4o”)

filter_conversation(conversation_history: List, max_tokens: int | None = None) → List

Filter conversation history for worker agents with enhanced filtering.

This method applies both tool result filtering and empty AI message filtering while maintaining OpenAI API requirements and preserving the current conversation state: 1. Applies the standard tool result filtering (most recent 2 kept, others truncated) 2. Removes OLD AI messages that contain only tool calls with no text content 3. CRITICAL: Preserves the MOST RECENT AI message (current tool call) to avoid infinite loops 4. Also removes corresponding tool results to maintain API compliance

Args:

conversation_history: Complete conversation history max_tokens: Optional maximum token limit

Returns:

Filtered conversation history with both tool result and empty AI message filtering