Utiliser la mémoire Foundry avec LangChain et LangGraph

Utilisez langchain-azure-ai et foundry Memory pour ajouter de la mémoire à long terme à vos applications. Dans cet article, vous créez une chaîne à mémoire, stockez les préférences utilisateur, rappelez-les dans une nouvelle session et exécutez des requêtes de mémoire directes.

Ce modèle fonctionne à la fois pour les applications LangChain et LangGraph. L'idée principale est de maintenir l'historique des conversations à court terme dans votre environnement d'exécution et d'utiliser Foundry Memory comme espace de stockage à long terme pour le contexte utilisateur.

La mémoire de fonderie met l'accent sur la mémoire à long terme. Conservez l’état à court terme tour par tour dans l’état d’exécution LangChain ou LangGraph.

Conditions préalables

• Un abonnement Azure. Créez-en un gratuitement.
• Un projet Foundry.
• Modèle de conversation Microsoft Foundry déployé pour la récupération de mémoire.
  • Ce tutoriel utilise « gpt-4.1 ».
• Un modèle de conversation et un modèle d’intégration déployés pour le magasin de mémoire.
  • Ce didacticiel utilise text-embedding-3-large.
• Python 3.10 ou version ultérieure.
• Azure CLI connecté (az login) afin que DefaultAzureCredential puisse s’authentifier auprès du rôle Azure AI Developer.

Configurer votre environnement

Installez les packages requis pour ce didacticiel. Utiliser langchain-azure-ai pour l’intégration de LangChain et LangGraph, azure-ai-projects pour la gestion du magasin de mémoire et azure-identity pour l’authentification.

pip install -U "langchain-azure-ai" azure-ai-projects azure-identity

Définissez vos variables d’environnement que nous utilisons dans ce tutoriel :

export AZURE_AI_PROJECT_ENDPOINT="https://<resource>.services.ai.azure.com/api/projects/<project>"

Comprendre le modèle de mémoire

Foundry Memory stocke et récupère deux types de mémoire à long terme :

• Mémoire du profil utilisateur : faits et préférences utilisateur stables, tels que le nom préféré ou les contraintes alimentaires.
• Mémoire récapitulative de conversation : résumés de sujets de discussion antérieurs.

La mémoire utilise l’idée de « portée » pour partitionner les informations afin qu’elles puissent être stockées et récupérées de manière cohérente. Les étendues sont comme des identificateurs ou des clés pour organiser les informations.

• Vous pouvez utiliser les ID utilisateur comme identité stable pour la mémoire à long terme. Gardez-le identique entre les sessions pour le même utilisateur.
• Vous pouvez utiliser les ID de session comme identité de conversation à court terme. Modifiez-le pour chaque session de chat.
• Vous pouvez utiliser des ID de ressource comme identificateur stable pour la mémoire à long terme sur plusieurs utilisateurs.

Cette séparation permet à votre application de mémoriser les préférences utilisateur entre les sessions sans mélanger les conversations non liées.

Créer le stockage de mémoire

Avant de commencer, vous devez configurer un stockage de mémoire. Pour cette opération, utilisez le Kit de développement logiciel (SDK) des projets Microsoft Foundry azure-ai-projects.

import os

from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import (
	MemoryStoreDefaultDefinition,
	MemoryStoreDefaultOptions,
)
from azure.core.exceptions import ResourceNotFoundError
from azure.identity import DefaultAzureCredential

endpoint = os.environ["AZURE_AI_PROJECT_ENDPOINT"]
credential = DefaultAzureCredential()
client = AIProjectClient(endpoint=endpoint, credential=credential)

store_name = "lc-integration-test-store"
try:
    store = client.beta.memory_stores.get(store_name)
    print(f"✓ Memory store '{store_name}' already exists")
except ResourceNotFoundError:
    print(f"Creating memory store '{store_name}'...")
    definition = MemoryStoreDefaultDefinition(
        chat_model="gpt-4.1", 						# Change for your LLM model
        embedding_model="text-embedding-3-large",	# Change for your emebddings model
        options=MemoryStoreDefaultOptions(
            user_profile_enabled=True,
            chat_summary_enabled=True,
        ),
    )
    store = client.beta.memory_stores.create(
        name=store_name,
        description="Long-term memory store",
        definition=definition,
    )
    print(f"✓ Memory store '{store_name}' created successfully")

✓ Memory store 'lc-integration-test-store' created successfully

Ce que fait cet extrait de code : Se connecte à votre projet Foundry, obtient ou crée le magasin de mémoire et permet l'extraction du profil utilisateur ainsi que du résumé de chat.

Utilisation de la mémoire dans LangGraph et LangChain

Foundry Memory s’intègre à LangGraph et LangChain en introduisant deux objets :

• La classe langchain_azure_ai.chat_message_history.AzureAIMemoryChatMessageHistory crée un historique de conversation en mémoire.
• La classe langchain_azure_ai.retrievers.AzureAIMemoryRetriever permet de récupérer des souvenirs à partir de l’historique des messages de conversation.

En général, vous pouvez utiliser les stratégies de récupération pratiques suivantes :

• Récupérez la mémoire du profil utilisateur au début d’une conversation pour personnaliser les réponses.
• Récupérez la mémoire récapitulative de conversation en fonction du tour actuel pour récupérer le contexte antérieur approprié.

Exemple : Ajouter une couche mémoire prenant en compte la session

Dans cet exemple, nous créons un seul exécutable dans LangChain qui récupère la mémoire à long terme pertinente, l’injecte dans l’invite et exécute le modèle avec l’historique des conversations à court terme et la mémoire à long terme.

Voyons comment l’implémenter :

Créer l’historique des messages de conversation

Cet exemple utilise un user_id stable comme étendue de mémoire. Utiliser session_id pour le contexte de conversation par session.

from langchain_azure_ai.chat_message_histories import AzureAIMemoryChatMessageHistory
from langchain_azure_ai.retrievers import AzureAIMemoryRetriever
from langchain_core.chat_history import InMemoryChatMessageHistory

_session_histories: dict[tuple[str, str], AzureAIMemoryChatMessageHistory] = {}

def get_session_history(user_id: str, session_id: str) -> AzureAIMemoryChatMessageHistory:
    """Get or create a session history for a user and session.
    
    Args:
        user_id: Stable user identifier (used as scope in Foundry Memory)
        session_id: Ephemeral session identifier
        
    Returns:
        AzureAIMemoryChatMessageHistory instance
    """
    cache_key = (user_id, session_id)
    if cache_key not in _session_histories:
        _session_histories[cache_key] = AzureAIMemoryChatMessageHistory(
            project_endpoint=endpoint,
            credential=credential,
            store_name=store_name,
            scope=user_id,
            base_history=InMemoryChatMessageHistory(),
            update_delay=0,  # TEST MODE: process updates immediately (default ~300s)
        )
    return _session_histories[cache_key]


def get_foundry_retriever(user_id: str, session_id: str) -> AzureAIMemoryRetriever:
    """Get a retriever tied to the cached session history.
    
    This preserves incremental search state across turns.
    
    Args:
        user_id: Stable user identifier
        session_id: Ephemeral session identifier
        
    Returns:
        AzureAIMemoryRetriever instance
    """
    return get_session_history(user_id, session_id).get_retriever(k=5)

Ce que fait cet extrait de code : Crée un historique en mémoire et un module de récupération pour chaque paire (user_id, session_id) et les met en cache afin que l’état de récupération persiste à travers les tours dans la même session. Pour cette procédure pas à pas, update_delay=0 rend les mises à jour de la mémoire immédiatement visibles. En production, utilisez le délai par défaut, sauf si vous avez spécifiquement besoin d’une extraction instantanée. session_histories est utilisé pour éviter de devoir recréer constamment les objets.

Composer le programme exécutable avec récupération de mémoire

Nous allons créer un runnable pour implémenter la boucle :

from typing import Any
import os

from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import ConfigurableFieldSpec, RunnablePassthrough
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_azure_ai.chat_models import AzureAIChatCompletionsModel

llm = init_chat_model("azure_ai:gpt4.1" temperature=0.7)

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are helpful and concise. Use prior memories when relevant."),
        MessagesPlaceholder("history"),
        ("system", "Memories:\n{memories}"),
        ("human", "{question}"),
    ]
)


def chain_for_session(user_id: str, session_id: str) -> RunnableWithMessageHistory:
    """Create a chain with message history for a specific user and session.
    
    Args:
        user_id: Stable user identifier
        session_id: Ephemeral session identifier
        
    Returns:
        Runnable chain with message history
    """
    retriever = get_foundry_retriever(user_id, session_id)

    def format_memories(x: dict[str, Any]) -> str:
        """Retrieve and format memories as text."""
        docs = retriever.invoke(x["question"])
        return (
            "\n".join([doc.page_content for doc in docs])
            if docs
            else "No relevant memories found."
        )

    # Use RunnablePassthrough.assign to add memories to the input dict
    # RunnableWithMessageHistory will inject history automatically
    chain = RunnablePassthrough.assign(memories=format_memories) | prompt | llm

    chain_with_history = RunnableWithMessageHistory(
        chain,
        get_session_history=get_session_history,
        input_messages_key="question",
        history_messages_key="history",
        history_factory_config=[
            ConfigurableFieldSpec(
                id="user_id",
                annotation=str,
                name="User ID",
                description="Unique identifier for the user.",
                default="",
                is_shared=True,
            ),
            ConfigurableFieldSpec(
                id="session_id",
                annotation=str,
                name="Session ID",
                description="Unique identifier for the session.",
                default="",
                is_shared=True,
            ),
        ],
    )
    return chain_with_history

Fonctionnalité de cet extrait : crée un exécutable qui insère les souvenirs récupérés dans le prompt, puis les encapsule dans RunnableWithMessageHistory afin que l’historique des conversations et la mémoire à long terme fonctionnent ensemble.

Ce modèle maintient votre prompt déterministe : chaque tour inclut explicitement la mémoire récupérée dans la section Memories.

Exécuter un scénario intersession pratique

Ce scénario montre la valeur complète de la mémoire à long terme :

1. Dans la session A, l’utilisateur partage les préférences.
2. Dans la session B, l’application rappelle automatiquement ces préférences.

import time

user_id = "user_001"
session_id = "session_2026_02_10_001"
chain = chain_for_session(user_id, session_id)

# 4) Session A: seed preferences (long-term memory extraction happens async)
print(
	"\n=== Turn 1 (Session A): Introduce a preference "
	"(will be extracted into long-term memory) ==="
)
r1 = chain.invoke(
	{"question": "Hi! Call me JT. I prefer dark roast coffee and budget trips."},
	config={"configurable": {"user_id": user_id, "session_id": session_id}},
)
print("ASSISTANT:", r1.content)

print("\n=== Turn 2 (Session A): Add another preference ===")
r2 = chain.invoke(
	{
		"question": "Also, I usually drink green tea in the afternoon "
		"and I like staying in hostels."
	},
	config={"configurable": {"user_id": user_id, "session_id": session_id}},
)
print("ASSISTANT:", r2.content)

# Because we set update_delay=0, extraction should happen immediately for demo.
# If you use the default delay, you may need to wait before querying from new session.
time.sleep(60)

# 5) Cross-session test: same user_id, new session_id
session_id_b = "session_2026_02_10_002"
chain_b = chain_for_session(user_id, session_id_b)

print("\n=== Turn 3 (Session B): New session should recall coffee preference ===")
r4 = chain_b.invoke(
	{"question": "Remind me of my coffee preference and travel style."},
	config={"configurable": {"user_id": user_id, "session_id": session_id_b}},
)
print("ASSISTANT:", r4.content)

print("\n=== Turn 4 (Session B): Retrieve another preference ===")
r5 = chain_b.invoke(
	{
		"question": "What do I usually drink in the afternoon, "
		"and where do I like to stay?"
	},
	config={"configurable": {"user_id": user_id, "session_id": session_id_b}},
)
print("ASSISTANT:", r5.content)

=== Turn 1 (Session A) ===
ASSISTANT: Nice to meet you, JT. I noted that you prefer dark roast coffee and budget trips.

=== Turn 2 (Session A) ===
ASSISTANT: Got it. I also noted that you usually drink green tea in the afternoon and prefer hostels.

=== Turn 3 (Session B) ===
ASSISTANT: Your coffee preference is dark roast, and your travel style is budget trips.

=== Turn 4 (Session B) ===
ASSISTANT: You usually drink green tea in the afternoon, and you like staying in hostels.

Ce que fait cet extrait de code : Initialise les préférences utilisateur dans la session A, démarre la session B avec le même utilisateur et montre que l'application peut retrouver les préférences antérieures d'une session à l'autre.

Exemple : Interroger la mémoire directement pour les cas d’usage non-conversationnels

Utilisez un récupérateur ad hoc lorsque vous souhaitez effectuer des lectures directes de la mémoire en dehors du pipeline de conversation, par exemple dans le middleware de personnalisation ou les outils d’analyse de profil.

adhoc = AzureAIMemoryRetriever(
	project_endpoint=endpoint,
	credential=credential,
	store_name=store_name,
	scope=user_id,
	k=5,
)
print("\n=== Turn 5 (Ad-hoc): Direct retriever query without session history ===")
adhoc_docs = adhoc.invoke("What are my drinking preferences?")
for i, doc in enumerate(adhoc_docs, start=1):
	print(f"MEMORY {i}:", doc.page_content)

MEMORY 1: Prefers dark roast coffee.
MEMORY 2: Prefers budget trips.
MEMORY 3: Usually drinks green tea in the afternoon.
MEMORY 4: Likes staying in hostels.

Ce que fait cet extrait de code : Exécute une recherche en mémoire directe pour l’étendue actuelle. Tous les souvenirs sont récupérés (limités par k) mais triés par pertinence.

Utilisez ce modèle lorsque vous avez besoin de lectures directes de mémoire pour des fonctionnalités telles que des cartes de profil, un intergiciel de personnalisation ou un routage de flux de travail.

Exemple : Utiliser la mémoire dans les graphiques

LangGraph utilise le même modèle conceptuel :

• Maintenez user_id stable pour une mémoire à long terme.
• Utilisez thread_id (ou équivalent) pour le contexte de thread à court terme.
• Récupérez la mémoire avant d’appeler le nœud de modèle.

Si vous avez déjà un StateGraph, injectez la récupération dans votre nœud de modèle et ajoutez le texte de mémoire à votre entrée de modèle. Une autre stratégie classique consiste à utiliser un crochet prémodélisé.

from langgraph.graph import MessagesState


def call_model_with_foundry_memory(state: MessagesState, config: dict):
	user_id = config["configurable"]["user_id"]
	session_id = config["configurable"]["thread_id"]
	query = state["messages"][-1].content

	retriever = get_foundry_retriever(user_id, session_id)
	docs = retriever.invoke(query)
	memory_text = "\n".join(d.page_content for d in docs) if docs else ""

	response = llm.invoke(
		[
			{"role": "system", "content": "Use prior memories when relevant."},
			{"role": "system", "content": f"Memories:\n{memory_text}"},
			*state["messages"],
		]
	)
	return {"messages": [response]}

Ce que fait cet extrait de code : Affiche un modèle de nœud LangGraph qui récupère la mémoire Foundry pour le tour actuel et l’injecte dans l’entrée du modèle.

Pour plus d’informations sur les concepts de mémoire LangGraph plus larges, consultez :

• Ajouter de la mémoire dans LangGraph

Comprendre les limites de l'aperçu et les directives opérationnelles

Avant de passer en production, validez ces contraintes :

• La mémoire est en version préliminaire et son comportement peut changer.
• La mémoire nécessite des déploiements de conversation et d’incorporation compatibles.
• Les quotas s’appliquent par magasin et par étendue, y compris les taux de recherche et de demande de mise à jour.

Prévoyez également des contrôles défensifs contre les tentatives d’empoisonnement de la mémoire ou d’injection de commandes. Validez les entrées non approuvées avant d’influencer la mémoire stockée.

Nettoyer les ressources

Après avoir exécuté des exemples, supprimez l’étendue pour éviter la fuite de données de test dans les exécutions ultérieures.

result = client.memory_stores.delete_scope(name=store_name, scope=user_id)
print(
	f"Deleted {getattr(result, 'deleted_count', 'all')} memories "
	f"for scope '{user_id}'."
)

Deleted 4 memories for scope 'user_001'.

Suivre les applications LangChain et LangGraph

Contenu connexe

• Utiliser le service d'agent Foundry avec LangGraph
• Commencez avec les SDKs de Microsoft Foundry et points-de-terminaison
• Utiliser la mémoire avec le service de l'Agent Foundry
• Qu’est-ce que la mémoire dans le service d'agent Foundry ?