gigaevo-memory 0.1.0

Python client for GigaEvo Memory Module — persistent storage for CARL artifacts

gigaevo-memory

Python client for the GigaEvo Memory Module — persistent storage for CARL artifacts (steps, chains, agents, concepts).

CARL (mmar-carl) is a required dependency and is installed automatically.

Installation

pip install gigaevo-memory

Quick Start

from mmar_carl import (
    ContextSearchConfig,
    ReasoningChain,
    StepDescription,
    ReasoningContext
)
from gigaevo_memory import MemoryClient

# Define a reasoning chain with RAG-like context queries
steps = [
    StepDescription(
        number=1,
        title="Initial Data Assessment",
        aim="Assess the quality and completeness of input data",
        reasoning_questions="What data patterns and anomalies are present?",
        step_context_queries=["data quality indicators", "missing values", "data consistency"],
        stage_action="Evaluate data reliability and identify potential issues",
        example_reasoning="High-quality data enables more reliable analysis and predictions"
    ),
    StepDescription(
        number=2,
        title="Pattern Recognition",
        aim="Identify significant patterns and trends in the data",
        reasoning_questions="What trends and correlations emerge from the analysis?",
        dependencies=[1],  # Depends on data quality assessment
        step_context_queries=["growth trends", "performance indicators", "correlation patterns"],
        stage_action="Analyze temporal patterns and statistical relationships",
        example_reasoning="Pattern recognition helps identify underlying business drivers and opportunities"
    )
]

chain = ReasoningChain(
    steps=steps,
    max_workers=3,
    metadata={},
    search_config=ContextSearchConfig(strategy="substring"),
)

client = MemoryClient(base_url="http://localhost:8000")

# Save a chain (accepts ReasoningChain or dict)
ref = client.save_chain(chain=chain, name="my_chain", tags=["finance"])

# Get chain as ReasoningChain (full DAG validation)
chain = client.get_chain(ref.entity_id, channel="stable")

# Get chain as raw dict
chain_dict = client.get_chain_dict(ref.entity_id)

# Search (q matches name and when_to_use; use tags for faceted filter)
hits = client.search(q="my_chain", entity_type="chain")

# Search by tag (e.g. "finance")
hits = client.search(tags=["finance"])
hits = client.search(entity_type="chain", tags=["finance"])

# Full-text + tag: name/when_to_use contains "triage" and tagged "finance"
hits = client.search(q="triage", entity_type="chain", tags=["finance"])

# Multiple tags (entity must have all)
hits = client.search(tags=["finance", "triage"], entity_type="chain")

# Watch for changes — callback fires when a new version is published
sub = client.watch_chain(
    ref.entity_id,
    callback=lambda new_chain: print(f"Chain updated: {new_chain}"),
)

Cache Policies

from gigaevo_memory import MemoryClient, CachePolicy

# TTL-based (default: 5 minutes)
client = MemoryClient(base_url="http://localhost:8000", cache_policy=CachePolicy.TTL, cache_ttl=300)

# Freshness check (conditional GET with ETag)
client = MemoryClient(base_url="http://localhost:8000", cache_policy=CachePolicy.FRESHNESS_CHECK)

# SSE push (reactive invalidation)
client = MemoryClient(base_url="http://localhost:8000", cache_policy=CachePolicy.SSE_PUSH)

Development

make client-install    # Install in editable mode
make client-test       # Run tests
make client-lint       # Lint
make client-build      # Build sdist + wheel