Preskoči na vsebino

Example: Workflow Agent

This example builds an employee onboarding workflow agent. Unlike a one-shot task, this agent handles the same type of request repeatedly, so memory is central: each completed onboarding makes the next one faster and more reliable.

The Domain​

An HR team needs to onboard new employees. Each onboarding involves: document collection, account provisioning, equipment assignment, training enrollment, and manager check-in. The workflow varies by department and role.

Setup​

from masar import MasarClient

client = MasarClient()

# Register the domain (one-time setup)
client.domains.register(
    name="onboarding",
    entity_types=["Employee", "Document", "Account", "Equipment", "Training"],
    state_labels=["pending", "in_progress", "completed", "blocked", "skipped"],
    event_labels=["START", "SUBMIT", "APPROVE", "PROVISION", "ASSIGN", "ENROLL", "COMPLETE"]
)

First Onboarding: No Memory​

def handle_onboarding(request: dict):
    # Check memory
    memories = client.memory.recall(
        context=f"Onboard {request['role']} in {request['department']}",
        domain="onboarding"
    )

    if memories.pattern:
        print(f"Following pattern: {memories.pattern.name}")
        plan = client.plan_instructions(
            current={"name": "Onboarding", "orbitals": []},
            goal=memories.pattern.schema,
            domain="onboarding"
        )
    else:
        print("No pattern found. Planning from scratch.")
        plan = client.plan_instructions(
            current={"name": "Onboarding", "orbitals": []},
            goal="full-onboarding",
            domain="onboarding"
        )

    # Execute the plan
    schema = execute_plan(plan)

    # Verify
    check = client.verify(schema=schema, domain="onboarding")

    # Store the episode
    client.memory.store(
        domain="onboarding",
        context=f"Onboard {request['role']} in {request['department']}",
        schema=schema,
        outcome="success" if check.valid else "failure",
        metadata={"department": request["department"], "role": request["role"]}
    )

    return schema

Week 1: Building Experience​

# Monday
handle_onboarding({"role": "engineer", "department": "platform"})  # 12 steps, 8.2s
handle_onboarding({"role": "engineer", "department": "frontend"})  # 12 steps, 7.9s

# Wednesday
handle_onboarding({"role": "designer", "department": "product"})   # 10 steps, 6.5s

# Friday: compress the week's episodes
client.memory.compress(domain="onboarding")
# Extracted 2 patterns: "engineer-onboarding", "designer-onboarding"

Week 4: Pattern-Driven​

handle_onboarding({"role": "engineer", "department": "data"})
# Output:
# Following pattern: engineer-onboarding (similarity: 0.96)
# 4 steps (skipped 8 known-good steps), 2.1s

The agent now completes engineer onboarding in 2 seconds instead of 8. It still plans (to handle department-specific variations) but uses the pattern as a starting point instead of building from zero.

Handling Variations​

When a new role appears that doesn't match existing patterns:

handle_onboarding({"role": "legal-counsel", "department": "legal"})
# Output:
# No pattern found. Planning from scratch.
# 14 steps, 9.8s (includes compliance-specific states)

After a few legal onboardings, a new "legal-onboarding" pattern emerges. The agent self-organizes its expertise by role category.

Monitoring Memory Health​

stats = client.memory.stats(domain="onboarding")
print(f"Episodes: {stats.total_episodes}")
print(f"Patterns: {stats.patterns}")
print(f"Storage: {stats.storage_used_mb:.1f} MB")

# Periodic cleanup
client.memory.forget(domain="onboarding", older_than_days=180, min_relevance=0.2)

Next Steps​