babylon.rag.lifecycle

Object lifecycle management for the RAG system.

This module provides lifecycle state tracking for RAG objects, allowing monitoring of object states through their lifetime.

Classes

LifecycleManager()	Manages the lifecycle of RAG objects.
ObjectState(*values)	Possible states for RAG objects in their lifecycle.
PerformanceMetrics([hot_objects, ...])	Performance metrics for RAG operations.

class babylon.rag.lifecycle.ObjectState(*values)

Bases: Enum

Possible states for RAG objects in their lifecycle.

CREATED = 'created'

PREPROCESSED = 'preprocessed'

CHUNKED = 'chunked'

EMBEDDED = 'embedded'

STORED = 'stored'

RETRIEVED = 'retrieved'

REMOVED_FROM_CONTEXT = 'removed_from_context'

ARCHIVED = 'archived'

ERROR = 'error'

class babylon.rag.lifecycle.PerformanceMetrics(hot_objects=<factory>, cache_hit_rate=<factory>, avg_token_usage=0.0, latency_stats=<factory>, memory_profile=<factory>)

Bases: object

Performance metrics for RAG operations.

Parameters:

• hot_objects (list[str])

• cache_hit_rate (dict[str, float])

• avg_token_usage (float)

• latency_stats (dict[str, float])

• memory_profile (dict[str, float])

hot_objects: list[str]

cache_hit_rate: dict[str, float]

avg_token_usage: float = 0.0

latency_stats: dict[str, float]

memory_profile: dict[str, float]

__init__(hot_objects=<factory>, cache_hit_rate=<factory>, avg_token_usage=0.0, latency_stats=<factory>, memory_profile=<factory>)

Parameters:

• hot_objects (list[str])

• cache_hit_rate (dict[str, float])

• avg_token_usage (float)

• latency_stats (dict[str, float])

• memory_profile (dict[str, float])

Return type:

None

class babylon.rag.lifecycle.LifecycleManager

Bases: object

Manages the lifecycle of RAG objects.

Tracks object states, transitions, and provides metrics for monitoring system health.

__init__()

Initialize the lifecycle manager.

Return type:

None

register_object(object_id, initial_state=ObjectState.CREATED)

Register a new object for lifecycle tracking.

Parameters:

• object_id (str) – Unique identifier for the object

• initial_state (ObjectState) – Initial state for the object

Return type:

None

update_object_state(object_id, new_state)

Update the state of a tracked object.

Parameters:

• object_id (str) – Unique identifier for the object

• new_state (ObjectState | str) – New state for the object (ObjectState enum or string)

Return type:

None

get_object_state(object_id)

Get the current state of an object.

Parameters:

object_id (str) – Unique identifier for the object

Return type:

ObjectState | str | None

Returns:

Current state or None if object not found

get_object(object_id)

Get full object information.

Parameters:

object_id (str) – Unique identifier for the object

Return type:

dict[str, Any] | None

Returns:

Object data dict or None if not found

get_state_history(object_id)

Get the state transition history for an object.

Parameters:

object_id (str) – Unique identifier for the object

Return type:

list[tuple[ObjectState, datetime]]

Returns:

List of (state, timestamp) tuples

get_objects_in_state(state)

Get all objects currently in a given state.

Parameters:

state (ObjectState) – The state to filter by

Return type:

list[str]

Returns:

List of object IDs in the specified state

get_performance_metrics()

Get current performance metrics.

Return type:

PerformanceMetrics

Returns:

PerformanceMetrics dataclass with current stats

clear()

Clear all tracked objects and history.

Return type:

None