Utilize 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 ao nível 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 da 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 incorporação para o armazenamento de memória.
  • Este tutorial utiliza text-embedding-3-large.
• Python 3.10 ou posterior.
• CLI do Azure iniciado (az login) de modo que DefaultAzureCredential possa autenticar-se com a função Azure AI Developer.

Configure 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 escopos são como identificadores ou chaves para organizar a informação.

• Pode usar IDs de utilizador como identidade estável para memória a longo prazo. Mantenha a mesma configuração entre sessões do mesmo utilizador.
• Podes usar os IDs de sessão como identidade de conversa a curto prazo. Altera-o para cada sessão de chat.
• Pode usar IDs de recursos como identificador estável para armazenamento 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 de projetos 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

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 por memória.
• A classe langchain_azure_ai.retrievers.AzureAIMemoryRetriever permite a recuperação de dados 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, a fim de 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 um histórico e um sistema de recuperação suportados por memória para cada (user_id, session_id) par e armazena-os em cache para que o estado de recuperação se mantenha ao longo das interações 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: Constrói um módulo executável que injeta memórias recuperadas no prompt e, em seguida, envolve-o com RunnableWithMessageHistory de modo a que o histórico de chat e a memória de longo prazo funcionem 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

Utilize um retriever ad-hoc quando necessitar de acessos diretos à memória fora do pipeline de comunicação, por exemplo, em software de personalização ou ferramentas de inspeção de perfis.

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 pesquisa direta de memória no âmbito 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:

• Mantém user_id estável para a memória a longo prazo.
• Use thread_id (ou equivalente) para contexto de tópico de curto prazo.
• Recupere a memória antes de chamar o nó modelo.

Se já tiver um StateGraph, inclua a recuperação no nó do seu modelo e adicione texto de memória à entrada de dados 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 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 versão de teste e o comportamento pode mudar.
• A memória requer implantações compatíveis de chat e embedding.
• As quotas aplicam-se por loja e por escopo, incluindo taxas de pesquisa e pedidos 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.

Liberar recursos

Após a execução das amostras, elimine o escopo para evitar que os dados de teste vazem para execuçõ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'.

Conteúdo relacionado

• Use o Serviço de Agente da Fundição com o LangGraph
• Comece a usar os SDKs e Endpoints Microsoft Foundry
• Usar memória com o Foundry Agent Service
• O que é a memória no Foundry Agent Service