ESS (Episodic State Storage)

Manage episodic state storage in AveniECA

Episodic State Storage (ESS) is the core data structure in AveniECA that stores state vectors along with associated metadata like valence, scores, and aggregates.

ESS Models

ESSInsert

Used to create or update ESS entries:

from avenieca.api.model import ESSInsert

ess = ESSInsert(
    state=[11],                    # Required: State vector
    module_id="air_conditioner",   # Required: Module identifier
    valence=10.0,                  # Required: Valence value
    avg_ess_valence=0.0,          # Optional: Average valence
    total_ess_score=0,            # Optional: Total score
    avg_ess_score=0,              # Optional: Average score
    score=0,                       # Optional: Score
    embedding_input=None,          # Optional: Embedding input ID
    aggregate_id=[],               # Optional: Aggregate IDs
    aggregate_valence=[],          # Optional: Aggregate valences
    aggregate_score=[],            # Optional: Aggregate scores
    aggregate_module_id=[],        # Optional: Aggregate module IDs
    aggregate_shape=[],            # Optional: Aggregate shapes
    aggregate_context=[],          # Optional: Aggregate contexts
    aggregate_emb_inp=[],          # Optional: Aggregate embedding inputs
    context=None                   # Optional: Context string
)

ESSResponse

Returned when retrieving ESS entries:

@dataclass
class ESSResponse(Base):
    id: int                        # Database ID
    state: List[float]             # State vector
    module_id: str                 # Module identifier
    valence: float                 # Valence value
    created_at: str                # Creation timestamp
    updated_at: str                # Last update timestamp
    avg_ess_valence: float         # Average valence
    total_ess_score: int           # Total score
    avg_ess_score: int             # Average score
    score: int                     # Score
    embedding_input: Optional[int] # Embedding input ID
    aggregate_id: List[int]        # Aggregate IDs
    aggregate_valence: List[float] # Aggregate valences
    aggregate_score: List[int]     # Aggregate scores
    aggregate_module_id: List[str] # Aggregate module IDs
    aggregate_shape: List[int]     # Aggregate shapes
    aggregate_context: list        # Aggregate contexts
    aggregate_emb_inp: list        # Aggregate embedding inputs
    context: Optional[str]         # Context string

ESS Methods

create()

Create a new ESS entry:

from avenieca.api.model import ESSInsert

ess = ESSInsert(
    module_id="air_conditioner",
    state=[11],
    valence=10.0,
    score=4,
    embedding_input=1,
    context=None
)

res, status = eca.ess.create(data=ess)
if status == 201:
    print(f"Created ESS with ID: {res.id}")

get_all()

Retrieve all ESS entries for a module:

res, status = eca.ess.get_all(module_id="air_conditioner")
if status == 200:
    for ess in res:
        print(f"ESS {ess.id}: state={ess.state}, valence={ess.valence}")

get_one()

Get a specific ESS entry by database ID:

res, status = eca.ess.get_one(module_id="air_conditioner", db_id=8)
if status == 200:
    print(f"State: {res.state}")
    print(f"Valence: {res.valence}")

update()

Update an existing ESS entry:

from avenieca.api.model import ESSInsert

updated_ess = ESSInsert(
    module_id="air_conditioner",
    state=[12],
    valence=15.0,
    score=5
)

res, status = eca.ess.update(
    module_id="air_conditioner",
    db_id=8,
    data=updated_ess
)

search()

Search for ESS entries with similar states:

from avenieca.api.model import Search

search_query = Search(
    module_id="air_conditioner",
    state=[18],
    limit=1
)

res, status = eca.ess.search(data=search_query)
if status == 200:
    for result in res:
        print(f"Score: {result.score}")
        print(f"ESS: {result.ess}")

upsert()

Upsert an ESS to the VSE (Vector Search Engine) collection:

res, status = eca.ess.upsert(module_id="aggregate001", db_id=5)
if status == 200:
    print("ESS upserted to VSE")

get_one_sequence()

Get the ESS referenced by a specific sequence:

res, status = eca.ess.get_one_sequence(
    module_id="air_conditioner",
    sequence_id=3
)

get_all_sequence()

Get all ESS entries from all sequences:

res, status = eca.ess.get_all_sequence(module_id="air_conditioner")

get_one_pretty()

Get an ESS with the state mapped to original input:

res, status = eca.ess.get_one_pretty(module_id="gwp_record", db_id=1)

get_one_with_embedding()

Get an ESS using the embedding input ID:

res, status = eca.ess.get_one_with_embedding(
    module_id="air_conditioner",
    emb_input=1
)

get_all_aggregates()

Get all aggregates that have the given ESS as an in-twin:

res, status = eca.ess.get_all_aggregates(
    module_id="team",
    aggregate_module_id="gwp_aggregate",
    ess_id=1
)

Creating Aggregates from ESS

Here’s a complete example of creating an aggregate from multiple ESS entries:

from avenieca.api.model import ESSInsert, ESSResponse
from typing import List

def create_aggregate_from_ess(array_ess: List[ESSResponse], aggregate_insert: ESSInsert):
    total_ess_score = 0
    total_ess_valence = 0.0
    for ess in array_ess:
        aggregate_insert.state.extend(ess.state)
        aggregate_insert.aggregate_module_id.append(ess.module_id)
        aggregate_insert.aggregate_id.append(ess.id)
        aggregate_insert.aggregate_context.append(ess.context)
        aggregate_insert.aggregate_valence.append(ess.valence)
        aggregate_insert.aggregate_score.append(ess.score)
        aggregate_insert.aggregate_emb_inp.append(ess.embedding_input)
        aggregate_insert.aggregate_shape.append(len(ess.state))
        total_ess_score += ess.score
        total_ess_valence += ess.valence
    aggregate_insert.total_ess_score = total_ess_score
    aggregate_insert.avg_ess_score = int(total_ess_score / len(array_ess))
    aggregate_insert.avg_ess_valence = total_ess_valence / len(array_ess)
    aggregate_insert.valence = total_ess_valence
    return aggregate_insert

# Create aggregate
aggregate_insert = ESSInsert(
    module_id="aggregate001",
    state=[],
    valence=10.0,
    aggregate_id=[],
    aggregate_valence=[],
    aggregate_score=[],
    aggregate_module_id=[],
    aggregate_shape=[]
)

agg_in = create_aggregate_from_ess([ess1, ess2, ess3], aggregate_insert)
res, status = eca.ess.create(data=agg_in)