Tutoriel : Ajouter des sessions et de la mémoire à un agent existant

info

Bêta

Cette fonctionnalité est en version bêta.

Si vous avez déjà créé un agent en Python, vous pouvez lui donner un historique de conversation durable et une mémoire à long terme sur Databricks sans le réécrire. Dans ce tutoriel, vous connectez l’agent à un magasin de sessions qui conserve chaque conversation, à un magasin de mémoire qui conserve des faits durables entre les conversations, et à un pipeline de mémoire qui distille les conversations stockées dans la mémoire. Les étapes couvrent les agents créés avec le SDK OpenAI Agents, LangGraph ou tout autre framework Python, où que l’agent s’exécute.

astuce

Si vous n'avez pas encore d'agent, start plutôt par le guide de start rapide d'Agent Bricks. La CLI d'Agent Bricks crée la structure d'un nouvel agent avec des sessions et une mémoire déjà configurées.

Dans cet exemple, un agent d’assistance devrait reprendre la conversation d’un client là où elle en était, et se souvenir des préférences du client lors de conversations ultérieures distinctes.

Pour savoir comment les sessions et la mémoire interagissent, consultez Mémoire des agents et sessions.

Conditions requises

• Un agent existant écrit en Python 3.10 ou version ultérieure. Les étapes spécifiques au framework couvrent le SDK OpenAI Agents et LangGraph. Les étapes pour les autres frameworks appellent directement le SDK AgentKit.
• Authentification Databricks. Le SDK AgentKit utilise l’authentification par default du SDK Databricks pour Python. Pour le développement local, exécutez databricks auth login --host <workspace-url> ou définissez les variables d’environnement DATABRICKS_HOST et DATABRICKS_TOKEN.
• La CLI Databricks, pour vérifier l’état des exécutions du pipeline de mémoire.
• Pour créer un pipeline de mémoire, l’autorisation de query le modèle utilisé par le pipeline, ainsi que les autres accès listés dans les conditions requises du pipeline de mémoire.

Étape 1 : installez le SDK AgentKit et l’interface de ligne de commande (CLI) Agent Bricks

Le package databricks-agentbricks contient les éléments suivants :

• Le SDK AgentKit (databricks_agentkit), le client Python pour les magasins de session et de mémoire.
• La CLI Agent Bricks (agentbricks).
• Adaptateurs pour le SDK OpenAI Agents et LangGraph qui connectent les magasins de sessions et de mémoire aux interfaces de session et d’outils existantes de chaque framework. Les adaptateurs utilisent les packages du framework que votre agent a déjà installés.

Installez la version 0.5.0 ou supérieure dans l’environnement de votre agent :

Bash

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

Étape 2 : Créer un stockage de session et un stockage de mémoire

Votre agent utilise deux magasins, que Databricks gère pour vous :

• Un magasin de sessions contient l'historique des conversations. Chaque session à l'intérieur est la transcription d'une conversation, que l'agent recharge pour poursuivre cette conversation là où elle s'est arrêtée.
• Un stockage de mémoire conserve les faits durables concernant chaque utilisateur, tels que ses préférences et ses décisions passées. L’agent y effectue des recherches pour se souvenir de ce qu’il a appris lors de conversations précédentes et distinctes.

Le magasin constitue la limite de sécurité : tout principal ayant accès à un magasin peut lire et écrire toutes ses entrées. Pour isoler les tenants les uns des autres, créez une paire de magasins distincte pour chaque tenant. Voir Partitionner et sécuriser la mémoire.

Créez chaque magasin une fois et réutilisez-le pour chaque conversation. Votre agent fait référence aux deux magasins par leur nom.

AgentKit SDK

Python

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

Agent Bricks CLI

Bash

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

Définissez DATABRICKS_HOST sur le Hostname de votre Workspace, puis exécutez les commandes suivantes :

Bash

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_name=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" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}" -H "Content-Type: application/json" \
  -d '{"display_name": "support-agent-memory", "description": "Support agent long-term memory"}'

Les noms de magasins doivent être uniques dans le workspace. Un nom de magasin de mémoire doit comporter de 3 à 56 caractères, start par une lettre minuscule, se terminer par une lettre ou un chiffre, et ne contenir que des lettres minuscules, des chiffres et des tirets. Un nom de magasin de session doit comporter de 3 à 63 caractères.

Étape 3 : choisir les ID d'acteur et de session

Chaque appel que votre agent effectue vers les magasins porte deux identifiants :

• actor_id identifie à qui appartiennent les conversations et les souvenirs. Les deux magasins l'utilisent, et un pipeline de mémoire écrit les souvenirs de chaque acteur sous la même actor_id que les sessions de cet acteur, de sorte que l'agent ne rappelle que les souvenirs distillés lorsque les deux correspondent.
• session_id identifie une conversation afin qu'une demande de suivi la reprenne.

Pour ce tutoriel, utilisez test-user comme actor_id et conversation-1 comme session_id. Les fonctions handle_message des étapes suivantes les utilisent comme user_id et conversation_id.

En production, définissez actor_id dans le code de l’application à partir d’un contexte de requête fiable, tel que l’identité vérifiée de l’utilisateur connecté, et ne laissez jamais le modèle ou le message d’un utilisateur le choisir. Pour partager la mémoire au sein d’une équipe ou isoler les tenants, consultez Partitionner et sécuriser la mémoire. Pour session_id, utilisez l’ID de conversation ou de fil de discussion existant de votre application, ou générez un UUID au start d’une conversation.

Étape 4 : Persister les conversations dans le magasin de sessions

L’historique des conversations est l’enregistrement ordonné d’une conversation : les messages de l’utilisateur, les réponses de l’agent, ainsi que les appels d’outils et les résultats. Un stockage de session enregistre cet historique pour chaque conversation afin que l’agent puisse le recharger lors de la demande suivante, même après un redémarrage. Remplacez l’historique des conversations en cours de votre agent par le stockage de session que vous avez créé à l’étape 2.

Les adaptateurs de framework se connectent à l’interface que chaque framework utilise déjà pour l’état de la session, de sorte que le modèle, les outils et les invites de votre agent restent les mêmes.

OpenAI Agents SDK

Le SDK OpenAI Agents lit et écrit l’historique des conversations via un objet Session. session_store() renvoie Session sauvegardé par le magasin de sessions géré. Transmettez-le à Runner.run:

Python

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

Sur le premier message d’une conversation, l’adaptateur crée la session. À chaque exécution, le SDK charge l’historique de la conversation à partir de la session et ajoute le nouveau message de l’utilisateur, la sortie du modèle et les éléments de l’outil.

LangGraph

LangGraph conserve l'état du graphe via un point de contrôle associé à un ID de thread. checkpointer() renvoie un point de contrôle pris en charge par le magasin de sessions géré, et thread_config() mappe une conversation et son utilisateur à un thread :

Python

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

Si votre agent utilise déjà un point de contrôle, remplacez-le par checkpointer(). Si vous créez directement un StateGraph, transmettez le point de contrôle à compile(). Le point de contrôle stocke l’état complet du graphe, y compris les interruptions nécessitant une intervention humaine, afin qu’une exécution en pause puisse reprendre après un redémarrage.

remarque

Le point de contrôle stocke les points de contrôle de graphe sérialisés, qu’un pipeline de mémoire ne lit pas. Pour utiliser un pipeline de mémoire avec un agent LangGraph, enregistrez également une transcription.

Other frameworks

Pour tout autre framework ou pour des appels de modèle directs, lisez et ajoutez l'historique de la conversation avec le SDK AgentKit. Stockez chaque message sous la forme {"type": "message", "role": "user" | "assistant", "content": "<text>"}. Un pipeline de mémoire distille les messages de l’utilisateur et de l’assistant sous cette forme, ainsi que sous la forme de message OpenAI Responses avec des parties de contenu textuel. Il ignore tous les autres éléments, tels que les appels d’outils et les résultats des outils.

Python

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

La recherche d'une session par session_id n'est pas limitée à un acteur. Vérifiez donc que la session appartient à l'utilisateur demandeur avant de la reprendre.

Le passage au stockage des sessions ne transfère pas les conversations que votre agent a stockées ailleurs. Pour transférer une conversation existante, ajoutez ses éléments à une nouvelle session.

Étape 5 : Donner de la mémoire à l'agent

L'historique de session couvre une conversation. Pour utiliser ce que l'agent a appris lors des conversations précédentes, connectez l'agent au magasin de mémoire. Il existe deux modèles que vous pouvez combiner :

• Outils de mémoire : Le modèle décide quand rappeler la mémoire et quand enregistrer un nouveau fait.
• Rappel avant chaque exécution : votre application recherche dans la mémoire le message entrant et ajoute les résultats au contexte de l’agent, afin que l’agent voie la mémoire pertinente sans appeler d’outil.

Si un pipeline de mémoire écrit votre mémoire (voir étape 6), l’agent n’a besoin que de se souvenir de la mémoire. Vous pouvez ignorer les outils qui économisent la mémoire.

OpenAI Agents SDK

L’exemple suivant utilise les deux modèles dans une copie de votre agent créée pour chaque requête :

• memory_tools() renvoie un outil recall qui effectue une recherche dans la mémoire de l'utilisateur et un outil remember qui enregistre un fait. Les outils sont liés à un acteur, le modèle ne peut donc pas lire ou écrire dans la mémoire d'un autre utilisateur.
• recalled_facts() recherche dans la mémoire le message entrant avant l’exécution et ajoute les résultats aux instructions de l’agent.

Python

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

Les faits rappelés vont dans les instructions, pas dans l’entrée, afin que la session ne les stocke pas. Pour n’utiliser qu’un seul modèle, supprimez l’autre : supprimez la ligne memory_tools() ou supprimez l’appel recalled_facts(). Si vous conservez les outils, mettez à jour les instructions de votre agent pour lui indiquer quand les utiliser, par exemple : « Utilisez le rappel pour consulter les préférences de l’utilisateur avant de répondre. Utilisez remember pour enregistrer les faits durables partagés par l’utilisateur. »

LangGraph

L'exemple suivant utilise les deux modèles et crée l'agent pour chaque demande. Le point de contrôle est créé une seule fois et partagé :

• memory_tools() renvoie un outil recall qui effectue une recherche dans la mémoire de l'utilisateur et un outil remember qui enregistre un fait. Les outils sont liés à un acteur, le modèle ne peut donc pas lire ou écrire dans la mémoire d'un autre utilisateur.
• recalled_facts() recherche dans la mémoire le message entrant avant l’exécution et ajoute les résultats à l’invite système.

Python

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 recall and remember.
    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 ajoute le système prompt à chaque appel de modèle sans l'enregistrer dans l'état du graphe ; le checkpointer ne stocke donc pas les faits rappelés. Pour utiliser un seul modèle, abandonnez l'autre : supprimez la ligne memory_tools() ou supprimez l'appel recalled_facts().

Other frameworks

Recherchez dans la mémoire le message entrant avant chaque exécution et transmettez les résultats à l'agent en tant que contexte. La recherche classe les entrées par pertinence en texte intégral, de sorte qu'une query construite à partir des propres mots de l'utilisateur fonctionne bien.

Python

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

Dans handle_message à partir de l'étape 4, ajoutez le contexte rappelé à l'entrée de l'agent. Ne l'ajoutez pas à la session :

Python

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

Pour permettre à l'agent d'enregistrer lui-même la mémoire, encapsulez les opérations search et add du magasin en tant qu'outils pour votre framework. Voir Donner à votre agent des outils de mémoire.

Étape 6 : créer un pipeline de mémoire

Un pipeline de mémoire lit les conversations dans votre stockage de session et extrait les faits durables, les préférences et les décisions de chaque acteur dans le stockage de mémoire. Il fusionne également les entrées qui se chevauchent au fil du temps, de sorte que la mémoire de votre agent reste consolidée sans code d’agent supplémentaire.

1. Créer un pipeline qui relie les deux magasins :

   Bash

   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 exécute automatiquement le pipeline environ toutes les 24 heures. Sans lui, le pipeline utilise le Trigger default, manual, et ne s’exécute que lorsque vous start une exécution. Voir Exécuter un pipeline selon un calendrier.
   • --model définit le modèle qui distille les mémoires. Vous pouvez utiliser n’importe quel modèle pour lequel vous disposez des autorisations de query. Si vous omettez --model, le pipeline utilise un modèle default choisi par Databricks.
   • --instructions est optionnel et dirige ce que le pipeline garde. Consultez Diriger la distillation avec des instructions.

   La commande retourne le nom de la ressource du pipeline, sous la forme memory-pipelines/<pipeline-id>. Pour chaque option, exécutez agentbricks memory pipeline create --help ou consultez la référence de la commande CLI d’Agent Bricks.

2. Le planificateur choisit le moment où la première exécution programmée start. Pour distiller les conversations déjà dans le magasin de sessions, start une exécution maintenant :

   Bash

   agentbricks memory pipeline run <pipeline-id>

   La commande renvoie le nom de la ressource de l'exécution, sous la forme memory-pipelines/<pipeline-id>/runs/<run-id>.

3. Vérifiez le résultat de l'exécution avec l'API REST. Définissez DATABRICKS_HOST sur le Hostname de votre Workspace, puis exécutez la commande suivante :

   Bash

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

   Lorsque le state de l'exécution est SUCCEEDED, les mémoires distillées se trouvent dans le magasin de mémoire.

Les exécutions de pipeline utilisent l’identité qui a créé le pipeline, et non l’identité de votre agent. Un pipeline peut réécrire, Merge, ou supprimer n’importe quelle entrée sous un actor_id qu’il traite, y compris les entrées que votre agent a enregistrées avec des outils de mémoire. Pour l’API REST, les états d’exécution et les limitations, consultez Pipelines de mémoire d’agent.

Enregistrer une transcription pour les agents LangGraph

Un pipeline de mémoire extrait uniquement les messages de l’utilisateur et de l’assistant. Le checkpointer LangGraph stocke des points de contrôle sérialisés. Par conséquent, un pipeline qui lit le stockage de session du checkpointer ne trouve aucun message à extraire. Pour utiliser un pipeline avec un agent LangGraph, ajoutez également les messages de chaque tour à un stockage de session distinct et créez le pipeline sur ce stockage :

Python

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

Appelez start_conversation au début d’une conversation, avant son premier ainvoke, et appelez record_turn après chaque ainvoke. Transmettez --session-store support-agent-transcripts lors de la création du pipeline. Sinon, ignorez le pipeline et laissez l’agent enregistrer la mémoire à l’aide des outils de mémoire.

Étape 7 : accordez l’accès à votre agent déployé

Lorsque vous exécutez l'agent localement, il utilise vos propres identifiants. Un agent déployé s'exécute généralement en tant que service principal, qui doit avoir accès à chaque magasin qu'il lit ou écrit. Accédez avec le SDK AgentKit, en passant l'ID de l'application (client) du service principal :

Python

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 accorde un accès en écriture par default. Pour accorder un accès en lecture seule, passez permission="READ". Si vous avez créé un stockage de transcriptions pour un agent LangGraph, accordez-lui également l'accès. Le pipeline de mémoire n’a pas besoin d’autorisations pour le service principal de votre agent, car les exécutions du pipeline utilisent l’identité qui a créé le pipeline.

Étape 8 : tester l'agent de bout en bout

1. Appelez handle_message avec user_id="test-user" et conversation_id="conversation-1", et partagez une préférence durable, par exemple « Je préfère l'e-mail au téléphone et je suis dans le fuseau horaire du Pacifique. » Si vous avez suivi les étapes pour d'autres frameworks, appelez d'abord start_conversation("test-user", "conversation-1"). Envoyez un message de suivi dans la même conversation et confirmez que l'agent dispose du contexte précédent.

2. Vérifiez que le magasin de sessions contient la conversation :

   Bash

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

3. start une exécution de pipeline de mémoire et attendez que son state soit SUCCEEDED, comme à l'étape 6.

4. Confirmez que le magasin de mémoire détient la préférence pour l'utilisateur de test :

   Bash

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

5. start une nouvelle conversation pour test-user avec conversation_id="conversation-2", et demandez "Comment dois-je vous contacter ?" L’agent répond à partir de sa mémoire.

Dépannage

• PERMISSION_DENIED (HTTP 403) de l'agent déployé. Le Service Principal de l'agent ne peut pas accéder à un magasin. Accordez-lui l’accès, comme à l’ étape 7.
• Une exécution de pipeline réussit mais n'écrit aucune entrée. Confirmez que le magasin de sessions contient les messages de l'utilisateur et de l'assistant dans une forme que le pipeline peut lire. Un pipeline ignore les appels d’outils, les résultats d’outils et les points de contrôle LangGraph. Une exécution ne traite que les sessions d'un acteur depuis la dernière exécution réussie de cet acteur.
• L’agent ne se souvient pas des souvenirs issus de la distillation. Vérifiez que l’agent effectue une recherche dans la mémoire avec le même actor_id que celui utilisé par ses sessions.
• Le démarrage d’une exécution échoue avec ABORTED. Une autre exécution du pipeline est en cours. Attendez qu’elle se termine, puis start une nouvelle exécution.

Nettoyer

Pour supprimer les ressources créées dans ce tutoriel, exécutez ce qui suit. La suppression d’un magasin supprime tout ce qu’il contient.

Bash

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

Étapes suivantes

• Sessions d'agent
• Mémoire de l’agent
• Pipelines de mémoire d’agent
• CLI Agent Bricks