Použití paměti Foundry s jazykem LangChain a LangGraph

K přidání dlouhodobé paměti do aplikací použijte langchain-azure-ai a Foundry Memory. V tomto článku vytvoříte řetěz založený na paměti, uložíte uživatelské předvolby, vzpomenete si je v nové relaci a spustíte přímé dotazy na paměť.

Tento model funguje pro aplikace LangChain i LangGraph. Základní myšlenkou je zachovat krátkodobou historii chatu v modulu runtime a používat Foundry Memory jako dlouhodobé úložiště kontextu na úrovni uživatele.

Foundry Memory se zaměřuje na dlouhodobou paměť. Uložte krátkodobý stav pro turn-by-turn navigaci ve stavu běhového prostředí LangChain nebo LangGraph.

Požadavky

• Předplatné Azure. Vytvořte si ho zdarma.
• Projekt Foundry.
• Použitý model chatu Microsoft Foundry pro vyhledávání paměti.
  • V tomto kurzu se používá gpt-4.1.
• Nasazený model chatu a vložený model pro úložiště paměti.
  • Tento tutoriál používá text-embedding-3-large.
• Python 3.10 nebo novější.
• Azure CLI je přihlášen (az login), aby se DefaultAzureCredential mohl autentizovat s rolí Azure AI Developer.

Konfigurace prostředí

Nainstalujte požadované balíčky pro tento kurz. Používá se langchain-azure-ai pro integraci LangChain a LangGraph, azure-ai-projects pro správu úložiště paměti a azure-identity pro ověřování.

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

Nastavte proměnné prostředí, které používáme v tomto kurzu:

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

Vysvětlení modelu paměti

Fondrová paměť ukládá a načítá dva typy dlouhodobé paměti:

• Paměť profilu uživatele: stabilní fakta a předvolby uživatelů, například upřednostňované jméno nebo dietní omezení.
• Souhrnná paměť chatu: destilované souhrny předchozích diskuzních témat.

Paměť používá myšlenku "scope" k rozdělení informací, aby je bylo možné ukládat a načítat konzistentně. Obory jsou jako identifikátory nebo klíče pro uspořádání informací.

• ID uživatelů můžete použít jako stabilní identitu pro dlouhodobou paměť. Udržujte jej konzistentní mezi relacemi pro stejného uživatele.
• ID relací můžete použít jako identitu krátkodobé konverzace. Změňte ji pro každou chatovací relaci.
• ID prostředků můžete použít jako stabilní identifikátor dlouhodobé paměti pro více uživatelů.

Díky tomuto oddělení si vaše aplikace pamatuje uživatelské předvolby napříč relacemi, aniž byste museli kombinovat nesouvisející konverzace.

Vytvoření úložiště paměti

Než začnete, musíte vytvořit úložiště paměti. Pro tuto operaci použijte sadu SDK Microsoft Foundry projects 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

Co tento fragment kódu dělá: Připojí se k projektu Foundry, získá nebo vytvoří úložiště paměti a povolí profil uživatele a extrakci souhrnu chatu.

Použití paměti v jazyce LangGraph a LangChain

Foundry Memory se integruje do LangGraphu a LangChainu zavedením dvou objektů:

• Třída langchain_azure_ai.chat_message_history.AzureAIMemoryChatMessageHistory vytvoří historii chatu založené na paměti.
• Třída langchain_azure_ai.retrievers.AzureAIMemoryRetriever umožňuje načítání pamětí z historie zpráv chatu.

Obecně platí, že s nimi můžete použít následující praktické strategie načítání:

• Načtení paměti profilu uživatele v rané fázi konverzace za účelem přizpůsobení odpovědí
• Načtení souhrnné paměti chatu na základě aktuálního turnu pro obnovení relevantního předchozího kontextu.

Příklad: Přidání vrstvy paměti pracující s relacemi

V tomto příkladu vytvoříme jeden spustitelný skript v LangChain, který načte relevantní dlouhodobou paměť, vloží ho do vstupu a spustí model, který společně využívá krátkodobou historii chatu a dlouhodobou paměť.

Pojďme se podívat, jak ji implementovat:

Vytvoření historie zpráv chatu

Tento příklad používá stabilní user_id jako paměťový rozsah. Použijte session_id pro kontext konverzace v rámci jednotlivých relací.

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)

Co tento fragment kódu dělá: Vytvoří historii zálohovanou v paměti pro každý pár (user_id, session_id) a uloží ji do mezipaměti, aby stav načítání přetrval během více tahů ve stejné relaci. V tomto názorném postupu činí update_delay=0 aktualizace paměti okamžitě viditelné. V produkčním prostředí použijte výchozí prodlevu, pokud výslovně nepotřebujete okamžitou extrakci. session_histories se používá k tomu, aby se zabránilo opakovanému vytváření objektů.

Vytvoření spustitelného příkazu s načtením paměti

Pojďme vytvořit objekt typu Runnable pro implementaci smyčky:

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

Co tento fragment kódu dělá: Vytvoří spustitelný objekt, který vloží načtené vzpomínky do výzvy, a pak je zabalí pomocí RunnableWithMessageHistory, aby historie chatu a dlouhodobá paměť spolupracovaly.

Tento vzor udržuje vaši výzvu deterministickou: každá otočka explicitně obsahuje načtenou paměť v oddílu Memories.

Spusťte praktický scénář napříč relacemi

Tento scénář ukazuje plnou hodnotu dlouhodobé paměti:

1. V relaci A uživatel sdílí předvolby.
2. V relaci B aplikace tyto předvolby automaticky připomene.

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.

Co tento fragment kódu dělá: Nastavení uživatelských předvoleb v relaci A, spustí relaci B pro stejného uživatele a ukazuje, že aplikace si může vzpomenout na předchozí předvolby napříč relacemi.

Příklad: Dotazování paměti přímo pro případy použití mimo chat

Ad-hoc retriever použijte, pokud chcete provádět přímé čtení paměti mimo konverzační proces, například v middleware systému pro přizpůsobení nebo nástrojích pro inspekci profilu.

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.

Co tento fragment kódu dělá: Spustí přímé vyhledávání paměti pro aktuální obor. Všechny vzpomínky jsou načteny (omezeny na k) a seřazeny podle relevance.

Tento vzor použijte, když potřebujete přímé čtení paměti pro funkce, jako jsou karty profilů, middleware pro přizpůsobení nebo směrování pracovního postupu.

Příklad: Použití paměti v grafech

LangGraph používá stejný koncepční vzor.

• Udržujte user_id stabilní pro dlouhodobé uchování paměti.
• Použijte thread_id (nebo ekvivalentní) pro krátkodobý kontext vlákna.
• Před voláním uzlu modelu načtěte paměť.

Pokud už máte StateGraph, vložte načtení do svého uzlu modelu a přidejte text paměti ke vstupu modelu. Další typickou strategií je použití předmodelového háku.

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

Co tento fragment kódu dělá: Zobrazuje vzor uzlu LangGraph, který načte paměť Foundry pro aktuální změnu a vloží ji do vstupu modelu.

Širší koncepty paměti jazyka LangGraph najdete tady:

• Přidání paměti v jazyce LangGraph

Vysvětlení limitů preview a provozních pokynů

Před přechodem do produkčního prostředí ověřte tato omezení:

• Paměť je ve verzi Preview a chování se může změnit.
• Paměť vyžaduje kompatibilní nasazení chatu a integrace.
• Kvóty se vztahují na úložiště a na rozsah, včetně sazeb žádostí o vyhledávání a aktualizaci.

Také naplánujte obranná opatření pro otrávení paměti nebo pokusy o injektáž vstupních parametrů. Ověřte nedůvěryhodné vstupy, než ovlivní uloženou paměť.

Vyčištění prostředků

Po spuštění vzorků odstraňte rozsah, abyste zabránili úniku testovacích dat do budoucích spuštění testů.

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

Sledujte aplikace LangChain a LangGraph

Související obsah

• Použijte službu Foundry Agent s LangGraph
• Začněte se sadami SDK Microsoft Foundry a koncovými body
• Použití paměti se službou Foundry Agent Service
• Co je paměť ve službě Foundry Agent Service