Używanie pamięci Foundry z bibliotekami LangChain i LangGraph

Użyj langchain-azure-ai i Foundry Memory, aby dodać pamięć długoterminową do aplikacji. W tym artykule tworzysz łańcuch oparty na pamięci, przechowujesz preferencje użytkownika, przywołujesz je w nowej sesji i uruchamiasz zapytania dotyczące pamięci bezpośredniej.

Ten wzorzec działa zarówno w przypadku aplikacji LangChain, jak i LangGraph. Podstawowym pomysłem jest przechowywanie krótkoterminowej historii czatów w systemie operacyjnym i używanie Foundry Memory jako długoterminowego magazynu dla kontekstu na poziomie użytkownika.

Pamięć odlewni koncentruje się na pamięci długoterminowej. Zachowaj krótkoterminowy stan kolejnych kroków w stanie podczas uruchamiania LangChain lub LangGraph.

Wymagania wstępne

• Subskrypcja Azure. Utwórz je bezpłatnie.
• Projekt Foundry.
• Wdrożony model czatu Microsoft Foundry na potrzeby pobierania pamięci.
  • W tym samouczku używa się modelu "gpt-4.1".
• Wdrożony model czatu i model osadzania dla magazynu pamięci.
  • W tym samouczku jest używany program text-embedding-3-large.
• Python 3.10 lub nowsza.
• Azure CLI zalogowany (az login), aby DefaultAzureCredential mogło uwierzytelniać się z rolą Azure AI Developer.

Konfigurowanie środowiska

Zainstaluj wymagane pakiety na potrzeby tego samouczka. Służy langchain-azure-ai do integracji LangChain i LangGraph, azure-ai-projects do zarządzania magazynem pamięci i azure-identity do uwierzytelniania.

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

Ustaw zmienne środowiskowe, które są używane w tym samouczku:

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

Omówienie modelu pamięci

Foundry Memory przechowuje i pobiera dwa typy pamięci długoterminowej.

• Pamięć profilu użytkownika: stabilne fakty i preferencje użytkownika, takie jak preferowane nazwy lub ograniczenia żywieniowe.
• Pamięć podsumowania czatu: destylowane podsumowania wcześniejszych tematów dyskusji.

Pamięć używa pojęcia "zakres" do partycjonowania informacji, aby można je było przechowywać i pobierać spójnie. Zakresy działają jak identyfikatory lub klucze służące do organizowania informacji.

• Identyfikatory użytkowników można używać jako stabilnej tożsamości dla pamięci długoterminowej. Zachowaj to samo między sesjami dla tego samego użytkownika.
• Identyfikatory sesji można używać jako krótkoterminowej tożsamości konwersacji. Zmień podczas sesji czatu.
• Identyfikatory zasobów można używać jako stabilnego identyfikatora pamięci długoterminowej dla wielu użytkowników.

Ta separacja pozwala aplikacji zapamiętać preferencje użytkownika między sesjami bez mieszania niepowiązanych konwersacji.

Tworzenie magazynu pamięci

Przed rozpoczęciem należy utworzyć magazyn pamięci. W przypadku tej operacji użyj zestawu SDK projektów 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

Co robi ten fragment kodu: Nawiązuje połączenie z projektem Foundry, pobiera lub tworzy magazyn pamięci oraz umożliwia wyodrębnianie podsumowania czatu oraz profil użytkownika.

Używanie pamięci w języku LangGraph i LangChain

Funkcja Foundry Memory integruje się w języku LangGraph i LangChain, wprowadzając dwa obiekty:

• Klasa langchain_azure_ai.chat_message_history.AzureAIMemoryChatMessageHistory tworzy historię czatów opartych na pamięci.
• Klasa langchain_azure_ai.retrievers.AzureAIMemoryRetriever umożliwia pobieranie wspomnień z historii wiadomości czatu.

Ogólnie rzecz biorąc, można użyć następujących praktycznych strategii pobierania, korzystając z nich:

• Pobieranie pamięci profilu użytkownika na wczesnym etapie konwersacji w celu spersonalizowania odpowiedzi.
• Pobierz pamięć podsumowania czatu na podstawie bieżącej rundy, aby odtworzyć odpowiedni poprzedni kontekst.

Przykład: Dodawanie warstwy pamięci obsługującej sesję

W tym przykładzie budujemy pojedynczy element wykonalny w frameworku LangChain, który pobiera odpowiednią pamięć długoterminową, integruje ją z promptem i uruchamia model, wykorzystując równocześnie historię krótkoterminowej rozmowy i pamięć długoterminową.

Zobaczmy, jak go zaimplementować:

Tworzenie historii wiadomości czatu

W tym przykładzie użyto stabilnego user_id zakresu pamięci. Użyj session_id do kontekstu konwersacji na sesję.

from langchain_azure_ai.chat_history 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 robi ten fragment kodu: Tworzy historię wspieraną przez pamięć i narzędzie pobierające dla każdej pary (user_id, session_id) i je buforuje, tak aby stan pobierania był utrzymywany w trakcie kolejnych tur w tej samej sesji. W tym przewodniku update_delay=0 aktualizacje pamięci są natychmiast widoczne. W środowisku produkcyjnym użyj opóźnienia domyślnego, chyba że potrzebujesz natychmiastowego wyodrębnienia. session_histories służy do unikania konieczności ciągłego ponownego tworzenia obiektów.

Tworzenie elementu runnable za pomocą pobierania pamięci

Utwórzmy element runnable w celu zaimplementowania pętli:

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

llm = init_chat_model("azure_ai:gpt-4.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 robi ten fragment kodu: Tworzy element runnable, który wprowadza pobrane wspomnienia do monitu, a następnie opakowuje je tak RunnableWithMessageHistory , aby historia czatu i długoterminowa pamięć współpracowała ze sobą.

Ten wzorzec zapewnia deterministyczne zapytanie: każda tura jawnie zawiera pobraną pamięć w sekcji Memories.

Uruchom praktyczny scenariusz wielosesyjny

W tym scenariuszu przedstawiono pełną wartość pamięci długoterminowej:

1. W sesji A użytkownik udostępnia preferencje.
2. W sesji B aplikacja automatycznie wspomina te preferencje.

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 robi ten fragment kodu: Inicjuje preferencje użytkownika w sesji A, uruchamia sesję B dla tego samego użytkownika i pokazuje, że aplikacja może przypominać wcześniejsze preferencje pomiędzy sesjami.

Przykład: wykonywanie zapytań o pamięć bezpośrednio w przypadku przypadków użycia innych niż czat

Użyj pobieracza ad-hoc, gdy chcesz uzyskać bezpośredni dostęp do pamięci poza przepływem konwersacji, na przykład w przypadku oprogramowania pośredniczącego do personalizacji lub narzędzi do przeglądania 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 robi ten fragment kodu: Uruchamia bezpośrednie wyszukiwanie pamięci dla bieżącego zakresu. Wszystkie wspomnienia są pobierane (ograniczone przez k), ale posortowane według istotności.

Użyj tego wzorca, gdy potrzebujesz bezpośrednich odczytów pamięci dla funkcji, takich jak karty profilów, oprogramowanie pośredniczące personalizacji lub routing przepływu pracy.

Przykład: używanie pamięci w grafach

LangGraph używa tego samego wzorca koncepcyjnego:

• Zachowaj stabilność user_id pamięci długoterminowej.
• Użyj thread_id (lub równoważnego) dla kontekstu wątku tymczasowego.
• Pobierz pamięć przed wywołaniem węzła modelu.

Jeśli masz już StateGraph, wstrzyknij proces pobierania danych do węzła modelu i dołącz tekst pamięci do danych wejściowych modelu. Inną typową strategią jest użycie haka przedmodelowego.

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 robi ten fragment kodu: Przedstawia wzorzec węzła LangGraph, który pobiera pamięć Foundry dla obecnej tury i wstrzykuje ją do danych wejściowych modelu.

Aby zapoznać się z szerszymi pojęciami dotyczącymi pamięci langgraph, zobacz:

• Dodawanie pamięci w języku LangGraph

Omówienie limitów wersji zapoznawczej i wskazówek operacyjnych

Przed przejściem do środowiska produkcyjnego zweryfikuj następujące ograniczenia:

• Pamięć jest w wersji testowej, a jej zachowanie może ulec zmianie.
• Pamięć wymaga zgodnych wdrożeń dotyczących czatu oraz osadzania.
• Limity obowiązują dla każdego sklepu i zakresu, w tym wskaźników żądań wyszukiwania i aktualizacji.

Zaplanuj również środki ochronne pod kątem zatrucia pamięci lub ataków typu prompt injection. Zweryfikuj niezaufane dane wejściowe, zanim wpłyną na przechowywaną pamięć.

Czyszczenie zasobów

Po przetworzeniu próbek usuń zakres, aby uniknąć wycieku danych testowych podczas przyszłych uruchomień.

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

Aplikacje Trace LangChain i LangGraph

Powiązana zawartość

• Używaj usługi agenta Foundry z LangGraph
• Rozpocznij pracę z zestawami SDK i punktami końcowymi Microsoft Foundry
• Używanie pamięci z usługą agenta programu Foundry
• Co to jest pamięć w agencie usługi Foundry