Menggunakan Memori Foundry dengan LangChain dan LangGraph

Gunakan langchain-azure-ai dan Memori Foundry untuk menambahkan memori jangka panjang ke aplikasi Anda. Dalam artikel ini, Anda membuat rantai yang didukung memori, menyimpan preferensi pengguna, memanggilnya kembali dalam sesi baru, dan menjalankan kueri memori langsung.

Pola ini berfungsi untuk aplikasi LangChain dan LangGraph. Ide intinya adalah menyimpan riwayat obrolan jangka pendek dalam runtime Anda dan menggunakan Memori Foundry sebagai penyimpanan jangka panjang untuk konteks tingkat pengguna.

Memori Foundry berfokus pada memori jangka panjang. Pertahankan status turn-by-turn jangka pendek dalam status runtime LangChain atau LangGraph.

Prasyarat

  • Langganan Azure. Buat satu secara gratis.
  • Proyek Foundry.
  • Model chat Microsoft Foundry yang diterapkan untuk pemulihan memori.
    • Tutorial ini menggunakan "gpt-4.1".
  • Model chat yang diterapkan dan model penghubung untuk penyimpanan memori.
    • Tutorial ini menggunakan text-embedding-3-large.
  • Python 3.10 atau yang lebih baru.
  • Azure CLI masuk (az login) sehingga DefaultAzureCredential dapat mengautentikasi dengan peran Azure Pengembang AI.

Mengonfigurasi lingkungan Anda

Instal paket yang diperlukan untuk tutorial ini. Gunakan langchain-azure-ai untuk integrasi LangChain dan LangGraph, azure-ai-projects untuk manajemen penyimpanan memori, dan azure-identity untuk autentikasi.

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

Atur variabel lingkungan Anda yang kami gunakan dalam tutorial ini:

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

Memahami model memori

Memori Foundry menyimpan dan mengakses dua tipe memori jangka panjang:

  • Memori profil pengguna: fakta dan preferensi pengguna yang stabil, seperti nama pilihan atau batasan diet.
  • Memori ringkasan obrolan: ringkasan disulur topik diskusi sebelumnya.

Memori menggunakan gagasan "cakupan" untuk mempartisi informasi sehingga dapat disimpan dan diambil secara konsisten. Seperti pengidentifikasi atau kunci, cakupan digunakan untuk mengatur informasi.

  • Anda dapat menggunakan ID pengguna sebagai identitas stabil untuk memori jangka panjang. Jaga agar tetap sama di seluruh sesi untuk pengguna yang sama.
  • Anda dapat menggunakan ID sesi sebagai identitas percakapan jangka pendek. Ubah pengaturan untuk setiap sesi obrolan.
  • Anda dapat menggunakan ID sumber daya sebagai pengidentifikasi stabil untuk memori jangka panjang di beberapa pengguna.

Pemisahan ini memungkinkan aplikasi Anda mengingat preferensi pengguna di seluruh sesi tanpa mencampur percakapan yang tidak terkait.

Membuat penyimpanan memori

Sebelum memulai, Anda perlu membuat penyimpanan memori. Untuk operasi ini, gunakan proyek Microsoft Foundry 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

Apa yang dilakukan cuplikan ini: Menyambungkan ke proyek Foundry Anda, mengambil atau membuat penyimpanan memori, dan memungkinkan ekstraksi profil pengguna beserta ringkasan obrolan.

Menggunakan memori di LangGraph dan LangChain

Memori Foundry terintegrasi dalam LangGraph dan LangChain dengan memperkenalkan dua objek:

  • Kelas langchain_azure_ai.chat_message_history.AzureAIMemoryChatMessageHistory membuat riwayat obrolan yang disimpan dalam memori.
  • langchain_azure_ai.retrievers.AzureAIMemoryRetriever Kelas ini memungkinkan pengambilan memori dari riwayat pesan obrolan.

Secara umum, Anda dapat menggunakan strategi pengambilan praktis berikut dengan mereka:

  • Ambil memori profil pengguna lebih awal dalam percakapan untuk mempersonalisasi respons.
  • Ambil memori ringkasan obrolan berdasarkan giliran saat ini untuk memulihkan konteks sebelumnya yang relevan.

Contoh: Menambahkan lapisan memori yang peka terhadap sesi

Dalam contoh ini, kami membangun satu runnable di LangChain yang mengambil memori jangka panjang yang relevan, menyuntikkannya ke dalam prompt, dan menjalankan model dengan riwayat obrolan jangka pendek dan memori jangka panjang bersama-sama.

Mari kita lihat cara mengimplementasikannya:

Membuat riwayat pesan obrolan

Contoh ini menggunakan stabil user_id sebagai cakupan memori. Gunakan session_id untuk konteks percakapan per sesi.

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)

Apa yang dilakukan cuplikan ini: Membuat riwayat berbasis memori dan retriever untuk setiap pasangan (user_id, session_id) dan menyimpannya dalam cache sehingga kondisi pengambilan bertahan di seluruh interaksi dalam sesi yang sama. Untuk panduan ini, update_delay=0 membuat pembaruan memori yang segera terlihat. Dalam produksi, gunakan penundaan default kecuali Anda secara khusus memerlukan ekstraksi instan. session_histories digunakan untuk menghindari harus membuat ulang objek secara konstan.

Buat yang dapat dijalankan dengan pengambilan memori

Mari kita buat program yang dapat dijalankan untuk melaksanakan perulangan:

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

Apa yang dilakukan cuplikan ini: Membangun komponen yang bisa dijalankan untuk menyuntikkan memori yang telah diambil ke dalam prompt, kemudian membungkusnya dengan RunnableWithMessageHistory sehingga riwayat obrolan dan memori jangka panjang dapat bekerja bersama.

Pola ini mempertahankan sifat deterministik prompt Anda: setiap giliran secara eksplisit menyertakan memori yang diambil di bagian Memories.

Menjalankan skenario praktis lintas sesi

Skenario ini menunjukkan nilai lengkap memori jangka panjang:

  1. Di sesi A, pengguna berbagi preferensi.
  2. Di sesi B, aplikasi mengingat preferensi tersebut secara otomatis.
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.

Apa yang dilakukan cuplikan ini: Menginisialisasi preferensi pengguna di sesi A, memulai sesi B untuk pengguna yang sama, dan menunjukkan bahwa aplikasi dapat mengingat preferensi sebelumnya di berbagai sesi.

Contoh: Kueri memori langsung untuk skenario penggunaan non-obrolan

Gunakan pengalih ad-hoc saat Anda ingin pembacaan memori langsung di luar alur percakapan, misalnya dalam middleware personalisasi atau alat inspeksi profil.

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.

Apa yang dilakukan cuplikan ini: Menjalankan pencarian memori langsung untuk cakupan saat ini. Semua memori diambil (dibatasi oleh k) tetapi diurutkan menurut relevansi.

Gunakan pola ini saat Anda memerlukan pembacaan memori langsung untuk fitur seperti kartu profil, middleware personalisasi, atau perutean alur kerja.

Contoh: Menggunakan memori dalam grafik

LangGraph menggunakan pola konseptual yang sama:

  • Pertahankan user_id dalam keadaan stabil untuk memori jangka panjang.
  • Gunakan thread_id (atau setara) untuk konteks thread jangka pendek.
  • Ambil memori sebelum memanggil simpul model.

Jika Anda sudah memiliki StateGraph, lakukan injeksi pengambilan di simpul model Anda dan tambahkan teks memori ke dalam input model Anda. Strategi umum lainnya adalah menggunakan kait pra-model.

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

Apa yang dilakukan cuplikan ini: Menampilkan pola simpul LangGraph yang mengambil memori Foundry untuk giliran saat ini dan menyuntikkannya ke input model.

Untuk konsep memori LangGraph yang lebih luas, lihat:

Memahami batas pratinjau dan panduan operasional

Sebelum pindah ke produksi, validasi batasan ini:

  • Saat ini, memori sedang dalam tahap pratinjau dan perilakunya dapat berubah.
  • Memori memerlukan obrolan yang kompatibel dan penerapan penyematan.
  • Kuota berlaku per toko dan per lingkup, termasuk tingkat permintaan untuk pencarian dan pembaruan.

Rencanakan juga kontrol defensif untuk keracunan memori atau upaya injeksi prompt. Validasi input yang tidak tepercaya sebelum memengaruhi memori tersimpan.

Membersihkan sumber daya

Setelah menjalankan sampel, hapus cakupan agar data pengujian tidak bocor ke proses berikutnya.

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

Melacak aplikasi LangChain dan LangGraph

Konten terkait

  • Terakhir diperbarui pada