Используйте память Foundry с LangChain и LangGraph

Используйте langchain-azure-ai и память Foundry для добавления долгосрочной памяти в приложения. В этой статье вы создаете цепочку на основе памяти, сохраняете настройки пользователя, вспоминаете их в новом сеансе и выполняете непосредственные запросы к памяти.

Этот шаблон работает как для приложений LangChain, так и для LangGraph. Основная идея заключается в сохранении краткосрочной истории чата в среде выполнения и использовании Foundry Memory в качестве долгосрочного хранилища для контекста на уровне пользователя.

Память Foundry фокусируется на долгосрочной памяти. Сохраняйте кратковременное пошаговое состояние в состоянии выполнения среды LangChain или LangGraph.

Необходимые условия

• Подписка Azure. Создайте его бесплатно.
• Проект Foundry.
• Развернутая модель чата Microsoft Foundry для извлечения информации из памяти.
  • В этом руководстве используется gpt-4.1.
• Развернутая модель чата и модель встраивания для хранилища памяти.
  • В этом руководстве используется text-embedding-3-large.
• Python 3.10 или более поздней версии.
• Azure CLI вошел в систему (az login), чтобы DefaultAzureCredential мог пройти проверку подлинности с помощью роли Azure AI Developer.

Настройка среды

Установите необходимые пакеты для этого руководства. Используйте langchain-azure-ai для интеграции LangChain и LangGraph, azure-ai-projects для управления хранилищем памяти и azure-identity для проверки подлинности.

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

Задайте переменные среды, которые мы используем в этом руководстве:

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

Общие сведения о модели памяти

Foundry Memory хранит и извлекает два долговременных типа памяти.

• Память профиля пользователя: стабильные факты и предпочтения пользователя, такие как предпочтительное имя или диетическое ограничение.
• Сводная память чата: дистиллированные сводки по предыдущим темам обсуждения.

Память использует идею "области действия" для разделения информации, чтобы ее можно было хранить и извлекать последовательно. Области похожи на идентификаторы или ключи для упорядочивания информации.

• Идентификаторы пользователей можно использовать в качестве стабильного удостоверения для долгосрочной памяти. Сохраняйте его в одном и том же виде на протяжении сеансов для одного и того же пользователя.
• Идентификаторы сеансов можно использовать для краткосрочной идентификации беседы. Изменяйте его для каждого сеанса чата.
• Идентификаторы ресурсов можно использовать в качестве стабильного идентификатора для долгосрочной памяти для нескольких пользователей.

Это разделение позволяет приложению запоминать предпочтения пользователей в сеансах, не смешивая несвязанные беседы.

Создание хранилища памяти

Перед началом работы необходимо создать хранилище памяти. Для этой операции используйте пакет SDK для проектов 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

Что делает этот фрагмент кода: Подключается к проекту Foundry, получает или создает хранилище памяти и включает профиль пользователя и извлечение сводки чата.

Использование памяти в LangGraph и LangChain

Память Foundry интегрируется в LangGraph и LangChain благодаря введению двух объектов:

• langchain_azure_ai.chat_message_history.AzureAIMemoryChatMessageHistory Класс создает журнал чата в памяти.
• Класс langchain_azure_ai.retrievers.AzureAIMemoryRetriever позволяет получить воспоминания из журнала сообщений чата.

Как правило, с ними можно использовать следующие практические стратегии извлечения:

• Получение памяти профиля пользователя в начале беседы для персонализации ответов.
• Получение сводки памяти чата на основе текущего шага для восстановления соответствующего контекста из предыдущих взаимодействий.

Пример: Добавление уровня памяти с поддержкой сеансов

В этом примере мы создаем в LangChain единый исполняемый модуль, который извлекает соответствующую долгосрочную память, внедряет её в запрос и запускает модель с объединенной короткосрочной историей чата и долгосрочной памятью.

Давайте посмотрим, как реализовать его:

Создание журнала сообщений чата

В этом примере используется стабильная user_id область памяти. ** Используйте 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)

Что делает этот фрагмент кода: Создает журнал на основе памяти и восстановитель для каждой (user_id, session_id) пары и кэширует их, чтобы состояние извлечения сохранялось в разных обращениях в одной сессии. В этом пошаговом руководстве update_delay=0 обеспечивает немедленную видимость обновлений памяти. В рабочей среде используйте задержку по умолчанию, если вам не потребуется мгновенное извлечение. session_histories используется для предотвращения постоянного пересоздания объектов.

Составление выполняемого модуля с использованием извлечения памяти

Давайте создадим исполняемый объект для реализации цикла:

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

Что делает этот фрагмент кода: Создает запускаемый элемент, который внедряет извлеченные воспоминания в запрос, а затем оборачивает его, чтобы журнал чата и долгосрочная память работали вместе.

Этот шаблон сохраняет детерминированный запрос: каждый поворот явно включает полученную память в разделе Memories.

Запуск практического межсеансового сценария

В этом сценарии показана полная ценность долгосрочной памяти:

1. В сеансе A пользователь делится предпочтениями.
2. В сеансе B приложение автоматически отзывает эти настройки.

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.

Что делает этот фрагмент кода: Устанавливает настройки пользователя в сеансе A, запускает сеанс B для того же пользователя и показывает, что приложение может восстановить предыдущие настройки между сеансами.

Пример. Запрос памяти непосредственно для вариантов использования, отличных от чата

Используйте временный поисковый механизм, если требуется для прямого чтения из памяти вне конвейера обработки диалога, например, в промежуточных слоях персонализации или инструментах анализа профиля.

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.

Что делает этот фрагмент кода: Выполняет прямой поиск памяти для текущей области. Все воспоминания извлекаются (ограничены) но отсортированы kпо релевантности.

Используйте этот шаблон, если требуется прямое чтение памяти для таких функций, как профильные карточки, промежуточный слой персонализации или маршрутизация рабочих процессов.

Пример. Использование памяти в графах

LangGraph использует тот же концептуальный шаблон:

• Сохраняйте user_id стабильным для долговременной памяти.
• Используйте thread_id (или эквивалент) для краткосрочного контекста потока.
• Получите данные памяти перед вызовом узла модели.

Если у вас уже есть StateGraph, интегрируйте функцию извлечения в узел модели и добавьте текст памяти к входным данным модели. Другая типичная стратегия заключается в использовании перехватчика предварительной модели.

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

Что делает этот фрагмент кода: Показывает шаблон узла LangGraph, который извлекает память Foundry для текущего цикла и внедряет её во вход модели.

Более широкие понятия памяти LangGraph см. в статье:

• Добавление памяти в LangGraph

Общие сведения об ограничениях предварительной версии и операционных рекомендациях

Перед переходом в рабочую среду проверьте следующие ограничения:

• Память находится в предварительной версии, и её поведение может изменяться.
• Для памяти требуется совместимый чат и внедрение развертываний.
• Квоты применяются для каждого хранилища и области применения, включая частоту запросов на поиск и обновление.

Кроме того, планируйте оборонительные элементы управления для отравлений памяти или попыток внедрения запроса. Проверьте ненадежные входные данные, прежде чем они влияют на хранимую память.

Очистка ресурсов

После выполнения тестов удалите область видимости, чтобы избежать утечки тестовых данных в будущие выполнения.

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

Приложения Trace LangChain и LangGraph

Связанное содержимое

• Используйте службу агента Foundry с LangGraph
• Начните с пакетов SDK Microsoft Foundry и точек доступа
• Используйте память вместе со службой агента Foundry
• Что такое память в Foundry Agent Service