Monitoring API

The monitoring module provides state tracking, visualization, and logging for LRS agents.

State Tracking

State tracking for LRS agents.

Maintains a rolling history of agent states for analysis and visualization.

class lrs.monitoring.tracker.StateSnapshot(timestamp: datetime, precision: Dict[str, float], prediction_errors: List[float], tool_history: List[Dict[str, Any]], adaptation_count: int, belief_state: Dict[str, Any])[source]

Bases: object

Snapshot of agent state at a specific point in time.

timestamp

When this snapshot was taken

Type:

datetime.datetime

precision

Precision values at all levels

Type:

Dict[str, float]

prediction_errors

Recent prediction errors

Type:

List[float]

tool_history

Tool execution history

Type:

List[Dict[str, Any]]

adaptation_count

Number of adaptations so far

Type:

int

belief_state

Current beliefs

Type:

Dict[str, Any]

timestamp: datetime
precision: Dict[str, float]
prediction_errors: List[float]
tool_history: List[Dict[str, Any]]
adaptation_count: int
belief_state: Dict[str, Any]
__init__(timestamp: datetime, precision: Dict[str, float], prediction_errors: List[float], tool_history: List[Dict[str, Any]], adaptation_count: int, belief_state: Dict[str, Any]) → None
class lrs.monitoring.tracker.LRSStateTracker(max_history: int = 100)[source]

Bases: object

Tracks agent state history for monitoring and analysis.

Maintains a rolling window of state snapshots with configurable size. Used by the dashboard and for post-execution analysis.

Examples

>>> tracker = LRSStateTracker(max_history=100)
>>>
>>> # Track state during execution
>>> for step in agent_execution:
...     tracker.track_state(step)
>>>
>>> # Analyze precision trajectory
>>> precision_history = tracker.get_precision_trajectory('execution')
>>> print(f"Average precision: {sum(precision_history) / len(precision_history)}")
>>>
>>> # Get adaptation events
>>> adaptations = tracker.get_adaptation_events()
>>> print(f"Total adaptations: {len(adaptations)}")
__init__(max_history: int = 100)[source]

Initialize state tracker.

Parameters:

max_history – Maximum number of states to keep in history

track_state(state: Dict[str, Any])[source]

Track a new state snapshot.

Parameters:

state – Current agent state (LRSState dict)

Examples

>>> tracker.track_state({
...     'precision': {'execution': 0.7, 'planning': 0.6},
...     'tool_history': [...],
...     'belief_state': {...}
... })
get_precision_trajectory(level: str = 'execution') → List[float][source]

Get precision trajectory for a specific level.

Parameters:

level – Precision level (‘abstract’, ‘planning’, or ‘execution’)

Returns:

List of precision values over time

Examples

>>> trajectory = tracker.get_precision_trajectory('execution')
>>> import matplotlib.pyplot as plt
>>> plt.plot(trajectory)
>>> plt.show()
get_all_precision_trajectories() → Dict[str, List[float]][source]

Get precision trajectories for all levels.

Returns:

Dict mapping level names to precision trajectories

get_prediction_errors() → List[float][source]

Get all prediction errors from history.

Returns:

Flat list of all prediction errors

get_adaptation_events() → List[Dict[str, Any]][source]

Get all recorded adaptation events.

Returns:

List of adaptation event dicts

get_tool_usage_stats() → Dict[str, Dict[str, Any]][source]

Calculate tool usage statistics.

Returns:

Dict mapping tool names to stats (calls, successes, avg_error)

Examples

>>> stats = tracker.get_tool_usage_stats()
>>> for tool, data in stats.items():
...     print(f"{tool}: {data['success_rate']:.1%} success rate")
get_current_state() → StateSnapshot | None[source]

Get most recent state snapshot.

Returns:

Latest StateSnapshot or None if no history

export_history(filepath: str)[source]

Export history to JSON file.

Parameters:

filepath – Output file path

Examples

>>> tracker.export_history('agent_history.json')
clear()[source]

Clear all tracked history

get_summary() → Dict[str, Any][source]

Get summary statistics of tracked execution.

Returns:

Dict with summary metrics

Examples

>>> summary = tracker.get_summary()
>>> print(f"Total steps: {summary['total_steps']}")
>>> print(f"Adaptations: {summary['total_adaptations']}")

Classes

class lrs.monitoring.tracker.StateSnapshot(timestamp: datetime, precision: Dict[str, float], prediction_errors: List[float], tool_history: List[Dict[str, Any]], adaptation_count: int, belief_state: Dict[str, Any])[source]

Bases: object

Snapshot of agent state at a specific point in time.

timestamp

When this snapshot was taken

Type:

datetime.datetime

precision

Precision values at all levels

Type:

Dict[str, float]

prediction_errors

Recent prediction errors

Type:

List[float]

tool_history

Tool execution history

Type:

List[Dict[str, Any]]

adaptation_count

Number of adaptations so far

Type:

int

belief_state

Current beliefs

Type:

Dict[str, Any]

Snapshot of agent state at a point in time.

Attributes:

  • timestamp (datetime): When snapshot was taken

  • precision (Dict[str, float]): Precision values by level

  • prediction_errors (List[float]): Recent prediction errors

  • tool_history (List[Dict]): Tool execution history

  • adaptation_count (int): Number of adaptations so far

  • belief_state (Dict): Current belief state

timestamp: datetime
precision: Dict[str, float]
prediction_errors: List[float]
tool_history: List[Dict[str, Any]]
adaptation_count: int
belief_state: Dict[str, Any]
__init__(timestamp: datetime, precision: Dict[str, float], prediction_errors: List[float], tool_history: List[Dict[str, Any]], adaptation_count: int, belief_state: Dict[str, Any]) → None
class lrs.monitoring.tracker.LRSStateTracker(max_history: int = 100)[source]

Bases: object

Tracks agent state history for monitoring and analysis.

Maintains a rolling window of state snapshots with configurable size. Used by the dashboard and for post-execution analysis.

Examples

>>> tracker = LRSStateTracker(max_history=100)
>>>
>>> # Track state during execution
>>> for step in agent_execution:
...     tracker.track_state(step)
>>>
>>> # Analyze precision trajectory
>>> precision_history = tracker.get_precision_trajectory('execution')
>>> print(f"Average precision: {sum(precision_history) / len(precision_history)}")
>>>
>>> # Get adaptation events
>>> adaptations = tracker.get_adaptation_events()
>>> print(f"Total adaptations: {len(adaptations)}")

Tracks agent state over time for analysis and visualization.

Methods:

track_state(state: Dict[str, Any])[source]

Track a new state snapshot.

Parameters:

state – Current agent state (LRSState dict)

Examples

>>> tracker.track_state({
...     'precision': {'execution': 0.7, 'planning': 0.6},
...     'tool_history': [...],
...     'belief_state': {...}
... })

Track current agent state.

get_precision_trajectory(level: str = 'execution') → List[float][source]

Get precision trajectory for a specific level.

Parameters:

level – Precision level (‘abstract’, ‘planning’, or ‘execution’)

Returns:

List of precision values over time

Examples

>>> trajectory = tracker.get_precision_trajectory('execution')
>>> import matplotlib.pyplot as plt
>>> plt.plot(trajectory)
>>> plt.show()

Get precision history for a specific level.

get_all_precision_trajectories() → Dict[str, List[float]][source]

Get precision trajectories for all levels.

Returns:

Dict mapping level names to precision trajectories

Get precision histories for all levels.

get_prediction_errors() → List[float][source]

Get all prediction errors from history.

Returns:

Flat list of all prediction errors

Get recent prediction errors.

get_adaptation_events() → List[Dict[str, Any]][source]

Get all recorded adaptation events.

Returns:

List of adaptation event dicts

Get timeline of adaptation events.

get_tool_usage_stats() → Dict[str, Dict[str, Any]][source]

Calculate tool usage statistics.

Returns:

Dict mapping tool names to stats (calls, successes, avg_error)

Examples

>>> stats = tracker.get_tool_usage_stats()
>>> for tool, data in stats.items():
...     print(f"{tool}: {data['success_rate']:.1%} success rate")

Get tool usage statistics and reliability.

get_summary() → Dict[str, Any][source]

Get summary statistics of tracked execution.

Returns:

Dict with summary metrics

Examples

>>> summary = tracker.get_summary()
>>> print(f"Total steps: {summary['total_steps']}")
>>> print(f"Adaptations: {summary['total_adaptations']}")

Get summary statistics.

export_history(filepath: str)[source]

Export history to JSON file.

Parameters:

filepath – Output file path

Examples

>>> tracker.export_history('agent_history.json')

Export history to JSON file.

clear()[source]

Clear all tracked history

Clear all tracked history.

Example:

from lrs.monitoring.tracker import LRSStateTracker

tracker = LRSStateTracker(max_history=1000)

# Track state after each step
tracker.track_state({
    'precision': {'execution': 0.7, 'planning': 0.6},
    'tool_history': [...],
    'adaptation_count': 2,
    'belief_state': {}
})

# Analyze precision trajectory
exec_prec = tracker.get_precision_trajectory('execution')
# Returns: [0.5, 0.6, 0.45, 0.55, 0.7, ...]

# Get tool statistics
stats = tracker.get_tool_usage_stats()
# Returns: {
#   'fetch_api': {
#     'calls': 10,
#     'successes': 7,
#     'failures': 3,
#     'success_rate': 0.7,
#     'avg_error': 0.35
#   }
# }

# Export for analysis
tracker.export_history('execution_history.json')
__init__(max_history: int = 100)[source]

Initialize state tracker.

Parameters:

max_history – Maximum number of states to keep in history

track_state(state: Dict[str, Any])[source]

Track a new state snapshot.

Parameters:

state – Current agent state (LRSState dict)

Examples

>>> tracker.track_state({
...     'precision': {'execution': 0.7, 'planning': 0.6},
...     'tool_history': [...],
...     'belief_state': {...}
... })
get_precision_trajectory(level: str = 'execution') → List[float][source]

Get precision trajectory for a specific level.

Parameters:

level – Precision level (‘abstract’, ‘planning’, or ‘execution’)

Returns:

List of precision values over time

Examples

>>> trajectory = tracker.get_precision_trajectory('execution')
>>> import matplotlib.pyplot as plt
>>> plt.plot(trajectory)
>>> plt.show()
get_all_precision_trajectories() → Dict[str, List[float]][source]

Get precision trajectories for all levels.

Returns:

Dict mapping level names to precision trajectories

get_prediction_errors() → List[float][source]

Get all prediction errors from history.

Returns:

Flat list of all prediction errors

get_adaptation_events() → List[Dict[str, Any]][source]

Get all recorded adaptation events.

Returns:

List of adaptation event dicts

get_tool_usage_stats() → Dict[str, Dict[str, Any]][source]

Calculate tool usage statistics.

Returns:

Dict mapping tool names to stats (calls, successes, avg_error)

Examples

>>> stats = tracker.get_tool_usage_stats()
>>> for tool, data in stats.items():
...     print(f"{tool}: {data['success_rate']:.1%} success rate")
get_current_state() → StateSnapshot | None[source]

Get most recent state snapshot.

Returns:

Latest StateSnapshot or None if no history

export_history(filepath: str)[source]

Export history to JSON file.

Parameters:

filepath – Output file path

Examples

>>> tracker.export_history('agent_history.json')
clear()[source]

Clear all tracked history

get_summary() → Dict[str, Any][source]

Get summary statistics of tracked execution.

Returns:

Dict with summary metrics

Examples

>>> summary = tracker.get_summary()
>>> print(f"Total steps: {summary['total_steps']}")
>>> print(f"Adaptations: {summary['total_adaptations']}")

Structured Logging

Structured logging for LRS-Agents.

Provides JSON-formatted logs for production monitoring and analysis.

class lrs.monitoring.structured_logging.LRSLogger(agent_id: str, log_file: str | None = None, console: bool = True, level: int = 20)[source]

Bases: object

Structured logger for LRS agents.

Logs events in JSON format for easy parsing and analysis. Captures: - Precision changes - Policy selections - Tool executions - Adaptation events - Performance metrics

Examples

>>> logger = LRSLogger(agent_id="agent_1", log_file="agent.jsonl")
>>>
>>> logger.log_precision_update(
...     level='execution',
...     old_value=0.8,
...     new_value=0.4,
...     prediction_error=0.95
... )
>>>
>>> logger.log_tool_execution(
...     tool_name="api_fetch",
...     success=False,
...     execution_time=0.5,
...     prediction_error=0.9,
...     error_message="Timeout"
... )
__init__(agent_id: str, log_file: str | None = None, console: bool = True, level: int = 20)[source]

Initialize structured logger.

Parameters:

  • agent_id – Unique identifier for this agent

  • log_file – Optional file path for JSON logs

  • console – Whether to also log to console

  • level – Logging level

log_precision_update(level: str, old_value: float, new_value: float, prediction_error: float, propagated: bool = False)[source]

Log precision update event.

Parameters:

  • level – Precision level (abstract/planning/execution)

  • old_value – Previous precision value

  • new_value – New precision value

  • prediction_error – Triggering prediction error

  • propagated – Whether error propagated from lower level

log_policy_selection(policies: list, selected_index: int, G_values: list, precision: float)[source]

Log policy selection via G.

Parameters:

  • policies – List of candidate policies

  • selected_index – Index of selected policy

  • G_values – Expected Free Energy values

  • precision – Current precision value

log_tool_execution(tool_name: str, success: bool, execution_time: float, prediction_error: float, error_message: str | None = None)[source]

Log tool execution.

Parameters:

  • tool_name – Name of executed tool

  • success – Whether execution succeeded

  • execution_time – Execution time in seconds

  • prediction_error – Observed prediction error

  • error_message – Error message if failed

log_adaptation_event(trigger: str, old_precision: Dict[str, float], new_precision: Dict[str, float], action_taken: str)[source]

Log adaptation event.

Parameters:

  • trigger – What triggered the adaptation

  • old_precision – Precision before adaptation

  • new_precision – Precision after adaptation

  • action_taken – Action taken by agent

log_performance_metrics(total_steps: int, success_rate: float, avg_precision: float, adaptation_count: int, execution_time: float)[source]

Log aggregate performance metrics.

Parameters:

  • total_steps – Total execution steps

  • success_rate – Overall success rate

  • avg_precision – Average precision value

  • adaptation_count – Number of adaptations

  • execution_time – Total execution time

log_error(error_type: str, message: str, stack_trace: str | None = None)[source]

Log error event.

Parameters:

  • error_type – Type of error

  • message – Error message

  • stack_trace – Optional stack trace

lrs.monitoring.structured_logging.create_logger_for_agent(agent_id: str, **kwargs) → LRSLogger[source]

Create logger for LRS agent.

Parameters:

  • agent_id – Agent identifier

  • **kwargs – Passed to LRSLogger

Returns:

Configured logger instance

Examples

>>> logger = create_logger_for_agent(
...     "production_agent_1",
...     log_file="logs/agent.jsonl",
...     console=True
... )

Classes

class lrs.monitoring.structured_logging.LRSLogger(agent_id: str, log_file: str | None = None, console: bool = True, level: int = 20)[source]

Bases: object

Structured logger for LRS agents.

Logs events in JSON format for easy parsing and analysis. Captures: - Precision changes - Policy selections - Tool executions - Adaptation events - Performance metrics

Examples

>>> logger = LRSLogger(agent_id="agent_1", log_file="agent.jsonl")
>>>
>>> logger.log_precision_update(
...     level='execution',
...     old_value=0.8,
...     new_value=0.4,
...     prediction_error=0.95
... )
>>>
>>> logger.log_tool_execution(
...     tool_name="api_fetch",
...     success=False,
...     execution_time=0.5,
...     prediction_error=0.9,
...     error_message="Timeout"
... )

Structured JSON logger for production monitoring.

Logs all agent events as structured JSON for integration with monitoring systems like ELK, Datadog, CloudWatch, or Grafana.

Methods:

log_precision_update(level: str, old_value: float, new_value: float, prediction_error: float, propagated: bool = False)[source]

Log precision update event.

Parameters:

  • level – Precision level (abstract/planning/execution)

  • old_value – Previous precision value

  • new_value – New precision value

  • prediction_error – Triggering prediction error

  • propagated – Whether error propagated from lower level

Log precision parameter update.

log_policy_selection(policies: list, selected_index: int, G_values: list, precision: float)[source]

Log policy selection via G.

Parameters:

  • policies – List of candidate policies

  • selected_index – Index of selected policy

  • G_values – Expected Free Energy values

  • precision – Current precision value

Log policy selection and G values.

log_tool_execution(tool_name: str, success: bool, execution_time: float, prediction_error: float, error_message: str | None = None)[source]

Log tool execution.

Parameters:

  • tool_name – Name of executed tool

  • success – Whether execution succeeded

  • execution_time – Execution time in seconds

  • prediction_error – Observed prediction error

  • error_message – Error message if failed

Log tool execution result.

log_adaptation_event(trigger: str, old_precision: Dict[str, float], new_precision: Dict[str, float], action_taken: str)[source]

Log adaptation event.

Parameters:

  • trigger – What triggered the adaptation

  • old_precision – Precision before adaptation

  • new_precision – Precision after adaptation

  • action_taken – Action taken by agent

Log adaptation trigger and response.

log_performance_metrics(total_steps: int, success_rate: float, avg_precision: float, adaptation_count: int, execution_time: float)[source]

Log aggregate performance metrics.

Parameters:

  • total_steps – Total execution steps

  • success_rate – Overall success rate

  • avg_precision – Average precision value

  • adaptation_count – Number of adaptations

  • execution_time – Total execution time

Log aggregate performance metrics.

log_error(error_type: str, message: str, stack_trace: str | None = None)[source]

Log error event.

Parameters:

  • error_type – Type of error

  • message – Error message

  • stack_trace – Optional stack trace

Log error or exception.

Example:

from lrs.monitoring.structured_logging import create_logger_for_agent

logger = create_logger_for_agent(
    agent_id="production_agent_1",
    log_file="logs/agent.jsonl",
    console=True
)

# Log tool execution
logger.log_tool_execution(
    tool_name="fetch_api",
    success=False,
    execution_time=150.5,
    prediction_error=0.9,
    error_message="Timeout after 30s"
)

# Log adaptation
logger.log_adaptation_event(
    trigger="precision_threshold",
    old_precision=0.6,
    new_precision=0.4,
    action="explore_alternatives"
)

# Log performance metrics
logger.log_performance_metrics(
    total_steps=50,
    success_rate=0.8,
    avg_precision=0.65,
    adaptation_count=3,
    total_time=125.3
)
__init__(agent_id: str, log_file: str | None = None, console: bool = True, level: int = 20)[source]

Initialize structured logger.

Parameters:

  • agent_id – Unique identifier for this agent

  • log_file – Optional file path for JSON logs

  • console – Whether to also log to console

  • level – Logging level

log_precision_update(level: str, old_value: float, new_value: float, prediction_error: float, propagated: bool = False)[source]

Log precision update event.

Parameters:

  • level – Precision level (abstract/planning/execution)

  • old_value – Previous precision value

  • new_value – New precision value

  • prediction_error – Triggering prediction error

  • propagated – Whether error propagated from lower level

log_policy_selection(policies: list, selected_index: int, G_values: list, precision: float)[source]

Log policy selection via G.

Parameters:

  • policies – List of candidate policies

  • selected_index – Index of selected policy

  • G_values – Expected Free Energy values

  • precision – Current precision value

log_tool_execution(tool_name: str, success: bool, execution_time: float, prediction_error: float, error_message: str | None = None)[source]

Log tool execution.

Parameters:

  • tool_name – Name of executed tool

  • success – Whether execution succeeded

  • execution_time – Execution time in seconds

  • prediction_error – Observed prediction error

  • error_message – Error message if failed

log_adaptation_event(trigger: str, old_precision: Dict[str, float], new_precision: Dict[str, float], action_taken: str)[source]

Log adaptation event.

Parameters:

  • trigger – What triggered the adaptation

  • old_precision – Precision before adaptation

  • new_precision – Precision after adaptation

  • action_taken – Action taken by agent

log_performance_metrics(total_steps: int, success_rate: float, avg_precision: float, adaptation_count: int, execution_time: float)[source]

Log aggregate performance metrics.

Parameters:

  • total_steps – Total execution steps

  • success_rate – Overall success rate

  • avg_precision – Average precision value

  • adaptation_count – Number of adaptations

  • execution_time – Total execution time

log_error(error_type: str, message: str, stack_trace: str | None = None)[source]

Log error event.

Parameters:

  • error_type – Type of error

  • message – Error message

  • stack_trace – Optional stack trace

Functions

lrs.monitoring.structured_logging.create_logger_for_agent(agent_id: str, **kwargs) → LRSLogger[source]

Create logger for LRS agent.

Parameters:

  • agent_id – Agent identifier

  • **kwargs – Passed to LRSLogger

Returns:

Configured logger instance

Examples

>>> logger = create_logger_for_agent(
...     "production_agent_1",
...     log_file="logs/agent.jsonl",
...     console=True
... )

Create configured logger for an agent.

Parameters:

  • agent_id – Unique agent identifier

  • log_file – Path to log file (JSONL format)

  • console – Also log to console (default: False)

  • level – Logging level (default: INFO)

Returns:

Configured LRSLogger instance

Example:

logger = create_logger_for_agent(
    agent_id="my_agent",
    log_file="/var/log/lrs/agent.jsonl",
    console=True,
    level="DEBUG"
)

Dashboard (Optional)

Note

The dashboard module requires streamlit and matplotlib which are optional dependencies. Install with: pip install lrs-agents[monitoring]

Real-time Streamlit dashboard for LRS agents.

Provides visualization of: - Precision trajectories (3-level hierarchy) - G-space map (epistemic vs pragmatic) - Prediction error stream - Adaptation timeline - Tool usage statistics

lrs.monitoring.dashboard.create_dashboard(tracker: LRSStateTracker)[source]

Create Streamlit dashboard for LRS agent monitoring.

Parameters:

tracker – LRSStateTracker instance with execution history

Examples

>>> import streamlit as st
>>> from lrs.monitoring import create_dashboard
>>>
>>> tracker = LRSStateTracker()
>>> # ... run agent with tracker ...
>>>
>>> create_dashboard(tracker)
lrs.monitoring.dashboard.run_dashboard(tracker: LRSStateTracker | None = None)[source]

Run dashboard as standalone Streamlit app.

Parameters:

tracker – Optional pre-populated tracker

Examples

>>> # In terminal:
>>> # streamlit run lrs/monitoring/dashboard.py
>>>
>>> # Or programmatically:
>>> from lrs.monitoring import run_dashboard
>>> run_dashboard()

The dashboard provides real-time visualization of agent execution including:

  • Precision trajectories - Execution, planning, and abstract levels over time

  • Prediction error stream - Real-time error monitoring

  • Tool usage statistics - Success rates and reliability by tool

  • Adaptation timeline - When and why adaptations occurred

  • Execution history - Complete step-by-step trace

Example Usage:

from lrs.monitoring.dashboard import run_dashboard
from lrs.monitoring.tracker import LRSStateTracker

tracker = LRSStateTracker()

# Run dashboard (blocks)
run_dashboard(tracker, port=8501)
# Dashboard available at http://localhost:8501

Command Line:

# Start dashboard
python -m lrs.monitoring.dashboard --port 8501

# Or with Streamlit directly
streamlit run lrs/monitoring/dashboard.py