Shared memory

• Four isolation scopes -- global, session, actor, and workflow, from widest to narrowest.
• Key-value storage -- set, get, delete, list_keys, exists across all scopes.
• Vector storage -- store embeddings and run similarity search for RAG patterns.
• Hierarchical lookup -- Python and TypeScript memory clients search workflow -> session -> actor -> global when you call get() without an explicit scope.
• Reactive events -- subscribe to memory changes with pattern matching (Python and TypeScript).
• Scoped accessors -- memory.session(id), memory.actor(id), memory.workflow(id), memory.global_scope for explicit scope targeting.
• Go default scope -- memory.Get() reads session scope (or the current run when no session is present). Use GlobalScope(), SessionScope(), UserScope(), and WorkflowScope() for explicit access.

Hierarchical lookup order (most specific first):

workflow -> session -> actor -> global

Python and TypeScript clients check each scope in order and return the first match when you call memory.get("key") without specifying a scope. Values in narrower scopes override broader scopes. Go Memory.Get() reads session scope by default; use explicit scoped accessors for other scopes.

Store embeddings and run similarity search for RAG, semantic routing, and knowledge retrieval patterns.

@app.reasoner()
async def index_and_search(text: str) -> dict:
    # Store a vector with metadata
    embedding = [0.1, 0.2, 0.3, 0.4]  # from your embedding model
    await app.memory.set_vector(
        "doc-001",
        embedding,
        metadata={"source": "manual", "topic": "billing"},
    )

    # Similarity search
    query_embedding = [0.1, 0.2, 0.35, 0.4]
    results = await app.memory.similarity_search(
        query_embedding,
        top_k=5,
        filters={"topic": "billing"},
    )

    # Delete a vector
    await app.memory.delete_vector("doc-001")

    # Scoped vector operations
    global_mem = app.memory.global_scope
    await global_mem.set_vector("shared-embedding", embedding, metadata={"public": True})
    global_results = await global_mem.similarity_search(query_embedding, top_k=3)

    return {"results": results}

Subscribe to memory changes with pattern matching. When a key matching your pattern is written in any scope, your handler fires. This enables reactive patterns where one agent's write triggers another agent's logic.

Availability: Python and TypeScript only. Go does not support memory events.

# Decorator-based subscription
@app.on_change("user.*")
async def on_user_change(event):
    print(f"Key changed: {event.key}")
    print(f"New data: {event.data}")
    print(f"Scope: {event.scope}")

# Scoped event subscription
@app.on_change("config.*")
async def on_config_change(event):
    print(f"Global config changed: {event.key}")

# Session-scoped events
session = app.memory.session("session-123")

@session.on_change("conversation.*")
async def on_conversation_change(event):
    print(f"Conversation updated: {event.key}")

Pattern matching uses wildcards: "user.*" matches "user.preferences", "user.history", etc.

Conversation History

Use session-scoped memory to maintain conversation state across turns.

@app.reasoner()
async def chat(message: str) -> dict:
    history = await app.memory.get("conversation.history", default=[])
    history.append({"role": "user", "content": message})

    response = await app.ai(
        system="You are a helpful assistant.",
        user=message,
        context={"history": history},
    )

    history.append({"role": "assistant", "content": response})
    await app.memory.set("conversation.history", history)

    return {"response": response}

Cross-Agent State Sharing

Use global memory as a shared blackboard between agents.

# Agent A: Write results to global memory
@app.reasoner()
async def analyzer(data: dict) -> dict:
    result = await app.ai(system="Analyze this data.", user=str(data))
    await app.memory.global_scope.set(f"analysis.{data['id']}", result)
    return result

# Agent B: Read results from global memory
@app.reasoner()
async def reporter(analysis_id: str) -> dict:
    analysis = await app.memory.global_scope.get(f"analysis.{analysis_id}")
    if not analysis:
        return {"error": "Analysis not found"}
    report = await app.ai(system="Generate a report.", user=str(analysis))
    return {"report": report}

Reactive Processing with Events

Trigger processing when data arrives without polling.

@app.on_change("analysis.*")
async def on_analysis_complete(event):
    analysis_id = event.key.split(".")[-1]
    report = await app.call(
        "reporter.generate",
        analysis_id=analysis_id,
        data=event.data,
    )
    await app.memory.global_scope.set(f"report.{analysis_id}", report)

Key-Value Methods

Vector Methods

Scope Accessors

Event Methods

Go -- GetTyped() for Type-Safe Retrieval

The Go SDK provides GetTyped() to deserialize memory values directly into Go structs, avoiding manual type assertions.

type UserPrefs struct {
    Tone     string `json:"tone"`
    Language string `json:"language"`
    Theme    string `json:"theme"`
}

// GetTyped deserializes the stored value into the target struct
// Available on ScopedMemory (e.g., mem.SessionScope(), mem.GlobalScope())
var prefs UserPrefs
err := mem.SessionScope().GetTyped(ctx, "preferences", &prefs)
if err != nil {
    // key not found or deserialization error
}
fmt.Println(prefs.Tone) // "concise"

// Works with any scope
var config AppConfig
err = mem.GlobalScope().GetTyped(ctx, "app_config", &config)

Signature:

func (s *ScopedMemory) GetTyped(ctx context.Context, key string, dest any) error

Go -- MemoryBackend Pluggable Interface

The Go SDK uses a pluggable backend interface for memory storage. Swap backends by passing them in agent.Config.

// MemoryBackend is the interface all storage backends implement
type MemoryBackend interface {
    Set(scope MemoryScope, scopeID, key string, value any) error
    Get(scope MemoryScope, scopeID, key string) (any, bool, error)
    Delete(scope MemoryScope, scopeID, key string) error
    List(scope MemoryScope, scopeID string) ([]string, error)
    GetVector(scope MemoryScope, scopeID, key string) (embedding []float64, metadata map[string]any, found bool, err error)
    SetVector(scope MemoryScope, scopeID, key string, embedding []float64, metadata map[string]any) error
    SearchVector(scope MemoryScope, scopeID string, embedding []float64, opts SearchOptions) ([]VectorSearchResult, error)
    DeleteVector(scope MemoryScope, scopeID, key string) error
}

Built-in backends:

// Testing: in-memory (default)
a, _ := agent.New(agent.Config{
    NodeID:        "test-agent",
    Version:       "1.0.0",
    MemoryBackend: agent.NewInMemoryBackend(),
})

// Production: control plane backend
a, _ := agent.New(agent.Config{
    NodeID:        "prod-agent",
    Version:       "1.0.0",
    MemoryBackend: agent.NewControlPlaneMemoryBackend(
        "http://localhost:8080",
        "bearer-token",
        "prod-agent",
    ),
})

MemoryConfig (Python)

Configure memory behavior in the Python SDK via MemoryConfig.

from agentfield import Agent, MemoryConfig

app = Agent(
    node_id="my-agent",
    memory_config=MemoryConfig(
        auto_inject=["conversation.history"],  # keys to auto-inject into context
        memory_retention="session",             # retention policy
        cache_results=True,                     # cache memory results locally
    ),
)