como usar o tairvector para busca aproximada de estruturas moleculares de compostos ou fármacos - Tair (Redis® OSS-Compatible): Central de Documentação do Alibaba Cloud

O TairVector é um mecanismo de vetores em memória que suporta atualizações de índice em tempo real e busca dos k-vizinhos mais próximos (KNN) com latência de milissegundos. Este tutorial demonstra como construir um sistema de busca por similaridade de estruturas moleculares no TairVector usando Python e RDKit — um fluxo de trabalho comum na descoberta de fármacos impulsionada por IA e na triagem de compostos.

Contexto

Triar grandes bibliotecas de compostos para encontrar moléculas estruturalmente semelhantes é uma tarefa fundamental na descoberta de fármacos. A abordagem típica converte cada estrutura molecular de sua representação SMILES em uma impressão digital vetorial de comprimento fixo, armazena essas impressões digitais em um índice de vetores e recupera as k estruturas mais semelhantes para uma molécula de consulta.

O TairVector armazena todos os dados na memória e permite atualizações de índice em tempo real, proporcionando menor latência de leitura e gravação em comparação a alternativas baseadas em disco. Principais características para este caso de uso:

Característica	Descrição
Armazenamento em memória	Mantém todo o índice de impressões digitais moleculares na RAM para acesso de baixa latência
Atualizações de índice em tempo real	Permite indexar novos compostos sem reconstruir o índice do zero
Busca KNN via `TVS.KNNSEARCH`	Recupera as k estruturas mais semelhantes em milissegundos
k configurável	Defina o número de resultados retornados para atender ao seu pipeline de triagem

Pré-requisitos

Antes de começar, verifique se você tem:

Uma instância Tair. Anote o endpoint e a senha correspondentes.
Python 3.8 ou superior
As seguintes dependências Python instaladas:
```
pip install numpy rdkit tair matplotlib
```

Como funciona

Tair Vector分子结构检索流程图..jpeg

O pipeline completo consiste em cinco etapas:

Baixe o conjunto de dados de estruturas moleculares no formato SMILES.conjunto de dados de estruturas moleculares
Conecte-se a uma instância Tair.
Crie um índice de vetores para armazenar as impressões digitais moleculares.
Converta cada string SMILES em um vetor de impressão digital Morgan de 512 dimensões e grave-o no índice.
Consulte o índice para obter as k estruturas mais semelhantes a uma molécula alvo.

O código de exemplo abaixo implementa todas as cinco etapas. Cada seção explica a função correspondente.

Etapa 1: Preparar o conjunto de dados

Baixe o conjunto de dados de amostra do PubChem. Ele contém 11.012 compostos no formato Simplified Molecular Input Line Entry System (SMILES), com duas colunas: fórmula química e ID exclusivo.

CCC1=CN=C2C(C(=O)N(C)C(=O)N2C)/C1=N/c1ccc(OC)cc1OC,168000001
CC(C)CN1C(=O)C2SCCC2N2C(=S)NNC12,168000002
CC1=C[NH+]=C2C(C(=O)N(C)C(=O)N2C)/C1=N/c1cccc(C(F)(F)F)c1,168000003
CC1=CN=C2C(C(=O)N(C)C(=O)N2C)/C1=N/c1cccc(C(F)(F)F)c1,168000004

Em ambiente de produção, carregue um conjunto de dados maior para testar o desempenho de recuperação em nível de milissegundos do TairVector em escala.

Se seus dados estiverem no formato Structure-Data File (SDF) provenientes do servidor FTP do PubChem, converta-os primeiro para SMILES usando o RDKit:

import sys
from rdkit import Chem

def converter(file_name):
    mols = [mol for mol in Chem.SDMolSupplier(file_name)]
    outname = file_name.split(".sdf")[0] + ".smi"
    out_file = open(outname, "w")
    for mol in mols:
        smi = Chem.MolToSmiles(mol)
        name = mol.GetProp("_Name")
        out_file.write("{},{}\n".format(smi, name))
    out_file.close()

if __name__ == "__main__":
    converter(sys.argv[1])

Etapa 2: Conectar-se a uma instância Tair

A função get_tair() estabelece uma conexão com sua instância Tair. Substitua os valores de espaço reservado pelo endpoint e pela senha reais.

Evite codificar credenciais diretamente no código. Armazene o endpoint e a senha em variáveis de ambiente e leia-os em tempo de execução usando os.environ.get() .

from tair import Tair

def get_tair() -> Tair:
    """
    Connect to the Tair instance.
    host: The endpoint of the Tair instance.
    port: The port number. Default is 6379.
    password: The password of the default account. To connect with a custom account, use 'username:password'.
    """
    tair: Tair = Tair(
        host="r-bp1mlxv3xzv6kf****pd.redis.rds.aliyuncs.com",
        port=6379,
        db=0,
        password="Da******3",
    )
    return tair

Etapa 3: Criar um índice de vetores

A função create_index() cria um índice de vetores chamado MOLSEARCH_TEST. Se o índice já existir, a criação será ignorada.

def create_index():
    """
    Create a vector index named MOLSEARCH_TEST with the following parameters:
    - Dimensions: 512 (matches the Morgan fingerprint bit length)
    - Distance metric: L2 (Euclidean distance)
    - Index algorithm: HNSW (Hierarchical Navigable Small World)
    """
    ret = tair.tvs_get_index(INDEX_NAME)
    if ret is None:
        tair.tvs_create_index(INDEX_NAME, 512, distance_type=DistanceMetric.L2, index_type="HNSW")
    print("create index done")

A dimensão do vetor está definida como 512 para corresponder à saída de smiles_to_vector() , que utiliza AllChem.GetMorganFingerprintAsBitVect(mol, radius=2, nBits=512*8) . A métrica de distância L2 mede a distância euclidiana entre os vetores de impressão digital; uma pontuação menor indica maior similaridade estrutural.

Etapa 4: Gravar dados de estrutura molecular

A função do_load() lê o conjunto de dados SMILES, converte cada entrada em um vetor de 512 dimensões usando a função smiles_to_vector() e grava o resultado no TairVector com TVS.HSET por meio de insert_data(). As gravações são agrupadas em lotes de 10 e enviadas simultaneamente usando ThreadPoolExecutor.

Cada entrada é armazenada no índice da seguinte forma:

Campo	Valor	Exemplo
Chave	ID exclusivo do composto	`168000001`
Vetor	Impressão digital Morgan de 512 dimensões	—
Atributo: `smiles`	String original da fórmula química	`CCC1=CN=C2...`

from concurrent.futures import ThreadPoolExecutor
from rdkit.Chem import AllChem
from rdkit import DataStructs, Chem
from rdkit import RDLogger
RDLogger.DisableLog('rdApp.*')

def do_load(file_path):
    num = 0
    lines = []
    with open(file_path, 'r') as f:
        for line in f:
            if line.find("smiles") >= 0:
                continue
            lines.append(line)
            if len(lines) >= 10:
                parallel_submit_lines(lines)
                num += len(lines)
                lines.clear()
                if num % 10000 == 0:
                    print("load num", num)
    if len(lines) > 0:
        parallel_submit_lines(lines)
    print("load done")

def parallel_submit_lines(lines):
    with ThreadPoolExecutor(len(lines)) as t:
        for line in lines:
            t.submit(handle_line, line=line)

def handle_line(line):
    if line.find("smiles") >= 0:
        return
    parts = line.strip().split(',')
    try:
        ids = parts[1]
        smiles = parts[0]
        vec = smiles_to_vector(smiles)
        insert_data(ids, smiles, vec)
    except Exception as result:
        print(result)

def smiles_to_vector(smiles):
    """Convert a SMILES string to a 512-dimensional Morgan fingerprint vector."""
    mols = Chem.MolFromSmiles(smiles)
    fp = AllChem.GetMorganFingerprintAsBitVect(mols, 2, 512 * 8)
    hex_fp = DataStructs.BitVectToFPSText(fp)
    vec = list(bytearray.fromhex(hex_fp))
    return vec

def insert_data(id, smiles, vector):
    """Write the vector and chemical formula to TairVector using TVS.HSET."""
    attr = {'smiles': smiles}
    tair.tvs_hset(INDEX_NAME, id, vector, **attr)

Etapa 5: Buscar estruturas moleculares semelhantes

A função do_search() recebe uma molécula de consulta no formato SMILES e um inteiro k, converte a consulta em um vetor de impressão digital e executa TVS.KNNSEARCH no índice MOLSEARCH_TEST. Em seguida, obtém a fórmula química de cada resultado usando TVS.HMGET.

def do_search(search_smiles, k):
    """
    Query the index for the k most similar molecular structures.
    Returns unique IDs, L2 distance scores, and chemical formulas.
    """
    vector = smiles_to_vector(search_smiles)
    result = tair.tvs_knnsearch(INDEX_NAME, k, vector)
    print("The 10 molecular structures most similar to the query target are as follows:")
    for key, value in result:
        similar_smiles = tair.tvs_hmget(INDEX_NAME, key, "smiles")
        print(key, value, similar_smiles)

Código de exemplo completo

Substitua D:\Test\Compound_168000001_168500000.smi na chamada do_load() pelo caminho real do arquivo do conjunto de dados baixado.

import os
import sys
from tair import Tair
from tair.tairvector import DistanceMetric
from rdkit.Chem import Draw, AllChem
from rdkit import DataStructs, Chem
from rdkit import RDLogger
from concurrent.futures import ThreadPoolExecutor
RDLogger.DisableLog('rdApp.*')

def get_tair() -> Tair:
    """
    Connect to the Tair instance.
    host: The endpoint of the Tair instance.
    port: The port number. Default is 6379.
    password: The password of the default account. To connect with a custom account, use 'username:password'.
    """
    tair: Tair = Tair(
        host="r-bp1mlxv3xzv6kf****pd.redis.rds.aliyuncs.com",
        port=6379,
        db=0,
        password="Da******3",
    )
    return tair

def create_index():
    """
    Create a vector index named MOLSEARCH_TEST:
    - Dimensions: 512
    - Distance metric: L2
    - Index algorithm: HNSW
    """
    ret = tair.tvs_get_index(INDEX_NAME)
    if ret is None:
        tair.tvs_create_index(INDEX_NAME, 512, distance_type=DistanceMetric.L2, index_type="HNSW")
    print("create index done")

def do_load(file_path):
    """
    Read the SMILES dataset, extract vector features, and write data to TairVector.
    Data is stored as: key=compound ID, vector=512-dim fingerprint, attribute smiles=chemical formula.
    """
    num = 0
    lines = []
    with open(file_path, 'r') as f:
        for line in f:
            if line.find("smiles") >= 0:
                continue
            lines.append(line)
            if len(lines) >= 10:
                parallel_submit_lines(lines)
                num += len(lines)
                lines.clear()
                if num % 10000 == 0:
                    print("load num", num)
    if len(lines) > 0:
        parallel_submit_lines(lines)
    print("load done")

def parallel_submit_lines(lines):
    with ThreadPoolExecutor(len(lines)) as t:
        for line in lines:
            t.submit(handle_line, line=line)

def handle_line(line):
    if line.find("smiles") >= 0:
        return
    parts = line.strip().split(',')
    try:
        ids = parts[1]
        smiles = parts[0]
        vec = smiles_to_vector(smiles)
        insert_data(ids, smiles, vec)
    except Exception as result:
        print(result)

def smiles_to_vector(smiles):
    """Convert a SMILES string to a 512-dimensional Morgan fingerprint vector."""
    mols = Chem.MolFromSmiles(smiles)
    fp = AllChem.GetMorganFingerprintAsBitVect(mols, 2, 512 * 8)
    hex_fp = DataStructs.BitVectToFPSText(fp)
    vec = list(bytearray.fromhex(hex_fp))
    return vec

def insert_data(id, smiles, vector):
    """Write the vector and chemical formula to TairVector using TVS.HSET."""
    attr = {'smiles': smiles}
    tair.tvs_hset(INDEX_NAME, id, vector, **attr)

def do_search(search_smiles, k):
    """
    Query the index for the k most similar molecular structures.
    Uses TVS.KNNSEARCH to find the nearest neighbors, then TVS.HMGET to retrieve their formulas.
    """
    vector = smiles_to_vector(search_smiles)
    result = tair.tvs_knnsearch(INDEX_NAME, k, vector)
    print("The 10 molecular structures most similar to the query target are as follows:")
    for key, value in result:
        similar_smiles = tair.tvs_hmget(INDEX_NAME, key, "smiles")
        print(key, value, similar_smiles)

if __name__ == "__main__":
    # Connect to Tair and create the molecular structure index.
    tair = get_tair()
    INDEX_NAME = "MOLSEARCH_TEST"
    create_index()
    # Load the sample dataset.
    do_load("D:\Test\Compound_168000001_168500000.smi")
    # Query the 10 structures most similar to the target compound.
    do_search("CCOC(=O)N1CCC(NC(=O)CN2CCN(c3cc(C)cc(C)c3)C(=O)C2=O)CC1", 10)

Resultados

Uma execução bem-sucedida produz uma saída semelhante à seguinte:

create index done
load num 10000
load done
The 10 molecular structures most similar to the query target are as follows:
b'168000009' 0.0 ['CCOC(=O)N1CCC(NC(=O)CN2CCN(c3cc(C)cc(C)c3)C(=O)C2=O)CC1']
b'168003114' 29534.0 ['Cc1cc(C)cc(N2CCN(CC(=O)NC3CCCC3)C(=O)C2=O)c1']
b'168000210' 60222.0 ['COc1ccc(N2CCN(CC(=O)Nc3cc(C)cc(C)c3)C(=O)C2=O)cc1OC']
b'168001000' 61123.0 ['COc1ccc(N2CCN(CC(=O)Nc3ccc(C)cc3)C(=O)C2=O)cc1OC']
b'168003038' 64524.0 ['CCN1CCN(c2cc(C)cc(C)c2)C(=O)C1=O']
b'168003095' 67591.0 ['O=C(CN1CCN(c2cccc(Cl)c2)C(=O)C1=O)NC1CCCC1']
b'168000396' 70376.0 ['COc1ccc(N2CCN(Cc3ccc(C)cc3)C(=O)C2=O)cc1OC']
b'168002227' 71121.0 ['CCOC(=O)CN1CCN(C2CC2)C(=O)C1=O']
b'168000441' 73197.0 ['Cc1cc(C)cc(NC(=O)CN2CCN(c3ccc(F)c(F)c3)C(=O)C2=O)c1']
b'168000561' 73269.0 ['Cc1cc(C)cc(N2CCN(CC(=O)Nc3ccc(C)cc3C)C(=O)C2=O)c1']

Cada linha contém:

Coluna	Descrição
Coluna 1	ID do composto
Coluna 2	Pontuação de distância L2 (valores menores indicam maior similaridade)
Coluna 3	Fórmula química

O resultado com pontuação 0.0 representa uma correspondência exata para a molécula consultada.

Para visualizar as estruturas recuperadas como imagens moleculares, use o código a seguir:

import numpy
from rdkit.Chem import Draw
from rdkit import Chem
import matplotlib.pyplot as plt

def to_images(data):
    imgs = []
    for smiles in data:
        mol = Chem.MolFromSmiles(smiles)
        img = Chem.Draw.MolToImage(mol, size=(500, 500))
        imgs.append(img)
        plt.imshow(img)
        plt.show()
    return imgs

if __name__ == "__main__":
    images = to_images(["CCOC(=O)N1CCC(NC(=O)CN2CCN(c3cc(C)cc(C)c3)C(=O)C2=O)CC1"])

相似分子结构检索..jpeg

Próximos passos

Carregue sua biblioteca completa de compostos e teste o desempenho de recuperação em escala.
Defina o parâmetro k em do_search() para controlar quantos candidatos o pipeline de triagem retorna.