Use a Memória Foundry com LangChain e LangGraph

Use langchain-azure-ai a Foundry Memory para adicionar memória de longo prazo às suas aplicações. Neste artigo, cria uma cadeia apoiada em memória, armazena as preferências do utilizador, recupera-as numa nova sessão e executa consultas diretas à memória.

Este padrão funciona tanto para aplicações LangChain como LangGraph. A ideia central é manter o histórico de chat de curto prazo no seu ambiente de execução e usar o Foundry Memory como o armazenamento de longo prazo para o contexto do utilizador.

A Foundry Memory foca-se na memória de longo prazo. Mantenha o estado de curto prazo passo a passo no estado de execução LangChain ou LangGraph.

Pré-requisitos

• Uma subscrição do Azure. Crie um gratuitamente.
• Um projeto Foundry.
• Um modelo de chat Microsoft Foundry implementado para recuperação de memória.
  • Este tutorial utiliza o "gpt-4.1".
• Um modelo de chat implementado e um modelo de embedding para o armazenamento de memória.
  • Este tutorial utiliza text-embedding-3-large.
• Python 3.10 ou posterior.
• Azure CLI iniciou sessão (az login) por isso DefaultAzureCredential pode autenticar com a função Azure AI Developer.

Configurar o seu ambiente

Instala os pacotes necessários para este tutorial. Use langchain-azure-ai para integração com LangChain e LangGraph, azure-ai-projects para gestão de memória e azure-identity para autenticação.

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

Defina as variáveis de ambiente que usamos neste tutorial:

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

Compreenda o modelo de memória

O Foundry Memory armazena e recupera dois tipos de memória de longo prazo:

• Memória do perfil do utilizador: factos e preferências estáveis do utilizador, como nomes preferidos ou restrições alimentares.
• Memória resumida do chat: resumos destilados de tópicos de discussões anteriores.

A memória usa a ideia de "escopo" para particionar a informação de modo a que possa ser armazenada e recuperada de forma consistente. Os âmbitos são como identificadores ou chaves para organizar a informação.

• Pode usar IDs de utilizador como identidade estável para a memória de longo prazo. Mantenha o mesmo ao longo das sessões para o mesmo utilizador.
• Podes usar os IDs de sessão como identidade de conversa a curto prazo. Altera por cada sessão de chat.
• Pode usar IDs de recursos como identificador estável para memória de longo prazo entre vários utilizadores.

Esta separação permite que a sua aplicação se lembre das preferências do utilizador entre sessões sem misturar conversas não relacionadas.

Criar o armazenamento de memória

Antes de começar, precisas de criar um armazenamento de memórias. Para esta operação, utilize o SDK azure-ai-projectsde projetos Microsoft Foundry.

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

O que este excerto faz: Liga-se ao seu projeto Foundry, obtém ou cria o armazenamento de memória e ativa a extração de resumos de perfil de utilizador e chat.

Utilização de memória no LangGraph e no LangChain

A Foundry Memory integra-se no LangGraph e no LangChain ao introduzir dois objetos:

• A classe langchain_azure_ai.chat_message_history.AzureAIMemoryChatMessageHistory cria um histórico de chat suportado pela memória.
• A classe langchain_azure_ai.retrievers.AzureAIMemoryRetriever permite recuperar memórias do histórico de mensagens do chat.

De um modo geral, pode usar as seguintes estratégias práticas de recuperação com eles:

• Recupere a memória do perfil do utilizador no início de uma conversa para personalizar as respostas.
• Recuperar a memória resumida do chat com base na interação atual para recuperar o contexto anterior relevante.

Exemplo: Adicionar uma camada de memória sensível à sessão

Neste exemplo, construímos um único executável no LangChain que recupera memória relevante de longo prazo, injeta-a no prompt e executa o modelo com histórico de chat de curto prazo e memória de longo prazo em conjunto.

Vamos ver como implementar isto:

Crie o histórico das mensagens do chat

Este exemplo utiliza um stable user_id como escopo de memória. Use session_id para contexto de conversa por sessão.

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)

O que este excerto faz: Cria para cada (user_id, session_id) par um histórico e um mecanismo de recuperação suportados por memória e armazena-os em cache para que o estado de recuperação sobreviva ao longo dos turnos na mesma sessão. Para este guia, update_delay=0 torna as atualizações de memória imediatamente visíveis. Em produção, usa o delay padrão a menos que precises especificamente de extração instantânea. session_histories é usado para evitar ter de recriar constantemente os objetos.

Compor o executável com recuperação de memória

Vamos criar um executável para implementar o ciclo:

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

O que este excerto faz: Cria um runnable que injeta memórias recuperadas no prompt e depois envolve-o com RunnableWithMessageHistory, permitindo que o histórico do chat e a memória de longo prazo trabalhem em conjunto.

Este padrão mantém o seu prompt determinista: cada turno inclui explicitamente a memória recuperada na secção Memories.

Executar um cenário prático entre sessões

Este cenário mostra o valor total da memória de longo prazo:

1. Na sessão A, o utilizador partilha preferências.
2. Na sessão B, a aplicação recupera automaticamente essas preferências.

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.

O que este excerto faz: Origina as preferências do utilizador na sessão A, inicia a sessão B para o mesmo utilizador e mostra que a aplicação consegue recordar preferências anteriores entre sessões.

Exemplo: consultar a memória diretamente para casos de uso que não sejam de chat

Use um extrator ad-hoc quando precisar de leituras de memória diretas fora do fluxo de conversação, por exemplo, em middleware de personalização ou ferramentas de inspeção de perfil.

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.

O que este excerto faz: Executa uma busca direta na memória para o escopo atual. Todas as memórias são recuperadas (limitadas por k) mas ordenadas por relevância.

Use este padrão quando precisar de leituras diretas de memória para funcionalidades como cartões de perfil, middleware de personalização ou encaminhamento de fluxos de trabalho.

Exemplo: Usar memória em grafos

O LangGraph utiliza o mesmo padrão conceptual:

• Mantenha user_id estável para a memória de longo prazo.
• Use thread_id (ou equivalente) para contexto de thread de curta duração.
• Recuperar memória antes de chamar o nó modelo.

Se já tiver um StateGraph, injete a recuperação no nó do seu modelo e adicione texto de memória à entrada do modelo. Outra estratégia típica é usar um gancho pré-modelo.

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

O que este excerto faz: Mostra um padrão de nó LangGraph que recupera a memória do Foundry para o turno atual e a injeta na entrada do modelo.

Para conceitos mais amplos de memória LangGraph, veja:

• Adicionar memória no LangGraph

Compreender os limites de pré-visualização e as orientações operacionais

Antes de passar para produção, valide estas restrições:

• A memória está em modo de pré-visualização, e seu comportamento pode mudar.
• A memória requer implantações compatíveis de chat e embedding.
• As quotas aplicam-se por loja e por âmbito, incluindo taxas de consultas e solicitações de atualização.

Também planeie controlos defensivos para envenenamento de memória ou tentativas de injeção rápida. Valide entradas não confiáveis antes que afetem a memória armazenada.

Limpeza de recursos

Após a execução das amostras, elimine o âmbito para evitar que os dados de teste vazem para iterações futuras.

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 e aplicações LangGraph

Conteúdo relacionado

• Utilize o Foundry Agent Service com o LangGraph
• Comece com os SDKs e Endpoints do Microsoft Foundry
• Usar memória com o Foundry Agent Service
• O que é a memória no Foundry Agent Service?