Foundry Memory gebruiken met LangChain en LangGraph

Gebruik langchain-azure-ai en Foundry Memory om langetermijngeheugen toe te voegen aan uw toepassingen. In dit artikel maakt u een keten met geheugensteun, slaat u gebruikersvoorkeuren op, herinnert u deze in een nieuwe sessie en voert u directe geheugenquery's uit.

Dit patroon werkt voor zowel LangChain- als LangGraph-toepassingen. Het belangrijkste idee is om de korte-termijnchatgeschiedenis in uw runtime te houden en Foundry Memory te gebruiken als het langetermijnarchief voor context op gebruikersniveau.

Foundry Memory richt zich op langetermijngeheugen. Houd de turn-by-turn-status op korte termijn in de runtimestatus van LangChain of LangGraph.

Voorwaarden

• Een Azure-abonnement. Maak er gratis een.
• Een Foundry-project.
• Een geïmplementeerd Microsoft Foundry-chatmodel voor het ophalen van geheugen.
  • In deze zelfstudie wordt 'gpt-4.1' gebruikt.
• Een geïmplementeerd chatmodel en insluitmodel voor het geheugenarchief.
  • In deze zelfstudie wordt gebruikgemaakt van text-embedding-3-large.
• Python 3.10 of hoger.
• Azure CLI aangemeld (az login), zodat DefaultAzureCredential kan verifiëren met rol Azure AI Developer.

Uw omgeving configureren

Installeer de vereiste pakketten voor deze zelfstudie. Gebruik langchain-azure-ai voor LangChain- en LangGraph-integratie, azure-ai-projects voor geheugenopslagbeheer en azure-identity voor verificatie.

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

Stel uw omgevingsvariabelen in die we in deze zelfstudie gebruiken:

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

Inzicht in het geheugenmodel

Foundry Memory slaat twee langetermijngeheugentypen op en haalt deze op:

• Geheugen van gebruikersprofiel: stabiele feiten en voorkeuren van gebruikers, zoals voorkeursnaam of dieetbeperkingen.
• Geheugensamenvatting voor chats: beknopte samenvattingen van voorafgaande discussieonderwerpen.

Geheugen maakt gebruik van het idee 'bereik' om gegevens te partitioneren, zodat deze consistent kunnen worden opgeslagen en opgehaald. Scopes zijn vergelijkbaar met identificatoren of sleutels om informatie te ordenen.

• U kunt gebruikers-id's gebruiken als de stabiele identiteit voor langetermijngeheugen. Zorg ervoor dat de sessie voor dezelfde gebruiker hetzelfde blijft.
• U kunt sessie-id's gebruiken als de kortetermijngespreksidentiteit. Wijzig deze per chatsessie.
• U kunt resource-id's gebruiken als de stabiele id voor langetermijngeheugen voor meerdere gebruikers.

Met deze scheiding kan uw app gebruikersvoorkeuren onthouden in sessies zonder niet-gerelateerde gesprekken te combineren.

Het geheugenarchief maken

Voordat u aan de slag gaat, moet u een geheugenopslag maken. Gebruik voor deze bewerking de SDK 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

Wat dit codefragment doet: Maakt verbinding met uw Foundry-project, haalt of maakt het geheugenarchief op en maakt gebruikersprofiel plus extractie van chatsamenvatting mogelijk.

Geheugen gebruiken in LangGraph en LangChain

Foundry Memory integreert in LangGraph en LangChain door twee objecten te introduceren:

• De klasse langchain_azure_ai.chat_message_history.AzureAIMemoryChatMessageHistory maakt een chatgeschiedenis met geheugensteun.
• Met de klasse langchain_azure_ai.retrievers.AzureAIMemoryRetriever kunnen herinneringen worden opgehaald uit de geschiedenis van het chatbericht.

In het algemeen kunt u de volgende praktische ophaalstrategieën gebruiken:

• Haal het geheugen van het gebruikersprofiel vroeg in een gesprek op om reacties aan te passen.
• Geheugen voor chatsamenvatting ophalen op basis van de huidige beurt om relevante eerdere context te herstellen.

Voorbeeld: Een sessiebewuste geheugenlaag toevoegen

In dit voorbeeld bouwen we een enkel uitvoerbaar bestand in LangChain dat relevante langetermijnherinneringen ophaalt, deze in de prompt injecteert en het model uitvoert met zowel de korte-termijn chatgeschiedenis als de langetermijnherinneringen samen.

Laten we eens kijken hoe u deze implementeert:

De chatberichtgeschiedenis maken

In dit voorbeeld wordt een stabiel user_id geheugenbereik gebruikt. Gebruiken session_id voor gesprekscontext per sessie.

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)

Wat dit codefragment doet: Maakt een geheugenondersteunde geschiedenis en opzoeker per (user_id, session_id) paar aan en slaat deze in de cache op, zodat de ophaaltoestand in dezelfde sessie behouden blijft. Voor deze instructie maken update_delay=0 geheugenupdates onmiddellijk zichtbaar. Gebruik in productie de standaardvertraging, tenzij u specifiek directe extractie nodig hebt. session_histories wordt gebruikt om te voorkomen dat de objecten voortdurend opnieuw moeten worden gemaakt.

De runnable opstellen met het ophalen van geheugen

Laten we een runnable maken om de lus te implementeren:

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

Wat dit codefragment doet: Bouwt een runnable die opgehaalde geheugens in de prompt injecteert en deze vervolgens verpakt met RunnableWithMessageHistory zodat chatgeschiedenis en langetermijngeheugen samenwerken.

Met dit patroon blijft uw prompt deterministisch: elke beurt bevat expliciet opgehaald geheugen in de Memories sectie.

Een praktisch scenario voor meerdere sessies uitvoeren

In dit scenario ziet u de volledige waarde van het langetermijngeheugen:

1. In sessie A deelt de gebruiker voorkeuren.
2. In sessie B roept de app deze voorkeuren automatisch in.

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.

Wat dit codefragment doet: Plant gebruikersvoorkeuren in sessie A, start sessie B voor dezelfde gebruiker en toont dat de app eerdere voorkeuren kan onthouden tussen sessies.

Voorbeeld: Rechtstreeks query's uitvoeren op geheugen voor niet-chatgebruiksscenario's

Gebruik een ad-hoc-retriever wanneer u direct geheugen wilt lezen buiten de gesprekspijplijn, bijvoorbeeld in hulpprogramma's voor personalisatie-middleware of profielinspectie.

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.

Wat dit codefragment doet: Voert een directe geheugenzoekactie uit voor het huidige bereik. Alle herinneringen worden opgehaald (beperkt door k) maar gesorteerd op relevantie.

Gebruik dit patroon wanneer u directe geheugenleesbewerkingen nodig hebt voor functies zoals profielkaarten, personalisatie-middleware of werkstroomroutering.

Voorbeeld: Geheugen gebruiken in grafieken

LangGraph maakt gebruik van hetzelfde conceptuele patroon:

• Houd user_id stabiel voor langetermijngeheugen.
• Gebruik thread_id (of gelijkwaardig) voor threadcontext op korte termijn.
• Geheugen ophalen voordat u het modelknooppunt aanroept.

Als u al een StateGraph heeft, injecteer het ophalen in het modelknooppunt en voeg de geheugentekst toe aan uw modelinvoer. Een andere typische strategie is om een premodelhook te gebruiken.

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

Wat dit codefragment doet: Toont een LangGraph-knooppuntpatroon waarmee Foundry-geheugen voor de huidige draai wordt opgehaald en in modelinvoer wordt ingevoerd.

Zie voor bredere LangGraph-geheugenconcepten:

• Geheugen toevoegen in LangGraph

Inzicht in preview-limieten en operationele richtlijnen

Voordat u naar productie gaat, valideert u deze beperkingen:

• Geheugen bevindt zich in de proeffase en het gedrag kan veranderen.
• Geheugen vereist compatibele chat- en insluitimplementaties.
• Quota zijn van toepassing per winkel en per bereik, inclusief zoek- en updateaanvraagtarieven.

Plan ook defensieve maatregelen voor prompt-injectiepogingen of geheugenvergiftiging. Valideer niet-vertrouwde invoer voordat ze invloed hebben op opgeslagen geheugen.

Resources opschonen

Nadat u de monsters hebt uitgevoerd, verwijdert u de scope om te voorkomen dat testgegevens in toekomstige uitvoeringen worden gelekt.

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

Verwante inhoud

• Foundry Agent Service gebruiken met LangGraph
• Begin met Microsoft Foundry SDK's en eindpunten
• Geheugen gebruiken met Foundry Agent Service
• Wat houdt geheugen in Foundry Agent Service in?