LangChain ve LangGraph ile Foundry Memory Kullanma

Uygulamalarınıza uzun süreli bellek eklemek için langchain-azure-ai ve Döküm Belleği kullanın. Bu makalede, bellek destekli bir zincir oluşturur, kullanıcı tercihlerini depolar, bunları yeni bir oturumda geri çağırır ve doğrudan bellek sorguları çalıştırırsınız.

Bu düzen hem LangChain hem de LangGraph uygulamalarında çalışır. Temel fikir, çalışma zamanınızda kısa süreli sohbet geçmişini tutmak ve kullanıcı düzeyinde bağlam için uzun vadeli depo olarak Foundry Belleği'ni kullanmaktır.

Foundry Memory uzun süreli belleğe odaklanır. LangChain veya LangGraph çalışma zamanı durumunda kısa süreli adım adım kontrol durumunu koruyun.

Önkoşullar

  • Azure aboneliği. Ücretsiz bir tane oluşturun.
  • Dökümhane projesi.
  • Dağıtılmış bir Microsoft Foundry sohbet modeli bellek geri getirme için kullanılır.
    • Bu öğreticide "gpt-4.1" kullanılır.
  • Bellek deposu için dağıtılan bir sohbet modeli ve ekleme modeli.
    • Bu öğreticide text-embedding-3-large kullanılır.
  • Python 3.10 veya üzeri.
  • Azure CLI (az login) oturum açtıktan sonra DefaultAzureCredential rol Azure Yapay Zeka Geliştirici ile kimlik doğrulaması yapabilir.

Ortamınızı yapılandırma

Bu öğretici için gerekli paketleri yükleyin. LangChain ve LangGraph tümleştirmesi için langchain-azure-ai kullanın, bellek deposu yönetimi için azure-ai-projects ve kimlik doğrulaması için azure-identity kullanın.

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

Bu öğreticide kullandığımız ortam değişkenlerinizi ayarlayın:

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

Bellek modelini anlama

Döküm belleği iki uzun süreli bellek türünü depolar ve alır:

  • Kullanıcı profili belleği: Tercih edilen ad veya diyet kısıtlamaları gibi kararlı kullanıcı bilgileri ve tercihleri.
  • Sohbet özeti belleği: Önceki tartışma konularının damıtılmış özetleri.

Bellek bilgileri bölümleme amacıyla "kapsam" fikrini kullanır, böylece tutarlı bir şekilde depolanabilir ve alınabilir. Kapsamlar, bilgileri düzenlemek için kullanılan tanımlayıcılar veya anahtarlar gibidir.

  • Uzun süreli bellek için kararlı kimlik olarak kullanıcı kimliklerini kullanabilirsiniz. Aynı kullanıcı için oturumlarda aynı kalsın.
  • Oturum kimliklerini kısa süreli konuşma kimliği olarak kullanabilirsiniz. Her sohbet oturumunda değiştirin.
  • Kaynak kimliklerini birden çok kullanıcı arasında uzun süreli bellek için kararlı tanımlayıcı olarak kullanabilirsiniz.

Bu ayrım, uygulamanızın ilgisiz konuşmaları karıştırmadan oturumlar arasında kullanıcı tercihlerini hatırlamasını sağlar.

Bellek depoyu oluşturma

Başlamadan önce bir bellek deposu oluşturmanız gerekir. Bu işlem için Microsoft Foundry projeleri SDK'sını azure-ai-projects kullanın.

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

Bu kod parçacığının yaptığı iş: Foundry projenize bağlanır, bellek depoyu alır veya oluşturur ve kullanıcı profili ile sohbet özeti ayıklamayı etkinleştirir.

LangGraph ve LangChain'de bellek kullanma

Foundry Memory, iki nesne sunarak LangGraph ve LangChain ile tümleşir:

  • sınıfı langchain_azure_ai.chat_message_history.AzureAIMemoryChatMessageHistory , bellek destekli bir sohbet geçmişi oluşturur.
  • sınıfı langchain_azure_ai.retrievers.AzureAIMemoryRetriever , sohbet iletisi geçmişinden belleklerin alınmasına izin verir.

Genel olarak, bunlarla aşağıdaki pratik alma stratejilerini kullanabilirsiniz:

  • Yanıtları kişiselleştirmek için konuşmanın başlarında kullanıcı profili belleği alın.
  • İlgili önceki bağlamı yeniden elde etmek için mevcut tur temel alınarak sohbet özet belleğini alın.

Örnek: Oturum farkında bellek katmanı ekleyin

Bu örnekte, LangChain'de ilgili uzun süreli belleği alan, istem içine yerleştiren ve modeli kısa vadeli sohbet geçmişiyle uzun süreli bellekle birlikte yürüten tek bir çalıştırılabilir oluşturacağız.

Şimdi uygulamayı görelim:

Sohbet iletisi geçmişini oluşturma

Bu örnekte kararlı bir bellek kapsamı olarak user_id kullanılır. Oturum başına konuşma bağlamı için kullanın session_id .

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)

Bu kod parçacığının yaptığı iş: Her (user_id, session_id) çifti için bellek destekli bir geçmiş ve geri alıcı oluşturur ve bunları önbelleğe alır, böylece alma durumu aynı oturum içerisinde devam eder. Bu kılavuz için update_delay=0 bellek güncellemelerini hemen görünür olmasını sağlar. Üretimde, özellikle anlık ayıklamaya ihtiyacınız olmadığı sürece varsayılan gecikmeyi kullanın. session_histories nesneleri sürekli yeniden oluşturmak zorunda kalmamak için kullanılır.

Bellek geri getirme ile çalıştırılabilir öğeyi oluşturma

Şimdi döngünün uygulanması için bir çalıştırılabilir oluşturalım:

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

Bu kod parçacığı ne yapar: Alınan bellekleri girdiye yerleştirerek, ardından sohbet geçmişiyle uzun süreli belleğin birlikte çalışmasını sağlamak için bunu RunnableWithMessageHistory ile sarar ve bir çalıştırılabilir oluşturur.

Bu desen, komut belirleyicinizi belirleyici tutar: her dönüşte alınan bellek Memories bölümünde açıkça yer alır.

Pratik bir oturumlar arası senaryo çalıştırma

Bu senaryo, uzun süreli belleğin tam değerini gösterir:

  1. A oturumunda kullanıcı tercihleri paylaşır.
  2. B oturumunda uygulama bu tercihleri otomatik olarak geri çağırır.
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.

Bu kod parçacığının yaptığı iş: A oturumundaki kullanıcı tercihlerini tohumlar, aynı kullanıcı için B oturumunu başlatır ve uygulamanın oturumlar arasında önceki tercihleri hatırlayabileceğini gösterir.

Örnek: Sohbet dışı kullanım örnekleri için belleği doğrudan sorgulama

Konuşma işlem hattının dışında, örneğin kişiselleştirme ara yazılımında veya profil inceleme araçlarında doğrudan bellek okumaları istediğinizde geçici bir retriever kullanın.

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.

Bu kod parçacığının yaptığı iş: Geçerli kapsam için doğrudan bellek araması çalıştırır. Tüm anılar alınır (sınırlandırılır k ile) ancak ilgi düzeyine göre sıralanır.

Profil kartları, kişiselleştirme ara yazılımı veya iş akışı yönlendirme gibi özellikler için doğrudan bellek okumalarına ihtiyacınız olduğunda bu düzeni kullanın.

Örnek: Grafiklerde bellek kullanma

LangGraph aynı kavramsal deseni kullanır:

  • Uzun süreli bellek için user_id'yi kararlı tutun.
  • Kısa vadeli iş parçacığı bağlamı için thread_id (veya eşdeğerini) kullanın.
  • Model düğümünü çağırmadan önce belleği alın.

Zaten bir StateGraph varsa, model düğümünüze geri alma ekleyin ve model girişinize bellek metni ekleyin. Bir diğer tipik strateji de model öncesi kanca kullanmaktır.

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

Bu kod parçacığının yaptığı iş: Mevcut tur için Foundry belleğini geri getiren ve bunu model girişine ekleyen bir LangGraph düğüm desenini gösterir.

Daha geniş LangGraph bellek kavramları için bkz:

Önizleme sınırlarını ve işlem kılavuzlarını anlama

Üretime geçmeden önce şu kısıtlamaları doğrulayın:

  • Bellek önizleme aşamasındadır ve davranış değişebilir.
  • Bellek için uyumlu sohbet ve gömme dağıtımları gerekir.
  • Kotalar, arama ve güncelleştirme isteği oranları da dahil olmak üzere mağaza başına ve kapsam başına uygulanır.

Ayrıca bellek zehirlenmesi veya prompt enjeksiyonu girişimlerine karşı savunma denetimleri planlayın. Depolanmış belleği etkilemeden önce güvenilmeyen girişleri doğrulayın.

Kaynakları temizleme

Örnekleri çalıştırdıktan sonra, test verilerinin gelecekteki çalıştırmalara sızmasını önlemek için kapsamı silin.

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

LangChain ve LangGraph uygulamalarını izleme

İlgili içerik

  • Last updated on