zialectics-signal 0.1.0

Domain-agnostic event-first observation and signal derivation framework

Zialectics Signal Framework

A domain-agnostic, event-first observation and signal derivation framework for building AI-powered systems that observe, learn, and assist.

Core Principles

1. Events are immutable facts — an Observation records something that happened; it is never modified
2. Signals are derived, never stored — importance, anomalies, and profiles are computed at query time
3. Three-state confirmation — unconfirmed, confirmed, or corrected by a human
4. Training gate — only human-verified signals (weight >= threshold) feed AI learning
5. Channel-agnostic schema — works with email, Slack, CRM, calendar, or any event source

Architecture

Channel Adapter  →  Observation Store  →  Derivation Engine
(Gmail, Slack,       (immutable event       (baselines, anomaly
 CRM, Calendar)       records)               detection, importance,
                                             time-decay scoring)
                                                    ↓
                                             Training Gate
                                             (weight >= 0.5)
                                                    ↓
                                             AI Learning Loop

Installation

pip install zialectics-signal

With Gmail adapter support:

pip install zialectics-signal[gmail]

Quick Start

1. Define your observation types

import enum
from zialectics_signal import BaseObservationType

class MyObservationType(enum.Enum):
    # Inherit universal types
    received = "received"
    read = "read"
    archived = "archived"
    classified = "classified"
    classification_confirmed = "classification_confirmed"
    classification_corrected = "classification_corrected"

    # Add domain-specific types
    escalated = "escalated"
    resolved = "resolved"
    sla_breached = "sla_breached"

2. Configure signal weights

from zialectics_signal.core import SignalWeights

weights = SignalWeights()

# Set signal strengths (0.0 = noise, 1.0 = strongest signal)
weights.set_weight(MyObservationType.classification_confirmed, 0.8)
weights.set_weight(MyObservationType.classification_corrected, 0.9)
weights.set_weight(MyObservationType.escalated, 0.7)
weights.set_weight(MyObservationType.resolved, 0.6)
weights.set_weight(MyObservationType.received, 0.1)

# Classify signal direction
weights.set_positive(
    MyObservationType.classification_confirmed,
    MyObservationType.resolved,
)
weights.set_negative(
    MyObservationType.classification_corrected,
    MyObservationType.sla_breached,
)

3. Create the framework instance

from zialectics_signal import SignalFramework

framework = SignalFramework(
    weights=weights,
    training_threshold=0.5,   # Only train on strong signals
    half_life_days=30.0,      # Exponential decay over 30 days
)

4. Derive insights

# Check training gate
framework.passes_training_gate(MyObservationType.classification_confirmed)  # True
framework.passes_training_gate(MyObservationType.received)                  # False

# Compute time-decay weight
from datetime import datetime, timedelta, timezone
recent = datetime.now(timezone.utc) - timedelta(days=1)
old = datetime.now(timezone.utc) - timedelta(days=60)

framework.compute_time_decay(recent)  # ~0.98 (very recent)
framework.compute_time_decay(old)     # ~0.25 (two half-lives old)

# Derive originator profile from observation dicts
observations = [
    {"observation_type": "received", "observed_at": "2024-01-15T10:00:00"},
    {"observation_type": "replied_to", "observed_at": "2024-01-15T10:30:00"},
    {"observation_type": "classified", "observed_at": "2024-01-15T10:01:00"},
]
profile = framework.derive_originator_profile(observations)
# {"total_received": 1, "total_interactions": 1, "engagement_rate": 1.0, ...}

5. Build a channel adapter

from zialectics_signal.adapters import ChannelAdapter

class SlackAdapter(ChannelAdapter):
    channel_name = "slack"

    def authenticate(self) -> bool:
        # Connect to Slack API
        ...

    def fetch_events(self, since=None, limit=100):
        # Fetch messages, reactions, threads
        return [{"event_id": "...", "event_type": "received", ...}]

    def record_observation(self, event, engine, account):
        # Map Slack event → Observation and persist
        ...

    def normalize_originator(self, raw_originator):
        # Parse Slack user format
        return display_name, user_id

Observation Schema

The framework uses a channel-agnostic naming convention:

Framework Term	Gmail	Slack	CRM
communication_id	Message ID	Message ts	Ticket ID
conversation_id	Thread ID	Channel + thread	Case ID
originator	Sender name	User name	Contact name
originator_address	Email address	User ID	Contact email
channel	"gmail"	"slack"	"salesforce"
audience_type	direct/cc/bcc	channel/dm	assigned/cc

Key Components

Component	Purpose
BaseObservationType	Universal event types (received, read, classified, etc.)
ObservationBase	Abstract SQLAlchemy model for immutable event records
SignalWeights	Configurable weight mapping with positive/negative/neutral classification
SignalFramework	Orchestrator: training gate, time decay, originator profiles
ChannelAdapter	Abstract interface for connecting data sources
GmailAdapter	Reference implementation for Gmail

License

MIT License. Copyright (c) 2024 Zialectics LLC.