Recipes

Support Agent Memory Loop

Build a support-agent loop with context assembly, checkpointing, reflection, and outcome reinforcement.

This is the default long-running support-agent pattern in current MuBit:

store customer facts, traces, and preferences
assemble context before each response
checkpoint before compaction or risky transitions
reflect after the attempt
record the outcome so the next similar case gets better guidance
archive exact artifacts when later steps need exact reuse instead of only semantic recall

Minimal implementation example

support_agent_memory_loop.py

from mubit import Client
import os
 
run_id = "support:acme:ticket-42"
client = Client(
    endpoint=os.getenv("MUBIT_ENDPOINT", "https://api.mubit.ai"),
    api_key=os.environ["MUBIT_API_KEY"],
    run_id=run_id,
    transport="http",
)
 
client.register_agent(
    session_id=run_id,
    agent_id="support-agent",
    role="support",
    # `archive_block` must be listed explicitly — archive() is its own scope,
    # distinct from "lesson"/"fact"/"trace". Agents without it get
    # PermissionDenied on archive() calls.
    read_scopes=["fact", "lesson", "rule", "handoff", "feedback", "archive_block"],
    write_scopes=["fact", "trace", "lesson", "handoff", "feedback", "archive_block"],
)
 
client.remember(
    session_id=run_id,
    agent_id="support-agent",
    content="Customer Taylor prefers concise Friday updates and wants billing fixes summarized in one paragraph.",
    intent="fact",
    metadata={"customer": "taylor", "source": "ticket"},
)
 
context = client.get_context(
    session_id=run_id,
    query="Draft the next customer-safe update for Taylor.",
    mode="full",
    max_token_budget=700,
)
 
checkpoint = client.checkpoint(
    session_id=run_id,
    agent_id="support-agent",
    label="pre-response-compaction",
    context_snapshot=context.get("context_block", ""),
    metadata={"stage": "drafting"},
)
 
archived = client.archive(
    session_id=run_id,
    agent_id="support-agent",
    content="Original billing diff and exact remediation note for ticket-42",
    artifact_kind="billing_postmortem",
    labels=["billing", "exact"],
)
 
exact = client.dereference(
    session_id=run_id,
    reference_id=archived["reference_id"],
)
 
reflection = client.reflect(session_id=run_id)
# Read the lesson id off the reflect response directly — it's populated
# server-side from the persistence call. Avoid a round-trip through
# `client.lessons()` which races with the vector store's index propagation
# and can return the run as empty for a few seconds after reflect.
lesson_id = next(
    (l.get("lesson_id") for l in (reflection.get("lessons") or []) if l.get("lesson_id")),
    None,
)
 
if lesson_id:
    client.record_outcome(
        session_id=run_id,
        agent_id="support-agent",
        reference_id=lesson_id,
        outcome="success",
        signal=0.8,
        rationale="The customer update matched the stored preference and resolved the billing question cleanly.",
    )
 
strategies = client.surface_strategies(
    session_id=run_id,
    lesson_types=["success", "failure"],
    max_strategies=3,
)
 
print({
    "checkpoint_id": checkpoint.get("checkpoint_id"),
        "lessons_stored": reflection.get("lessons_stored"),
        "strategies": len(strategies.get("strategies", [])),
        "exact_reference_found": exact.get("found"),
    })

What matters operationally

get_context is the pre-compaction memory surface.
checkpoint preserves the critical context snapshot before the LLM window is compacted.
archive plus dereference is the exact-reference path for artifacts that must come back verbatim later.
reflect extracts reusable lessons and rules from the run.
record_outcome changes later ranking so good lessons appear earlier next time.

ℹ️Note

A freshly reflected lesson does not always land as a fully trusted, immediately-surfaced entry. Reflected lessons enter long-term memory as pending candidates and are promoted to active only once their score crosses the accept threshold (default 0.6); the control stream emits context.lesson_validation_passed / context.lesson_validation_failed alongside context.lesson_promoted. The record_outcome calls above are part of how a pending candidate accrues positive evidence and gets promoted, so a brand-new lesson may not be surfaced or trusted on the very next case until enough outcomes accumulate. Set MUBIT_CONTROL_LESSON_VALIDATION_ENABLED=0 on the control service to restore the legacy behavior where every reflected lesson is active immediately.

Self-contained runnable example

Copy the two files below into an empty directory, fill in MUBIT_API_KEY, and python support_agent.py will run the full loop end-to-end without any LLM provider setup. The "LLM response" is a deterministic stub so you can see every MuBit call succeed before wiring in a real model.

.env

MUBIT_API_KEY=mbt_<instance>_<key_id>_<secret>
MUBIT_ENDPOINT=https://api.mubit.ai

support_agent.py

"""Self-contained support-agent loop against MuBit.
 
Prereqs: pip install mubit-sdk>=0.6.0 python-dotenv
Run:     python support_agent.py
"""
from __future__ import annotations
 
import os
import uuid
from dotenv import load_dotenv
from mubit import Client
 
load_dotenv()
 
RUN_ID = f"support:demo:{uuid.uuid4().hex[:8]}"
AGENT = "support-agent"
 
 
def stub_llm(prompt: str, context: str) -> str:
    """Deterministic stand-in for a real LLM so the example is self-contained."""
    return (
        "Hi Taylor — here is your Friday billing summary in one paragraph: "
        "the duplicate charge was refunded, the recurring invoice was corrected, "
        "and no further action is needed. (context_chars=%d)" % len(context)
    )
 
 
def main() -> None:
    client = Client(run_id=RUN_ID)
 
    # 1. Register the agent and its scopes
    client.register_agent(
        session_id=RUN_ID,
        agent_id=AGENT,
        role="support",
        read_scopes=["fact", "lesson", "rule", "handoff"],
        write_scopes=["fact", "trace", "lesson", "handoff"],
    )
 
    # 2. Store a durable customer fact
    client.remember(
        session_id=RUN_ID,
        agent_id=AGENT,
        content="Customer Taylor prefers concise Friday updates; bills should be "
                "summarized in one paragraph, not a bulleted list.",
        intent="fact",
        metadata={"customer": "taylor", "source": "ticket-42"},
    )
 
    # 3. Assemble context before the response
    ctx = client.get_context(
        session_id=RUN_ID,
        query="Draft the next customer-safe billing update for Taylor.",
        mode="full",
        max_token_budget=700,
    )
    context_block = ctx.get("context_block", "")
 
    # 4. Checkpoint before the generation (compaction safety)
    checkpoint = client.checkpoint(
        session_id=RUN_ID,
        agent_id=AGENT,
        label="pre-response",
        context_snapshot=context_block,
        metadata={"stage": "drafting"},
    )
 
    # 5. Generate the response (stub in place of a real LLM)
    response = stub_llm("Draft the billing update.", context_block)
 
    # 6. Ingest the response as a trace (kept distinct from the fact)
    client.remember(
        session_id=RUN_ID,
        agent_id=AGENT,
        content=f"Agent response: {response}",
        intent="trace",
    )
 
    # 7. Reflect after the attempt. The lesson id is returned inline on the
    # response so we don't need a separate `lessons()` call (which can race
    # with the vector-store index right after a write).
    reflection = client.reflect(session_id=RUN_ID)
    lesson_id = next(
        (l.get("lesson_id") for l in (reflection.get("lessons") or []) if l.get("lesson_id")),
        None,
    )
 
    # 8. Reinforce the lesson that drove the right format
    if lesson_id:
        client.record_outcome(
            session_id=RUN_ID,
            agent_id=AGENT,
            reference_id=lesson_id,
            outcome="success",
            signal=0.8,
            rationale="Response matched the stored preference for concise paragraph format.",
        )
 
    # 9. Surface reusable strategies for the next similar ticket
    strategies = client.surface_strategies(
        session_id=RUN_ID,
        lesson_types=["success", "failure"],
        max_strategies=3,
    )
 
    print({
        "run_id": RUN_ID,
        "checkpoint_id": checkpoint.get("checkpoint_id"),
        "lessons_stored": reflection.get("lessons_stored", 0),
        "strategies_surfaced": len(strategies.get("strategies", [])),
        "response_preview": response[:80],
    })
 
 
if __name__ == "__main__":
    main()

Expected output (values will differ, but the shape should match):

{
  "run_id": "support:demo:a1b2c3d4",
  "checkpoint_id": "ck_...",
  "lessons_stored": 1,
  "strategies_surfaced": 2,
  "response_preview": "Hi Taylor — here is your Friday billing summary..."
}

Once this works, swap stub_llm for a real provider (openai.ChatCompletion.create, anthropic.Anthropic().messages.create, etc.) and feed context_block as the system message. Nothing else in the loop needs to change.

Next steps

Coordinate multiple specialists at Multi-agent shared state.
Debug weak memory quality with Control HTTP reference.