Usare la memoria Foundry con LangChain e LangGraph

Usare langchain-azure-ai e Foundry Memory per aggiungere memoria a lungo termine alle applicazioni. In questo articolo si crea una catena supportata dalla memoria, si archiviano le preferenze utente, le richiamano in una nuova sessione ed eseguono query di memoria diretta.

Questo modello funziona sia per le applicazioni LangChain che per le applicazioni LangGraph. L'idea principale consiste nel mantenere la cronologia delle chat a breve termine nel runtime e usare Foundry Memory come archivio a lungo termine per il contesto a livello di utente.

Foundry Memory è incentrato sulla memoria a lungo termine. Mantenere lo stato turn-by-turn a breve termine nello stato di runtime di LangChain o LangGraph.

Prerequisiti

• Sottoscrizione Azure. Creane uno gratis.
• Un progetto Fonderia.
• Modello di chat Microsoft Foundry distribuito per il recupero della memoria.
  • Questa esercitazione usa "gpt-4.1".
• Modello di chat distribuito e modello di incorporamento per l'archivio di memoria.
  • In questa esercitazione viene usato text-embedding-3-large.
• Python 3.10 o versione successiva.
• L'interfaccia della riga di comando di Azure è connessa (az login) quindi DefaultAzureCredential può autenticarsi con il ruolo Azure AI Developer.

Configurare il tuo ambiente

Installare i pacchetti necessari per questa esercitazione. Usare langchain-azure-ai per l'integrazione LangChain e LangGraph, azure-ai-projects per la gestione dell'archivio memoria e azure-identity per l'autenticazione.

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

Impostare le variabili di ambiente usate in questa esercitazione:

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

Informazioni sul modello di memoria

Foundry Memory archivia e recupera due tipi di memoria a lungo termine:

• Memoria del profilo utente: dati e preferenze utente stabili, ad esempio il nome preferito o i vincoli alimentari.
• Memoria di riepilogo della chat: riepiloghi distillati degli argomenti di discussione precedenti.

Memoria usa l'idea di "ambito" per partizionare le informazioni in modo che possa essere archiviata e recuperata in modo coerente. Gli ambiti sono come identificatori o chiavi per organizzare le informazioni.

• È possibile usare gli ID utente come identità stabile per la memoria a lungo termine. Mantenere la stessa operazione tra le sessioni per lo stesso utente.
• È possibile usare gli ID sessione come identità di conversazione a breve termine. Modifica in base alla sessione di chat.
• È possibile usare gli ID risorsa come identificatore stabile per la memoria a lungo termine tra più utenti.

Questa separazione consente all'app di ricordare le preferenze utente tra le sessioni senza combinare conversazioni non correlate.

Creare il datastore di memoria

Prima di iniziare, è necessario creare un archivio memoria. Per questa operazione, usare Microsoft Foundry projects SDK 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

Cosa fa questo frammento di codice: Si connette al progetto Foundry, ottiene o crea l'archivio memoria e abilita l'estrazione del profilo utente più l'estrazione di riepilogo delle chat.

Uso della memoria in LangGraph e LangChain

Foundry Memory si integra in LangGraph e LangChain introducendo due oggetti:

• La classe langchain_azure_ai.chat_message_history.AzureAIMemoryChatMessageHistory crea una cronologia di chat supportata dalla memoria.
• La classe langchain_azure_ai.retrievers.AzureAIMemoryRetriever consente il recupero di ricordi dalla cronologia dei messaggi di chat.

In generale, è possibile usare le seguenti strategie pratiche di ricerca:

• Recuperare la memoria del profilo utente nelle prime fasi di una conversazione per personalizzare le risposte.
• Recuperare la memoria di riepilogo della chat in base al turno corrente per ripristinare il contesto precedente pertinente.

Esempio: Aggiungere un livello di memoria compatibile con la sessione

In questo esempio viene creato un singolo eseguibile in LangChain che recupera la memoria a lungo termine pertinente, lo inserisce nel prompt ed esegue il modello con la cronologia delle chat a breve termine e la memoria a lungo termine insieme.

Vediamo come implementarlo:

Creare la cronologia dei messaggi di chat

In questo esempio viene usata una classe stabile user_id come ambito di memoria. Usare session_id per il contesto di conversazione per sessione.

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)

Cosa fa questo frammento di codice: Crea una cronologia supportata dalla memoria e un retriever per (user_id, session_id) ogni coppia e li memorizza nella cache, in modo che lo stato di recupero superi i turni nella stessa sessione. Per questa guida, update_delay=0 rende immediatamente visibili gli aggiornamenti della memoria. Nell'ambiente di produzione usare il ritardo predefinito, a meno che non sia necessaria in modo specifico l'estrazione immediata. session_histories viene utilizzato per evitare di dover ricreare costantemente gli oggetti.

Comporre il runnable con il recupero della memoria

Creiamo un runnable per implementare il ciclo:

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

Cosa fa questo frammento di codice: Crea un codice eseguibile che inserisce i ricordi recuperati nel prompt, poi lo avvolge con RunnableWithMessageHistory in modo che la cronologia delle chat e la memoria a lungo termine funzionino insieme.

Questo modello mantiene il prompt deterministico: ogni turno include in modo esplicito la memoria recuperata nella sezione Memories.

Eseguire uno scenario pratico tra sessioni

Questo scenario mostra il valore completo della memoria a lungo termine:

1. Nella sessione A l'utente condivide le preferenze.
2. Nella sessione B, l'app richiama automaticamente tali preferenze.

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.

Cosa fa questo frammento di codice: Inizializzazione delle preferenze utente nella sessione A, avvia la sessione B per lo stesso utente e mostra che l'app può richiamare le preferenze precedenti tra le sessioni.

Esempio: Eseguire query sulla memoria direttamente per i casi d'uso non di chat

Usare un recuperatore dedicato quando si desidera effettuare letture dirette della memoria all'esterno della pipeline di conversazione, ad esempio nel middleware di personalizzazione o negli strumenti di ispezione del profilo.

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.

Funzione di questo frammento: esegue una ricerca nella memoria diretta dell'ambito corrente. Tutti i ricordi vengono recuperati (con estensione limitata da k) ma ordinati in base alla pertinenza.

Usare questo modello quando sono necessarie letture di memoria diretta per funzionalità quali schede profilo, middleware di personalizzazione o routing del flusso di lavoro.

Esempio: Usare la memoria nei grafici

LangGraph usa lo stesso modello concettuale:

• Mantenere user_id stabile per la memoria a lungo termine.
• Usare thread_id (o equivalente) per il contesto del thread a breve termine.
• Recuperare memoria prima di chiamare il nodo del modello.

Se si dispone già di un StateGraph, inserire il recupero nel nodo del modello e accodare il testo della memoria all'input del modello. Un'altra strategia tipica consiste nell'usare un hook pre-modello.

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

Cosa fa questo frammento di codice: Mostra un modello di nodo LangGraph che recupera la memoria Foundry per il turno corrente e lo inserisce nell'input del modello.

Per i concetti relativi alla memoria LangGraph più ampia, vedere:

• Aggiungere memoria in LangGraph

Informazioni sui limiti di anteprima e sulle linee guida operative

Prima di passare all'ambiente di produzione, convalidare questi vincoli:

• La funzionalità di memoria è in modalità anteprima e il comportamento potrebbe variare.
• La memoria richiede implementazioni di chat e incorporazione compatibili.
• Le quote si applicano per archivio e per ambito, incluse le tariffe delle richieste di ricerca e aggiornamento.

Pianificare anche controlli difensivi per tentativi di avvelenamento da memoria o inserimento di prompt. Convalidare gli input non attendibili prima di influenzare la memoria archiviata.

Pulire le risorse

Dopo l'esecuzione di esempi, eliminare l'ambito per evitare la perdita di dati di test nelle esecuzioni future.

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'.

Contenuti correlati

• Usare il Servizio agente di Foundry con LangGraph
• Introduzione agli SDK e agli endpoint di Microsoft Foundry
• Usare la memoria con il Servizio Agente Fonderia
• Che cos'è la memoria nel servizio agente Foundry