Skip to main content

​
What is Memory?

Memory is Ziet’s built-in durable storage system that persists data across actions, agent steps, and even failures. It provides:
  • Durable storage - Data persists even if agent steps fail
  • Key-value storage - Store and retrieve by unique keys
  • Semantic search - Find data by meaning, not just exact matches
  • Metadata filtering - Tag and filter entries
  • Automatic scoping - Data is isolated per run by default
  • Cross-agent sharing - Agents can hand off and share memory
Important: Memory is durable and survives failures. If an agent step fails, memory is preserved and the workflow can resume from where it left off. No database setup required - just import and use. 
from ziet import memory

# Store data (key comes first)
memory.add(key="best_flight", value={"price": 350})

# Retrieve by key
best = memory.get("best_flight")  # Returns {"price": 350}

# Search by meaning
results = memory.search("cheapest flight option")

# Remove when done
memory.remove(key="best_flight")

​
Core Operations

​
add()

Store data in memory: 
memory.add(
    key="user_data",                      # Required: semantic label
    value={"user": "alice", "score": 95},
    metadata={"type": "user", "v": 1}    # Optional: tags for filtering
)
Parameters:
​
key
string
required
Semantic label for easy retrieval. Always provide a descriptive key.
memory.add(key="flight_options", value=data)
memory.add(key="user_preferences", value=data)
​
value
any
required
Any JSON-serializable data (dict, list, str, int, bool, etc.)
memory.add(key="user_name", value={"name": "Alice"})
memory.add(key="numbers", value=[1, 2, 3])
memory.add(key="note", value="important note")
memory.add(key="count", value=42)
​
metadata
dict
Additional tags for filtering and organization.
memory.add(
    key="flight_option",
    value=data,
    metadata={
        "type": "flight",
        "airline": "United",
        "priority": "high"
    }
)
Returns: entry_id (string) - Unique identifier for the stored entry Example: 
entry_id = memory.add(
    key="best_option",
    value={"flight": "UA123", "price": 350},
    metadata={"type": "flight", "airline": "United"}
)
print(f"Stored with ID: {entry_id}")

​
get()

Retrieve the most recent value for a key: 
value = memory.get("best_option")
Parameters:
​
key
string
required
The key to retrieve
Returns: The stored value, or None if key doesn’t exist Example: 
# Store data
memory.add(key="best_flight", value={"price": 350})

# Retrieve data
flight = memory.get("best_flight")
if flight:
    print(f"Price: ${flight['price']}")
else:
    print("No flight data found")
Semantic or filtered search across all memories: 
results = memory.search(
    query="cheapest flight under $400",  # Optional: natural language query
    metadata={"type": "flight"},          # Optional: filter by metadata
    limit=10                              # Optional: max results
)
Parameters:
​
query
string
Natural language or keyword query. Searches across both values and keys.
memory.search("cheapest option")
memory.search("flights to new york")
memory.search("user preferences")
​
metadata
dict
Filter by metadata fields
memory.search(metadata={"type": "flight"})
memory.search(metadata={"airline": "United", "priority": "high"})
​
limit
int
default:"10"
Maximum number of results to return
memory.search("flights", limit=5)
Returns: List of matching entries with scores 
[
    {
        "entry_id": "1234567890_abc",
        "key": "flight_option_1",
        "value": {"price": 350, "airline": "United"},
        "metadata": {"type": "flight"},
        "score": 0.95  # Relevance score (0-1)
    },
    ...
]
Examples: 
# Semantic search
results = memory.search("what was the cheapest flight?")
for entry in results:
    print(f"Score: {entry['score']}, Value: {entry['value']}")

# Metadata filter only
flights = memory.search(metadata={"type": "flight"})

# Combined
high_priority = memory.search(
    query="urgent",
    metadata={"priority": "high"},
    limit=5
)

​
remove()

Delete entries by entry_id or key: 
# Remove specific entry
memory.remove(entry_id="1234567890_abc")

# Remove all entries with a key
memory.remove(key="temp_data")
Parameters:
​
entry_id
string
Specific entry to remove
entry_id = memory.add(data, key="temp")
memory.remove(entry_id=entry_id)
​
key
string
Remove all entries with this key
memory.remove(key="temp_data")
Returns: Number of deleted entries (int) Examples: 
# Remove by entry_id
entry_id = memory.add(key="temporary", value={"temp": True})
count = memory.remove(entry_id=entry_id)
print(f"Removed {count} entries")

# Remove by key
memory.add(key="data", value={"x": 1})
memory.add(key="data", value={"x": 2})  # Same key
count = memory.remove(key="data")  # Removes both
print(f"Removed {count} entries")  # Output: Removed 2 entries

​
Usage Patterns

​
Share Data Between Actions

Store data in one action, retrieve in another: 
from ziet import Action, memory

@Action(
    id="fetch_user",
    name="Fetch User",
    description="Fetch user data from database"
)
def fetch_user(user_id: str) -> None:
    user = database.get_user(user_id)
    
    # Store for later use
    memory.add(key=f"user_{user_id}", value=user)


@Action(
    id="send_notification",
    name="Send Notification",
    description="Send notification email"
)
def send_notification(user_id: str, message: str) -> None:
    # Retrieve cached user data
    user = memory.get(f"user_{user_id}")
    
    if not user:
        # Fetch if not in memory
        fetch_user(user_id)
        user = memory.get(f"user_{user_id}")
    
    result = send_email(user["email"], message)
    memory.add(key="notification_sent", value=result)

​
Store Intermediate Results

Actions automatically store progress in memory: 
from ziet import Agent, Action, memory

@Action(
    id="search_topic",
    name="Search Topic",
    description="Search for information on topic"
)
def search_topic(topic: str) -> None:
    results = search(topic)
    memory.add(key="search_results", value=results)

@Action(
    id="filter_results",
    name="Filter Results",
    description="Filter search results"
)
def filter_results() -> None:
    results = memory.get("search_results")
    filtered = filter(results)
    memory.add(key="filtered_results", value=filtered)

@Action(
    id="summarize_results",
    name="Summarize Results",
    description="Create summary from filtered results"
)
def summarize_results() -> None:
    filtered = memory.get("filtered_results")
    summary = summarize(filtered)
    memory.add(key="final_summary", value=summary)

@Agent(
    id="research_agent",
    name="ResearchAgent",
    instructions="""
    1. Use search_topic action to search the topic
    2. Use filter_results action to filter results
    3. Use summarize_results action to create final summary
    All results are stored in memory automatically.
    """,
    actions=["search_topic", "filter_results", "summarize_results"]
)
class ResearchAgent:
    pass

​
Semantic Search for Context

Find relevant information from past actions: 
from ziet import Action, memory
from ziet.integrations import openai

@Action(
    id="answer_question",
    name="Answer Question",
    description="Answer question using memory context"
)
def answer_question(question: str) -> None:
    # Find relevant context from memory
    context = memory.search(question, limit=3)
    
    # Extract values
    relevant_data = [entry["value"] for entry in context]
    
    # Generate answer with context
    prompt = f"Question: {question}\nContext: {relevant_data}"
    answer = openai.chat([{"role": "user", "content": prompt}])
    
    # Store answer in memory
    memory.add(key=f"answer_{question[:20]}", value=answer)

​
Metadata for Organization

Tag and filter entries: 
# Store with metadata
memory.add(
    key="flight_1",
    value={"flight": "UA123", "price": 350},
    metadata={"type": "flight", "airline": "United", "status": "available"}
)

memory.add(
    key="hotel_1",
    value={"hotel": "Marriott", "price": 200},
    metadata={"type": "hotel", "chain": "Marriott", "status": "available"}
)

# Find all flights
flights = memory.search(metadata={"type": "flight"})

# Find available options
available = memory.search(metadata={"status": "available"})

# Find United flights
united_flights = memory.search(metadata={"type": "flight", "airline": "United"})

​
Clean Up Temporary Data

Remove data you don’t need anymore: 
from ziet import Action, memory

@Action(
    id="process_data",
    name="Process Data",
    description="Process data with temporary storage"
)
def process_data(data: dict) -> None:
    # Store temporary working data
    entry_id = memory.add(key="temp_working_data", value=data)
    
    # Process
    result = expensive_computation(data)
    
    # Store final result
    memory.add(key="final_result", value=result)
    
    # Clean up temporary data
    memory.remove(entry_id=entry_id)

​
Deduplication

Avoid processing the same data twice: 
from ziet import Action, memory

@Action(
    id="process_item",
    name="Process Item",
    description="Process item with caching"
)
def process_item(item_id: str) -> None:
    # Check if already processed
    existing = memory.get(f"processed_{item_id}")
    if existing:
        # Already cached, mark as skipped
        memory.add(key="processing_status", value="cached")
        return
    
    # Process
    result = expensive_operation(item_id)
    
    # Cache result
    memory.add(key=f"processed_{item_id}", value=result)
    memory.add(key="processing_status", value="processed")

​
Memory Durability & Agent Handoffs

​
Durable Across Failures

Memory persists even when agent steps fail: 
from ziet import Agent, Action, memory

@Action(
    id="risky_operation",
    name="Risky Operation",
    description="Operation that might fail"
)
def risky_operation() -> None:
    result = dangerous_computation()
    memory.add(key="operation_result", value=result)

@Agent(
    id="resilient_agent",
    name="ResilientAgent",
    instructions="""
    1. Store progress in memory with key 'progress'
    2. Execute risky_operation action
    3. If it fails, memory is preserved and workflow can retry from last step
    
    Memory is durable - even if operations fail, stored data persists.
    """,
    actions=["risky_operation"]
)
class ResilientAgent:
    pass

# Even if risky_operation fails, memory.get("progress") is still available

​
Shared Across Agent Handoffs

When agents hand off to each other, memory is preserved and shared: 
from ziet import Agent, Action, HandOffStrategy, memory

@Action(
    id="search_data",
    name="Search Data",
    description="Search for data"
)
def search_data(query: str) -> None:
    results = search(query)
    memory.add(key="search_results", value=results)

@Action(
    id="process_data",
    name="Process Data",
    description="Process search results"
)
def process_data() -> None:
    results = memory.get("search_results")
    processed = process(results)
    memory.add(key="processed_results", value=processed)

@Action(
    id="finalize_data",
    name="Finalize Data",
    description="Finalize processed results"
)
def finalize_data() -> None:
    search_results = memory.get("search_results")
    processed = memory.get("processed_results")
    final = finalize(search_results, processed)
    memory.add(key="final_result", value=final)

# Agent 1: Searcher
@HandOffStrategy(agents=["processor_agent"])
@Agent(
    id="searcher_agent",
    name="SearcherAgent",
    instructions="""
    1. Use search_data action to search
    2. Results stored in memory with key 'search_results'
    3. Hand off to processor_agent
    
    Memory persists across the handoff.
    """,
    actions=["search_data"]
)
class SearcherAgent:
    pass

# Agent 2: Processor (receives memory from Agent 1)
@HandOffStrategy(agents=["finalizer_agent"])
@Agent(
    id="processor_agent",
    name="ProcessorAgent",
    instructions="""
    1. Access 'search_results' from memory (set by SearcherAgent)
    2. Use process_data action to process results
    3. Results stored in memory with key 'processed_results'
    4. Hand off to finalizer_agent
    
    Memory from previous agent is accessible.
    """,
    actions=["process_data"]
)
class ProcessorAgent:
    pass

# Agent 3: Finalizer (receives memory from both previous agents)
@Agent(
    id="finalizer_agent",
    name="FinalizerAgent",
    instructions="""
    1. Access 'search_results' from memory (from SearcherAgent)
    2. Access 'processed_results' from memory (from ProcessorAgent)
    3. Use finalize_data action to create final result
    
    All memory from previous agents is available.
    """,
    actions=["finalize_data"]
)
class FinalizerAgent:
    pass

​
Best Practices

Make keys self-documenting✅ Good
memory.add(key="search_results_sfo_to_nyc_2024_12_25", value=flights)
memory.add(key="customer_profile_12345", value=user)
memory.add(key="final_summary_v2", value=summary)
❌ Bad
memory.add(key="data", value=flights)
memory.add(key="x", value=user)
memory.add(key="result", value=summary)
Tag entries for easy retrieval and organization
memory.add(
    key="flight_option",
    value=data,
    metadata={
        "type": "flight",
        "airline": "United",
        "price_range": "mid",
        "priority": "high",
        "stage": "research"
    }
)

# Later: powerful filtering
high_priority = memory.search(
    metadata={"priority": "high", "stage": "research"}
)
Remove data you don’t need to keep
# Store temporary data
entry_id = memory.add(key="temp", value=temp_data)

# Use it
result = process(memory.get("temp"))

# Clean up
memory.remove(key="temp")
Keep memory entries reasonably sized✅ Good: Store summaries or references
memory.add(
    key="dataset_info",
    value={
        "file_url": "s3://bucket/large_file.csv",
        "row_count": 10000,
        "summary": "User data from Q4 2024"
    }
)
❌ Bad: Store entire large objects
# Don't do this - 10,000 rows in memory
memory.add(key="data", value={"rows": [... 10000 rows ...]})
Write natural queries for better results✅ Good: Natural language
memory.search("what was the cheapest flight option?")
memory.search("user preferences for airlines")
memory.search("errors during payment processing")
❌ Bad: Single keywords (use exact key instead)
memory.search("flight")  # Too vague
memory.search("user")    # Use memory.get("user") instead

​
Limitations

Current limitations:
  • Max entry size: 6MB per entry
  • Max entries per run: 10,000 entries
  • Retention: Data retained for 30 days after run completion
  • Search results: Limited to 100 results per query
  • Cross-run memory: Not yet available (coming soon)

​
Examples

​
Complete Flight Search Example

from ziet import Action, Agent, memory
from ziet.integrations import google, apify, openai

@Action(
    id="search_flights",
    name="Search Flights",
    description="Search Google for flights"
)
def search_flights(origin: str, dest: str) -> None:
    results = google.search(f"flights {origin} to {dest}")
    memory.add(key="search_results", value=results)

@Action(
    id="scrape_prices",
    name="Scrape Prices",
    description="Scrape prices from flight websites"
)
def scrape_prices() -> None:
    search_results = memory.get("search_results")
    
    for url in search_results[:3]:  # Top 3 results
        data = apify.scrape(url)
        memory.add(
            key=f"scraped_{url}", 
            value=data,
            metadata={"type": "scrape"}
        )

@Action(
    id="summarize_options",
    name="Summarize Options",
    description="Generate AI summary of flight options"
)
def summarize_options() -> None:
    # Get data from memory
    search_results = memory.get("search_results")
    all_scrapes = memory.search(metadata={"type": "scrape"})
    scraped_data = [entry["value"] for entry in all_scrapes]
    
    # Generate summary
    prompt = f"Summarize: {search_results}, {scraped_data}"
    summary = openai.chat([{"role": "user", "content": prompt}])
    
    memory.add(key="final_summary", value=summary)

@Agent(
    id="flight_agent",
    name="FlightAgent",
    instructions="""
    1. Use search_flights action to search for flights
    2. Use scrape_prices action to get detailed pricing
    3. Use summarize_options action to generate summary
    
    All data is stored in memory and can be retrieved:
    - 'search_results': raw search results
    - 'scraped_*': scraped data for each URL
    - 'final_summary': AI-generated summary
    """,
    actions=["search_flights", "scrape_prices", "summarize_options"]
)
class FlightAgent:
    pass

​
Next Steps

Actions

Learn how to build actions that use memory

Agents

Build agents that orchestrate memory operations

Integrations

Combine memory with integrations

API Reference

Complete API documentation