Metadata-Version: 2.4
Name: cognitive-load-monitor
Version: 0.1.0
Summary: Lightweight cognitive load monitoring primitive for AI agents
Author-email: Synapse Data <contact@synapsedata.example>
License: Proprietary - See LICENSE file
Project-URL: Homepage, https://github.com/synapse-data/cognitive-load-monitor
Project-URL: Documentation, https://github.com/synapse-data/cognitive-load-monitor#readme
Project-URL: Repository, https://github.com/synapse-data/cognitive-load-monitor
Project-URL: Issues, https://github.com/synapse-data/cognitive-load-monitor/issues
Keywords: ai-agents,cognitive-load,monitoring,observability,orchestration,reliability,agent-infrastructure
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Operating System :: OS Independent
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Dynamic: license-file

# Cognitive Load Monitor

A lightweight, drop-in Python module for AI agents to self-report cognitive load using observable runtime metrics. Designed for operational reliability and external orchestration (throttling, cloning, load balancing).

## Installation

```bash
pip install cognitive-load-monitor
```

## Quick Start

```python
from cognitive_load_monitor import CognitiveLoadMonitor

# Initialize monitor
monitor = CognitiveLoadMonitor()

# Record agent state and get load report
report = monitor.record(
    tokens_used=1500,
    tokens_budget=2000,
    reasoning_steps=8,
    latency_ms=450,
    unresolved_assumptions=2,
    total_assumptions=5,
)

# Check load status
print(f"Load Index: {report.load_index:.2f}")
print(f"Trend: {report.trend.value}")

if report.is_overloaded():
    # Signal orchestrator to throttle or clone agent
    emit_signal("THROTTLE", report.to_dict())
```

## What is Cognitive Load?

In this context, **cognitive load** is an operational reliability metric—not a philosophical concept. It measures resource constraints and processing stress that affect agent performance, similar to:

- CPU/memory pressure in systems monitoring
- Queue depth in message brokers  
- Latency percentiles in web services

## Metrics

The monitor computes a normalized **Cognitive Load Index (0–1)** using five proxy metrics:

### 1. Context Pressure (0–1)
Ratio of tokens used to budget. Measures memory/context constraints.

### 2. Reasoning Complexity (0–1)
Normalized combination of reasoning steps and backtracking. High values indicate difficult problem-solving.

### 3. Temporal Stress (0–1)
Ratio of actual latency to expected latency. Measures time pressure and processing delays.

### 4. Uncertainty (0–1)
Ratio of unresolved to total assumptions. High values indicate ambiguity or missing information.

### 5. Error Recovery (0–1)
Ratio of self-corrections to total operations. Measures instability and rework overhead.

## Configuration

### Custom Weights

```python
from cognitive_load_monitor import CognitiveLoadMonitor, MetricWeights

# Define custom weights (must sum to 1.0)
weights = MetricWeights(
    context_pressure=0.30,
    reasoning_complexity=0.30,
    temporal_stress=0.20,
    uncertainty=0.10,
    error_recovery=0.10,
)

monitor = CognitiveLoadMonitor(weights=weights)
```

### Trend Detection

```python
monitor = CognitiveLoadMonitor(
    history_window=15,        # Track last 15 samples
    trend_threshold=0.08,     # Sensitivity for rising/falling detection
)
```

## Use Cases

### Multi-Agent Orchestration
```python
# Load balancer decides which agent gets next request
agents = [agent1, agent2, agent3]
loads = [a.monitor.get_current_load() for a in agents]
selected = agents[loads.index(min(loads))]
```

### Dynamic Model Selection
```python
report = monitor.record(...)

if report.load_index < 0.3:
    model = "gpt-4o-mini"  # Use cheaper model for low-load tasks
elif report.load_index > 0.7:
    model = "gpt-4o"        # Use powerful model for high-load tasks
```

### Quality Gates
```python
if report.is_rising_fast(threshold=0.6):
    # Escalate to human review
    escalate_to_human(task, report)
```

### Cost Optimization
```python
# Track load vs output quality for capacity planning
metrics_db.log({
    "load_index": report.load_index,
    "output_quality": evaluate_output(result),
    "cost": compute_cost(tokens_used),
})
```

## API Reference

### `CognitiveLoadMonitor`

#### `__init__(weights=None, history_window=10, trend_threshold=0.05)`
Initialize monitor with optional custom configuration.

#### `record(**kwargs) -> CognitiveLoadReport`
Record current agent state and compute load report.

**Parameters:**
- `tokens_used` (int): Current token count consumed
- `tokens_budget` (int): Maximum tokens available
- `reasoning_steps` (int): Number of reasoning steps taken
- `max_reasoning_steps` (int): Expected maximum steps
- `backtrack_count` (int): Number of backtracks/revisions
- `latency_ms` (float): Actual processing time in milliseconds
- `expected_latency_ms` (float): Baseline expected latency
- `unresolved_assumptions` (int): Count of uncertain assumptions
- `total_assumptions` (int): Total assumptions made
- `self_corrections` (int): Number of self-corrections
- `total_operations` (int): Total operations attempted

#### `reset_history()`
Clear historical data. Useful when agent context resets.

#### `get_current_load() -> Optional[float]`
Get most recent load index, or None if no data.

### `CognitiveLoadReport`

#### `to_dict() -> dict`
Convert report to dictionary for serialization.

#### `is_overloaded(threshold=0.75) -> bool`
Check if cognitive load exceeds threshold.

#### `is_rising_fast(threshold=0.60) -> bool`
Check if load is rising and already above threshold.

**Attributes:**
- `timestamp`: Unix timestamp when report was generated
- `load_index`: Normalized cognitive load score (0.0–1.0)
- `trend`: Direction of load change (`LoadTrend` enum)
- `metrics`: Raw metric values used to compute load_index
- `weights`: Weight configuration used for this report
- `history_size`: Number of historical samples

## Design Philosophy

- **Treat cognitive load as an operational signal**, not a philosophical concept
- **Use only metrics an agent can observe at runtime**
- **No inspection of chain-of-thought or model internals**
- **Simple, transparent, configurable**
- **Zero external dependencies**

## Non-Goals

This module intentionally does NOT include:
- Orchestration logic
- Networking or API calls
- Dashboards or visualization
- Logging frameworks

It's a primitive building block for you to integrate into your infrastructure.

## License

**All Rights Reserved**  
Free to use for non-commercial and free applications.  
For commercial use, please contact Synapse Data.

## Requirements

- Python 3.10+
- No external dependencies

## Contributing

This is a commercial product by Synapse Data. For feature requests or commercial licensing inquiries, please contact us.

## Support

For issues or questions:
- Open an issue on GitHub
- Contact: Synapse Data

---

**Built by Synapse Data** • Production-quality infrastructure for AI agents
