Tutorial: Aggiungi sessioni e memoria a un agente esistente

Se hai già creato un agente in Python, puoi dargli una cronologia delle conversazioni duratura e memoria a lungo termine su Azure Databricks senza riscriverlo. In questo tutorial, colleghi l'agente a un archivio sessioni che persiste ogni conversazione, a uno store di memoria che contiene fatti duraturi tra le conversazioni e a una pipeline di memoria che distilla le conversazioni memorizzate in memoria. I passaggi coprono gli agenti costruiti con l'SDK OpenAI Agents, LangGraph o qualsiasi altro framework Python, ovunque l'agente venga eseguito.

In questo esempio, un operatore di supporto dovrebbe riprendere la conversazione del cliente dal punto in cui si era interrotta e ricordare le preferenze del cliente nelle conversazioni successive e separate.

Per come le sessioni e la memoria si relazionano, vedi Memoria e sessioni dell'agente.

Requirements

• Un agente esistente scritto in Python 3.10 o superiore. I passaggi specifici del framework coprono l'SDK degli Agenti OpenAI e LangGraph. I passaggi per altri framework richiamano direttamente l'SDK AgentKit.
• Autenticazione di Azure Databricks. L'SDK AgentKit utilizza l'autenticazione predefinita dell'SDK Databricks per Python. Per lo sviluppo locale, esegui databricks auth login --host <workspace-url>, oppure imposta le variabili d'ambiente DATABRICKS_HOST e DATABRICKS_TOKEN.
• La CLI Databricks, per controllare lo stato delle esecuzioni della pipeline di memoria.
• Per creare una pipeline di memoria, sono necessari il permesso di interrogare il modello utilizzato dalla pipeline e gli altri accessi elencati nei requisiti della pipeline di memoria.

Passo 1: Installa l'SDK AgentKit e la CLI Agent Bricks

Il databricks-agentbricks pacchetto contiene quanto segue:

• AgentKit SDK (databricks_agentkit), client Python per gli archivi di sessione e memoria.
• La CLI di Agent Bricks (agentbricks).
• Adattatori per l'SDK degli Agenti OpenAI e LangGraph che collegano i storage di sessione e memoria alle interfacce di sessione e strumenti esistenti di ciascun framework. Gli adattatori utilizzano i pacchetti framework che il tuo agente ha già installati.

Installa la versione 0.6.0 o superiore nell'ambiente del tuo agente:

pip install --upgrade "databricks-agentbricks>=0.6.0"

Passo 2: Crea uno store sessione e uno store di memoria

Il tuo agente utilizza due store, che Azure Databricks gestisce per te:

• Un archivio di sessione contiene la cronologia delle conversazioni. Ogni sessione è la trascrizione di una conversazione, che l'agente ricarica per continuare la conversazione da dove aveva interrotto.
• Una memoria contiene fatti durevoli su ogni utente, come preferenze e decisioni passate. L'agente lo esamina per ricordare cosa ha appreso in conversazioni precedenti e separate.

Lo store è il confine di sicurezza: qualsiasi principal con accesso a uno store può leggere e scrivere tutte le sue voci. Per isolare i tenant tra loro, crea una coppia separata di store per ciascun tenant. Vedi Partizionare e proteggere la memoria.

Crea ogni store una volta e riutilizzalo per ogni conversazione. Il tuo agente si riferisce a entrambi i negozi per nome.

AgentKit SDK

from databricks.sdk import WorkspaceClient
from databricks_agentkit import AgentKitClient

client = AgentKitClient(WorkspaceClient())
client.session_stores.create("support-agent-sessions", description="Support agent conversations")
client.memory_stores.create("support-agent-memory", description="Support agent long-term memory")

Mattoni agente

agentbricks sessions stores create --name support-agent-sessions --description "Support agent conversations"
agentbricks memory stores create --display-name support-agent-memory --description "Support agent long-term memory"

REST API

Imposta DATABRICKS_HOST sul nome host del tuo workspace, quindi esegui quanto segue:

export DATABRICKS_TOKEN=$(databricks auth token --profile <profile> | jq -r .access_token)

curl -X POST "https://${DATABRICKS_HOST}/api/2.0/agents/session-stores?session_store_id=support-agent-sessions" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}" -H "Content-Type: application/json" \
  -d '{"description": "Support agent conversations"}'

curl -X POST "https://${DATABRICKS_HOST}/api/2.0/agents/memory-stores?managed_memory_store_id=support-agent-memory" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}" -H "Content-Type: application/json" \
  -d '{"description": "Support agent long-term memory"}'

Gli ID dello store devono essere unici nello spazio di lavoro. Un ID dell'archivio di memoria deve essere di 3 a 56 caratteri, iniziare con una lettera minuscola, terminare con una lettera o un numero e contenere solo lettere minuscole, numeri e trattini. Un ID di session store deve essere di 3 a 55 caratteri.

Passo 3: Scegli gli ID degli attori e delle sessioni

Ogni chiamata che il tuo agente effettua ai negozi porta due identificatori:

• actor_id determina a chi appartengono le conversazioni e i ricordi. Entrambi gli store lo usano, e una pipeline di memoria scrive i ricordi di ogni attore nello stesso actor_id delle sessioni di quell'attore, così l'agente richiama i ricordi distillati solo quando i due corrispondono.
• session_id identifica una conversazione, così che una richiesta di follow-up la riprenda.

Per questo tutorial, usa test-user come actor_id e conversation-1 come session_id. Le handle_message funzioni nei passaggi seguenti le considerano come user_id e conversation_id.

In produzione, imposta actor_id nel codice dell'applicazione dal contesto delle richieste affidabili, come l'identità verificata dell'utente che ha effettuato l'accesso, e non lasciare mai che il modello o il messaggio dell'utente lo scelgano. Per condividere la memoria tra un team o isolare i tenant, vedi Partizionare e proteggere la memoria. Per session_id, usa l'ID della conversazione o thread già presente nella tua app, oppure genera un UUID quando una conversazione inizia.

Passo 4: Persisti le conversazioni nello store delle sessioni

La cronologia della conversazione è il registro ordinato di una conversazione: i messaggi dell'utente, le risposte dell'agente e qualsiasi chiamata e risultato degli strumenti. Un archivio di sessione salva quella cronologia per ogni conversazione, così l'agente può ricaricarla alla prossima richiesta, anche dopo un riavvio. Sostituisci la cronologia delle conversazioni in corso del tuo agente con lo store sessioni che hai creato nel Passo 2.

Gli adattatori del framework si collegano all'interfaccia che ciascun framework già usa per lo stato della sessione, così il modello, gli strumenti e i prompt del tuo agente rimangono gli stessi.

OpenAI Agents SDK

L'SDK OpenAI Agents legge e scrive la cronologia delle conversazioni tramite un Session oggetto. session_store() restituisce un Session supportato dal Managed Session Store. Passalo a Runner.run:

from agents import Agent, Runner
from databricks_agentkit.openai import session_store

SESSION_STORE = "support-agent-sessions"

# Your existing agent, unchanged.
agent = Agent(name="Support agent", instructions="You help customers with their Databricks workspaces.")

async def handle_message(user_id: str, conversation_id: str, message: str) -> str:
    session = session_store(conversation_id, actor=user_id, store=SESSION_STORE)
    result = await Runner.run(agent, message, session=session)
    return result.final_output

Al primo messaggio di una conversazione, l'adattatore crea la sessione. Ad ogni esecuzione, l'SDK carica la cronologia della conversazione dalla sessione e aggiunge il nuovo messaggio utente, l'output del modello e gli elementi dello strumento.

LangGraph

LangGraph persiste lo stato del grafo tramite un checkpointer associato a un ID del thread. checkpointer() restituisce un checkpointer supportato dallo store di sessione gestito e thread_config() mappa una conversazione e il suo utente su un thread:

from langchain.agents import create_agent
from databricks_agentkit.langgraph import checkpointer, thread_config

SESSION_STORE = "support-agent-sessions"

# Your existing model and tools, unchanged, with the managed checkpointer added.
agent = create_agent(model=model, tools=tools, checkpointer=checkpointer(store=SESSION_STORE))

async def handle_message(user_id: str, conversation_id: str, message: str) -> str:
    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": message}]},
        config=thread_config(conversation_id, actor=user_id),
    )
    return result["messages"][-1].content

Se il tuo agente usa già un checkpointer, sostituiscilo con checkpointer(). Se costruisci un StateGraph direttamente, passa il checkpointer a compile(). Il checkpointer memorizza lo stato completo del grafo, incluse le interruzioni con intervento umano, così un'esecuzione in pausa può riprendere dopo un riavvio.

Altri framework

Per qualsiasi altro framework, o per chiamate dirette al modello, leggi e aggiungi la cronologia delle conversazioni con l'SDK AgentKit. Memorizza ogni messaggio come {"type": "message", "role": "user" | "assistant", "content": "<text>"}. Una pipeline di memoria distilla i messaggi dell'utente e dell'assistente in questa forma, e nella forma di messaggi OpenAI Responses con parti di contenuto testuale. Salta tutti gli altri elementi, come chiamate e risultati degli strumenti.

from databricks.sdk import WorkspaceClient
from databricks_agentkit import AgentKitClient

client = AgentKitClient(WorkspaceClient())
session_store = client.session_stores.get("support-agent-sessions")

def start_conversation(user_id: str, conversation_id: str) -> None:
    session_store.add(actor_id=user_id, session_id=conversation_id)

def handle_message(user_id: str, conversation_id: str, message: str) -> str:
    session = session_store.get(conversation_id)
    if session.actor_id != user_id:
        raise PermissionError("This conversation belongs to a different user.")

    # Rebuild context from the stored history, oldest first.
    history = [item.data for item in session.list_items(order_by="create_time asc")]
    user_item = {"type": "message", "role": "user", "content": message}
    reply = run_agent([*history, user_item])  # Your framework or model call.

    session.append_items([user_item, {"type": "message", "role": "assistant", "content": reply}])
    return reply

Cercare una sessione per session_id non è limitato a un attore, quindi verifica che la sessione appartenga all'utente richiedente prima di riprenderla.

Passare al session store non sposta le conversazioni che il tuo agente aveva memorizzato altrove. Per portare avanti una conversazione esistente, aggiungi i suoi elementi a una nuova sessione.

Passo 5: Dai memoria all'agente

La cronologia della sessione copre una conversazione. Per utilizzare ciò che l'agente ha appreso nelle conversazioni precedenti, collega l'agente al memory store. Ci sono due schemi, e puoi combinarli:

• Strumenti di memoria: Il modello decide quando richiamare la memoria e quando salvare un nuovo fatto.
• Promemoria prima di ogni esecuzione: La tua applicazione cerca in memoria il messaggio in arrivo e aggiunge i risultati al contesto dell'agente, così l'agente vede la memoria rilevante senza chiamare uno strumento.

Se una pipeline di memoria scrive la tua memoria (vedi Passaggio 6), l'agente deve solo richiamare la memoria. Puoi ignorare gli strumenti che salvano memoria.

OpenAI Agents SDK

Il seguente esempio utilizza entrambi i pattern in una copia per richiesta del tuo agente:

• memory_tools() restituisce quattro strumenti: recall cerca nella memoria dell'utente, remember salva un fatto sotto un argomento, forget rimuove un fatto e list_memories elenca ciò che è memorizzato. Gli strumenti sono legati a un solo attore, quindi il modello non può leggere o scrivere la memoria di un altro utente.
• recalled_facts() cerca nella memoria il messaggio in arrivo prima dell'esecuzione e aggiunge i risultati alle istruzioni dell'agente.

from databricks.sdk import WorkspaceClient
from databricks_agentkit import AgentKitClient
from databricks_agentkit.openai import memory_tools, session_store

MEMORY_STORE = "support-agent-memory"
memory_store = AgentKitClient(WorkspaceClient()).memory_stores.get(MEMORY_STORE)

def recalled_facts(user_id: str, message: str) -> str:
    results = memory_store.search(actor_id=user_id, query=message, limit=5)
    return "\n".join(f"- {r.memory.content}" for r in results)

async def handle_message(user_id: str, conversation_id: str, message: str) -> str:
    instructions = agent.instructions  # Your agent's existing instructions, as a string.
    if facts := recalled_facts(user_id, message):  # Recall before the run.
        instructions += f"\n\nWhat you know about this user:\n{facts}"
    user_agent = agent.clone(
        instructions=instructions,
        tools=[*agent.tools, *memory_tools(user_id, store=MEMORY_STORE)],  # Memory tools.
    )
    session = session_store(conversation_id, actor=user_id, store=SESSION_STORE)
    result = await Runner.run(user_agent, message, session=session)
    return result.final_output

I fatti richiamati vanno nelle istruzioni, non nell'input, quindi la sessione non li memorizza. Per usare un solo pattern, elimina l'altro: rimuovi la memory_tools() linea o rimuovi la recalled_facts() chiamata. Se tieni gli strumenti, aggiorna le istruzioni del tuo agente per indicare quando usarli, ad esempio: "Usa recall per consultare le preferenze dell'utente prima di rispondere. Usa Remember per conservare dati duraturi che l'utente condivide. Usa il comando "forget" quando l'utente ti chiede di dimenticare qualcosa.

LangGraph

Il seguente esempio utilizza entrambi i pattern e costruisce l'agente per ogni richiesta. Il checkpointer viene creato una volta e condiviso:

• memory_tools() restituisce quattro strumenti: recall cerca nella memoria dell'utente, remember salva un fatto sotto un argomento, forget rimuove un fatto e list_memories elenca ciò che è memorizzato. Gli strumenti sono legati a un solo attore, quindi il modello non può leggere o scrivere la memoria di un altro utente.
• recalled_facts() cerca in memoria il messaggio in arrivo prima dell'esecuzione e aggiunge i risultati al prompt di sistema.

from databricks.sdk import WorkspaceClient
from databricks_agentkit import AgentKitClient
from langchain.agents import create_agent
from databricks_agentkit.langgraph import checkpointer, memory_tools, thread_config

MEMORY_STORE = "support-agent-memory"
memory_store = AgentKitClient(WorkspaceClient()).memory_stores.get(MEMORY_STORE)

def recalled_facts(user_id: str, message: str) -> str:
    results = memory_store.search(actor_id=user_id, query=message, limit=5)
    return "\n".join(f"- {r.memory.content}" for r in results)

async def handle_message(user_id: str, conversation_id: str, message: str) -> str:
    system_prompt = SYSTEM_PROMPT  # Tell the agent when to use the memory tools.
    if facts := recalled_facts(user_id, message):  # Recall before the run.
        system_prompt += f"\n\nWhat you know about this user:\n{facts}"
    agent = create_agent(
        model=model,
        tools=[*tools, *memory_tools(user_id, store=MEMORY_STORE)],  # Memory tools.
        system_prompt=system_prompt,
        checkpointer=checkpointer(store=SESSION_STORE),
    )
    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": message}]},
        config=thread_config(conversation_id, actor=user_id),
    )
    return result["messages"][-1].content

LangGraph aggiunge il prompt di sistema a ogni chiamata al modello senza salvarlo nello stato del grafo, quindi il checkpointer non memorizza i fatti richiamati. Per usare un solo pattern, elimina l'altro: rimuovi la memory_tools() linea o rimuovi la recalled_facts() chiamata.

Altri framework

Cerca in memoria il messaggio in arrivo prima di ogni esecuzione e passa i risultati all'agente come contesto. La ricerca corrisponde sia per parole chiave che per significato, quindi il messaggio dell'utente funziona bene come query.

memory_store = client.memory_stores.get("support-agent-memory")

def recall(user_id: str, message: str) -> list[dict]:
    results = memory_store.search(actor_id=user_id, query=message, limit=5)
    if not results:
        return []
    facts = "\n".join(f"- {r.memory.content}" for r in results)
    return [{"role": "system", "content": f"What you know about this user:\n{facts}"}]

In handle_message del Passo 4, aggiungi il contesto richiamato all'input dell'agente. Non aggiungerlo alla sessione:

reply = run_agent([*recall(user_id, message), *history, user_item])

Per permettere all'agente di salvare autonomamente la memoria, incapsula le operazioni search e add dello store come strumenti per il tuo framework. Consulta Dai al tuo agente strumenti di memoria.

Passo 6: Crea una pipeline di memoria

Una pipeline di memoria legge le conversazioni nel tuo archivio sessione e distilla fatti, preferenze e decisioni durature per ogni attore nello store di memoria. Inoltre, unisce le voci sovrapposte nel tempo, così la memoria del tuo agente rimane consolidata senza codice aggiuntivo dell'agente.

1. Crea una pipeline che colleghi i due store:

   agentbricks memory pipeline create \
     --session-store support-agent-sessions \
     --memory-store support-agent-memory \
     --trigger scheduled \
     --model system.ai.claude-sonnet-5 \
     --instructions "Focus on durable customer preferences and account details. Ignore one-off troubleshooting steps."
   

   • --trigger scheduled La pipeline viene eseguita automaticamente circa ogni 24 ore. Senza di essa, la pipeline utilizza il trigger predefinito, manual, e funziona solo quando inizi una run. Vedi Eseguire una pipeline in base a una pianificazione.
   • --model imposta il modello che distilla i ricordi. Puoi usare qualsiasi modello che hai l'autorizzazione a interrogare. Se ometti --model, la pipeline utilizza un modello predefinito che Azure Databricks sceglie.
   • --instructions è opzionale e determina cosa la pipeline conserva. Vedi la distillazione di Steer con istruzioni.

   Il comando restituisce il nome della risorsa della pipeline, nella forma memory-pipelines/<pipeline-id>. Per ogni opzione, esegui agentbricks memory pipeline create --help o consulta il riferimento del comando CLI di Agent Bricks.

2. L’utilità di pianificazione sceglie quando inizia la prima esecuzione pianificata. Per condensare le conversazioni già presenti nell'archivio sessioni, avvia un'esecuzione ora:

   agentbricks memory pipeline run <pipeline-id>
   

   Il comando restituisce il nome della risorsa della run, nella forma memory-pipelines/<pipeline-id>/runs/<run-id>.

3. Controlla il risultato della run con l'API REST. Imposta DATABRICKS_HOST sul nome host del tuo workspace, quindi esegui quanto segue:

   export DATABRICKS_TOKEN=$(databricks auth token --profile <profile> | jq -r .access_token)
   curl "https://${DATABRICKS_HOST}/api/2.0/agents/memory-pipelines/<pipeline-id>/runs/<run-id>" \
     -H "Authorization: Bearer ${DATABRICKS_TOKEN}"
   

   Quando l'esecuzione state è SUCCEEDED, i ricordi distillati sono nell'archivio di memoria.

Le esecuzioni di pipeline usano l'identità che ha creato la pipeline, non quella del tuo agente. Una pipeline può riscrivere, unire o cancellare qualsiasi voce sotto un actor_id che elabora, inclusi i voci che il tuo agente ha salvato con strumenti di memoria. Per l'API REST, gli stati di esecuzione e le limitazioni, vedi Pipeline di memoria agente.

Registra una trascrizione per gli agenti LangGraph

Una pipeline di memoria distilla solo i messaggi dell'utente e dell'assistente. Il checkpointer LangGraph memorizza checkpoint serializzati, quindi una pipeline che legge lo store sessione del checkpointer non trova messaggi da distillare. Per utilizzare una pipeline con un agente LangGraph, aggiungi anche i messaggi di ogni turno a uno store sessione separato e crea la pipeline su quel store:

transcripts = client.session_stores.create("support-agent-transcripts")  # Create once.

def start_conversation(user_id: str, conversation_id: str) -> None:
    transcripts.add(actor_id=user_id, session_id=conversation_id)

def record_turn(conversation_id: str, message: str, reply: str) -> None:
    transcripts.get(conversation_id).append_items(
        [
            {"type": "message", "role": "user", "content": message},
            {"type": "message", "role": "assistant", "content": reply},
        ]
    )

Chiama start_conversation quando una conversazione inizia, prima della sua prima ainvoke, e chiama record_turn dopo ogni ainvoke. Passa a --session-store support-agent-transcripts quando crei la pipeline. In alternativa, saltare la pipeline e lasciare che l'agente salvi memoria con strumenti di memoria.

Passo 7: Concedi l'accesso al tuo agente distribuito

Quando esegui l'agente localmente, usa le tue credenziali. Un agente distribuito di solito funziona come entità servizio, che necessita di accesso a ogni store che legge o scrive. Concedere l'accesso con l'SDK AgentKit, passando l'ID dell'applicazione (client) del principale del servizio:

from databricks.sdk import WorkspaceClient
from databricks_agentkit import AgentKitClient

w = WorkspaceClient()
client = AgentKitClient(w)

# For an agent deployed as a Databricks app, look up the app's service principal.
sp_client_id = w.apps.get("support-agent").service_principal_client_id

client.session_stores.get("support-agent-sessions").grant_permission(sp_client_id)
client.memory_stores.get("support-agent-memory").grant_permission(sp_client_id)

grant_permission Garantisce l'accesso in scrittura per impostazione predefinita. Per concedere l'accesso di sola lettura, specifica permission="READ". Se hai creato un archivio di trascrizioni per un agente LangGraph, concedi anche l'accesso all'archivio. La pipeline di memoria non ha bisogno di una concessione per l'entità servizio del tuo agente, perché le esecuzioni della pipeline utilizzano l'identità che ha creato la pipeline.

Passo 8: Testa l'agente end-to-end

1. Chiama handle_message con user_id="test-user" e conversation_id="conversation-1", e condividi una preferenza duratura, come "Preferisco l'email al telefono e sono nel fuso orario del Pacifico." Se hai seguito i passaggi per altri framework, chiama start_conversation("test-user", "conversation-1") prima. Invia un messaggio di follow-up nella stessa conversazione e conferma che l'agente abbia il contesto precedente.

2. Conferma che lo store della sessione contenga la conversazione:

   agentbricks sessions items list --store support-agent-sessions --session-id conversation-1
   

3. Avvia un'esecuzione della pipeline di memoria e aspetta che il suo state sia SUCCEEDED, come nel Passo 6.

4. Conferma che lo store di memoria contenga la preferenza per l'utente di test:

   agentbricks memory entries search --store support-agent-memory --actor-id test-user --query "contact preferences"
   

5. Inizia una nuova conversazione per test-user con conversation_id="conversation-2", e chiedi "Come dovresti contattarmi?" L'agente risponde a memoria.

Troubleshooting

• PERMISSION_DENIED (HTTP 403) dall'agente distribuito. Il service principal dell'agente non può accedere a un archivio. Concedigli l'accesso, come nel Passo 7.
• Una pipeline ha successo ma non scrive voci. Conferma che lo store sessioni conservi i messaggi utente e assistente in una forma che la pipeline legge. Una pipeline salta chiamate di strumenti, risultati di strumenti e checkpoint LangGraph. Una run elabora solo i turni aggiunti dal checkpoint dell'attore, e i turni degli ultimi 60 secondi attendono la successiva esecuzione.
• L'agent non ricorda i ricordi distillati. Conferma che l'agente cerca la memoria con la stessa actor_id che usano le sue sessioni.
• L'avvio di una run non riesce con ABORTED. Un'altra esecuzione della pipeline è in corso. Aspetta che finisca, poi avvia una nuova esecuzione.

Pulizia

Per eliminare le risorse create da questo tutorial, esegui quanto segue. Eliminare un archivio cancella tutto ciò che contiene.

agentbricks memory pipeline delete <pipeline-id>
agentbricks sessions stores delete support-agent-sessions
agentbricks memory stores delete support-agent-memory

Passaggi successivi

• Sessioni agente
• Memoria agente
• Pipeline di memoria degli agenti
• CLI di Agent Bricks