Skip to main content
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 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)

Build docs developers (and LLMs) love

Get started for freeTalk to us