Memory

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")

search()

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

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

Use descriptive keys

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)

Add metadata for filtering

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"}
)

Clean up temporary data

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")

Avoid storing large objects

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 ...]})

Use semantic search effectively

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

Getting Started

Agentic

Application

Integrations

Deployment

What is Memory?

Core Operations

add()

get()

search()

remove()

Usage Patterns

Store Intermediate Results

Semantic Search for Context

Metadata for Organization

Clean Up Temporary Data

Deduplication

Memory Durability & Agent Handoffs

Durable Across Failures

Shared Across Agent Handoffs

Best Practices

Limitations

Examples

Complete Flight Search Example

Next Steps

Actions

Agents

Integrations

API Reference

Getting Started

Agentic

Application

Integrations

Deployment

Documentation Index

​What is Memory?

​Core Operations

​add()

​get()

​search()

​remove()

​Usage Patterns

​Share Data Between Actions

​Store Intermediate Results

​Semantic Search for Context

​Metadata for Organization

​Clean Up Temporary Data

​Deduplication

​Memory Durability & Agent Handoffs

​Durable Across Failures

​Shared Across Agent Handoffs

​Best Practices

​Limitations

​Examples

​Complete Flight Search Example

​Next Steps

Actions

Agents

Integrations

API Reference

What is Memory?

Core Operations

add()

get()

search()

remove()

Usage Patterns

Share Data Between Actions

Store Intermediate Results

Semantic Search for Context

Metadata for Organization

Clean Up Temporary Data

Deduplication

Memory Durability & Agent Handoffs

Durable Across Failures

Shared Across Agent Handoffs

Best Practices

Limitations

Examples

Complete Flight Search Example

Next Steps