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

Для этого вспомним, что мы можем подсчитать целую и остаточную части при делении на 10 (или другое число, смотря в какой системе счисления работаем).

Допустим, имеем задачу:

"Дано 32-битное знаковое целое число x. Верните x с перевернутыми цифрами. Если переворачивание x приводит к выходу значения за пределы диапазона 32-битных знаковых целых чисел [-2**31, 2**31 - 1], верните 0"

Ключевая часть алгоритма к решению может выглядеть так - итеративно, пока целая часть от деления на 10 не равна 0 берем дробную и добавляем к результату, умноженному на 10:

def f(x):

    num = 2**31
    min_x = -num
    max_x = num - 1

    if x<0:
        sign = -1
        x = -x
    else:
        sign = 1

    res = 0

    while x!=0:
        ost = x%10
        x = x//10
        res = res*10 + ost

        if (res<min_x) or (res>max_x):
            return 0

    return res*sign

f(123)

Чтобы не мучиться с остатком, посчитаем знак и будем работать с положительным x. Напомню, что остаток в Python имеет знак делителя и работает правило:

a % b = a - b * floor(a / b),

floor(a / b) — округление вниз до ближайшего целого.

7%3 = 7 - 3*2=1

7%-3 = 7 + 3(-3) = -2

-7%-3 = -7 + 3*(2) = -1

-7%3 = -7 - 3*(-3)=2

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

x = 123
s = ''
while x!=0:
    ost = x%2
    s = f'{ost}{s}'
    x  = x//2

и обратно:

res = 0
s_len = len(s)
for i, num in enumerate(s):
    res = res + int(num)*2**(s_len-1-i)
res

В современных rag системах центральным инструментом являются ретриверы - объекты, которые отвечают за поиск близкой к запросу информации (контекста). Одним из них является BM25Retriever, основанный на частоте встречаемости. В отличие от аналогов, использующих векторные представления, он полагается на точные совпадение единиц, на которые разбит текст (токенов).

Для демонстрационных целей возмем набор текстов с описанием компьютерных угроз с сайта MITRE ATT&CK (тут и тут):

import numpy as np

train_sents = ["Contagious Interview has utilized open-source indicator of compromise repositories to determine their exposure to include VirusTotal, and MalTrail",
"Kimsuky has used LLMs to identify think tanks, government organizations, etc. that have information",
'''Sandworm Team researched Ukraine's unique legal entity identifier (called an "EDRPOU" number), including running queries on the EDRPOU website, in preparation for the NotPetya attack. Sandworm Team has also researched third-party websites to help it craft credible spearphishing emails''',
"During the 2015 Ukraine Electric Power Attack, Sandworm Team moved their tools laterally within the corporate network and between the ICS and corporate network",
"During the 2022 Ukraine Electric Power Attack, Sandworm Team used a Group Policy Object (GPO) to copy CaddyWiper's executable msserver.exe from a staging server to a local hard drive before deployment"]


query_sent="Mustang Panda has used open-source research to identify information about victims to use in targeting to include creating weaponized phishing lures and attachments"

BM25Retriever

Для корректной работы BM25Retriever важен способ разбиения текста на единицы, для этого используется параметр preprocess_func. Зададим функцию, осуществляющую деление по словам и их стемминг:

import nltk
from nltk.stem import SnowballStemmer
nltk.download('punkt_tab')

def preprocess_func(text):
    text = text.lower()
    words_l = nltk.tokenize.word_tokenize(text)
    stemmer = SnowballStemmer('english')
    words_l = [stemmer.stem(w) for w in words_l]
    return words_l

Верхнеуровнево для работы надо уметь создать BM25Retriever, используя, например, метод from_texts и найти ближайшие тексты с invoke:

from langchain_community.retrievers import BM25Retriever

# Возвращается объект pydantic BaseModel (в __init__ Serializable)
bm_retriever = BM25Retriever.from_texts(train_sents, k=2, preprocess_func=preprocess_func,
                                        metadatas=[{'len':len(it)} for it in train_sents],
                                        ids=range(len(train_sents)))

bm_retriever.invoke(query_sent)

from_texts

from_texts получает аргументы:

• texts - список текстов, формирующих базу для поиска;
• k - количество ближайших текстов в ответ на запрос поиска;
• preprocess_func - функция для разбиения текстов на токены;
• metadatas - список словарей с метаданными для каждого текста в texts;
• ids - список идентификаторов для каждого текста в texts.

Следует отметить, что k можно поменять (например, когда вы загружаете настроенный дамп retriever-а) так: bm_retriever.k = 1.

from_texts выполняет следующий код:

• texts_processed = [preprocess_func(t) for t in texts]
• vectorizer = BM25Okapi(texts_processed...)

invoke

В invoke выполняются:

• invoke из BaseRetriever вызывает _get_relevant_documents;
• _get_relevant_documents осуществляет препроцессинг и возврат наиболее подходящих через get_top_n:
• processed_query = self.preprocess_func(query)
• self.vectorizer.get_top_n(processed_query, self.docs, n=self.k)

Подытоживая, from_texts инициирует препроцессинг текстов и создает класс BM25Okapi, а invoke - препроцессинг запроса и вызывает метод get_top_n объекта класса BM25Okapi. Теперь разберемся с BM25Okapi и его особенностями.

okapi

• в конструкторе подсчитываются:
• словарь встречаемости слов в документах (doc_freqs)
• длины каждого документа в токенах (doc_len) и средняя длина документа в токенах (avgdl)
• idf токенов (idf)
• в get_top_n:
• каждый документ (объект с текстом и метаданными) получает скор близости к запросу get scores in BM25Okapi
• документы сортируются по убыванию скора и выбирается заданное число
• соответствующий код:
• scores = self.get_scores(query)
• top_n = np.argsort(scores)[::-1][:n]
• [documents[i] for i in top_n]

Создадим вручную объект класса BM25Okapi и набор документов (без метаданных для простоты):

from rank_bm25 import BM25Okapi
from langchain_core.documents import Document


texts_processed = [preprocess_func(t) for t in train_sents]
vectorizer = BM25Okapi(texts_processed)

docs = [Document(page_content=t) for t in train_sents]

processed_query = preprocess_func(query_sent)

свойства

• idf - содержит idf токенов
• doc_freqs - содержит список словарей с частотами токенов для каждого документа
• get_scores - возвращает список скоров каждого документа относительно query
• get_top_n - возвращает топ k самых близких документов
• doc_len - содержит список количества токенов в каждом документе
• avgdl - содержит среднее количество токенов в документах

формула скора

Посчитаем вручную скор для нулевого документа, который возвращает get_scores:

vectorizer.get_scores(processed_query)

idf для каждого токена считается в BM25Okapi конструкторе по следующей формуле:

Добавление константы (0.5) не дает знаменателю или числителю стать равным нулю и сглаживает рост idf для слов, которые встречаются только в единичных документах.

Посчитаем idf для токена "open-sourc", который встречается в 1 из 5 текстов:

idf = np.log((5-1+0.5)/(1+0.5))
vectorizer.idf['open-sourc'], idf

tf для "open-sourc" в нулевом документе получим так:

N = 0
tf = vectorizer.doc_freqs[N]['open-sourc']
tf

Для извлечения скора документа надо итерировать по всем токенам query и посчитать сумму для каждого по формуле:

Из формулы следует, что для документов с длиной меньше средней слагаемое будет увеличено (короткие получают бонус), а для больше средней - уменьшено (штраф за размытость). Посчитаем добавку для токена "open-sourc":

idf*(tf * (1.5 + 1) /(tf + 1.5 * (1 - 0.75 + 0.75 * vectorizer.doc_len[N] / vectorizer.avgdl)))

А теперь выведем скор для всего нулевого документа:

score = 0
# для нулевого документа
for q in processed_query:
    q_freq = vectorizer.doc_freqs[N].get(q,0)
    adding = vectorizer.idf.get(q, 0) * (q_freq * (1.5 + 1) /
                                        (q_freq + 1.5 * (1 - 0.75 + 0.75 * vectorizer.doc_len[N] / vectorizer.avgdl)))
    if q=='open-sourc':
      print(f'принт добавка от "open-sourc" - {adding}')
    score += adding
score

По скору наш (нулевой) документ второй, в такой очередности он и выводится в get_top_n:

k = 2

vectorizer.get_top_n(processed_query, docs, n=k)

Логируй так, будто завтра сервер упадет, и только твои записи спасут мир.

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

Нашими строительными блоками будут составляющие модуля logging:

• функция getLogger для получения логгера с заданным именем;
• классы handler-ов: FileHandler для логирования в файл, StreamHandler для вывода в консоль, которые добавляются в полученный на предыдущем этапе логгер;
• метод логгера setLevel для задания уровня вывода сообщений;
• методы логгера info, debug, error, warning для вывода сообщений с заданным уровнем.

Про уровни вывода сообщений и методы я рассказывал ранее. Создадим тестовый logger:

import logging

file_handler = logging.FileHandler('journal.log')
console_handler = logging.StreamHandler()

logger = logging.getLogger('test_logger')
logger.setLevel(logging.INFO)

logger.addHandler(file_handler)
logger.addHandler(console_handler)

logger.info('test message 1')

!cat journal.log

propagate

Можно заметить 2 сообщения в консоли и одно - в журнале. Почему в консоль отправлен лишний вывод?

Поведение по умолчанию предполагает передачу сообщений вверх по иерархии к root-логгеру, который является предком всех logger-ов и обрабатывает сообщение путем вывода на консоль. Это регулируется атрибутом propagate (который по умолчанию равен True). Соответственно, чтобы остановить передачу "наверх" надо задать: logger.propagate = False:

logger.propagate = False
logger.info('test message 2')

разные хэндлеры

Для вывода списка обработчиков, которые отвечают за отправку сообщений в канал, можно обратиться к свойству logger.handlers:

logger.handlers

Если попытаться добавить одинаковые хэндлеры, то список не изменится:

logger.addHandler(file_handler)
logger.addHandler(console_handler)
logger.handlers

Однако при создании новых объектов (и с другими id, соответственно) система будет считать их другими и произойдет дублирование вывода, поэтому при двухкратном обращении к одному логгеру в нашем классе надо будет предусмотреть такое поведение:

file_handler = logging.FileHandler('journal.log')
console_handler = logging.StreamHandler()

logger.addHandler(file_handler)
logger.addHandler(console_handler)
logger.handlers

закрытие хэндлеров

По завершении работы с логгером надо удалить из него привязку к обработчикам, что можно сделать, очистив соответствующий список. Однако следует помнить о коварной ошибке при удалении handler-ов из того же списка, по которому происходит итерация:

for handler in logger.handlers:
    logger.removeHandler(handler)

logger.handlers

А так (logger.handlers[:]) создаем копию списка для итерации, поэтому результат будет пустым, и новые сообщения при обращении к логгеру не появляются:

for handler in logger.handlers[:]:
    logger.removeHandler(handler)
logger.handlers

Для корректного освобождения ресурсов (закрытия файла, сброса буферов) также рекомендуется вызывать метод close обработчиков:

file_handler.stream.closed

file_handler.close()
console_handler.close()

класс логгера

На основании описанного выше можно создать класс логгера, который одновременно выводит и в журнал, и на консоль. Также добавим удобный формат сообщений (подробнее было тут).

import logging

LOG_FN = 'journal.log'

class ActivityLogger():

    def __init__(self, level=logging.INFO, encoding='utf-8'):

        self.level = level
        self.file_handler = logging.FileHandler(LOG_FN, encoding=encoding)
        self.console_handler = logging.StreamHandler()

    def get_logger(self, name):

        file_formatter = logging.Formatter(f'%(asctime)s \nMODULE:{name}, LEVEL:%(levelname)s, LINE:%(lineno)s, MSG:%(message)s\n')
        cons_formatter = logging.Formatter(f'%(asctime)s \n%(lineno)s %(message)s\n\n')

        # Применяем форматтер к обоим Handler
        self.file_handler.setFormatter(file_formatter)
        self.console_handler.setFormatter(cons_formatter)

        # Создаем объект logger и применяем к нему оба Handler
        logger = logging.getLogger(name)
        logger.setLevel(self.level)

        if len(logger.handlers)==0:
            logger.addHandler(self.file_handler)
            logger.addHandler(self.console_handler)

        # Disable propagation of messages to the root logger
        logger.propagate = False

        return logger

    def close_logger(self, logger):

        # import pdb;pdb.set_trace()
        for handler in logger.handlers[:]:
            logger.removeHandler(handler)

        self.file_handler.close()
        self.console_handler.close()

Для демонстрационных целей имитируем наличие проекта и класс logger-а в отдельном модуле:

скрипты передают свои имена в журнал

Теперь создадим 2 одинаковых скрипта, использующих логгер:

t = r'''
from constants import ActivityLogger

try:

    logger = ActivityLogger().get_logger(__file__)
    logger.info(f'тестовое сообщение')

    ActivityLogger().close_logger(logger)

except Exception:
    logger.exception('ВОЗНИКЛО ИСКЛЮЧЕНИЕ!!!')
    ActivityLogger().close_logger(logger)

'''

with open('project/script1.py', 'wt') as f:
    f.write(t)

with open('project/script2.py', 'wt') as f:
    f.write(t)

Запустим оба скрипта и увидим, что сообщения корректно выводятся и в журнал и на консоль с указанием файла-источника, что удобно для отладки:

нет дублирования при многократном обращении к логгеру

Так как мы предусмотрели проверку наличия обработчиков (в блоке - if len...), при обращении к одному и тому же логгеру дважды сообщения не дублируются:

from project.constants import ActivityLogger
import logging

logger1 = ActivityLogger(level = logging.ERROR).get_logger('note1')
logger1.error("тестовое сообщение note 1")

print(logging.getLogger('note1').handlers)

logger2 = ActivityLogger().get_logger('note1')

logger2.error("тестовое сообщение note 2")
print(logger2.handlers)

Следует отметить, что вместо проверки наличия в логгере обработчиков, можно было бы задать их на уровне переменных класса (а не экземпляра), тогда они бы при повторном добавлении id не меняли и, соответственно, каналы бы не дублировались:

import logging

LOG_FN = 'journal.log'

class ActivityLogger():

    file_handler = logging.FileHandler(LOG_FN, encoding='utf-8')
    console_handler = logging.StreamHandler()

    def __init__(self, level=logging.INFO):
        self.level = level

    def get_logger(self, name):

        file_formatter = logging.Formatter(f'%(asctime)s \nMODULE:{name}, LEVEL:%(levelname)s, LINE:%(lineno)s, MSG:%(message)s\n')
        cons_formatter = logging.Formatter(f'%(asctime)s \n%(lineno)s %(message)s\n\n')

        # Применяем форматтер к обоим Handler
        self.file_handler.setFormatter(file_formatter)
        self.console_handler.setFormatter(cons_formatter)

        # Создаем объект logger и применяем к нему оба Handler
        logger = logging.getLogger(name)
        logger.setLevel(logging.INFO)

        logger.addHandler(self.file_handler)
        logger.addHandler(self.console_handler)

        # Disable propagation of messages to the root logger
        logger.propagate = False

        return logger

    def close_logger(self, logger):

        for handler in logger.handlers[:]:
            logger.removeHandler(handler)

        self.file_handler.close()
        self.console_handler.close()

параллельная работа с несколькими логгерами

Если мы работаем с двумя логгерами, то закрытие одного не влияет на работу другого:

from project.constants import ActivityLogger

logger1 = ActivityLogger().get_logger('note1')
logger1.info("тестовое сообщение note 1")

logger2 = ActivityLogger().get_logger('note2')
logger2.info("тестовое сообщение note 2")

"Путь к успеху начинается с первого шага". Даже для опытных пользователей процесс настройки окружения с PyTorch и CUDA может превратиться в настоящую головную боль, если нет четкого алгоритма. Поэтому описание порядка установки станет в центре внимания в данном материале .

Первый этап - сбор информации о видеокарте, поддерживаемой версии CUDA и ее установка (все этапы тестировал на Windows).

1) Инфо о видеокарте можно получить через браузер Chrome:

chrome://gpu/

2) Далее гуглим, какая версия CUDA поддерживается вашей видеокартой, и скачиваем с официального сайта (в моем случае - CUDA 11.8 взял тут) и устанавливаем. О подробностях инсталляции и CUDA можно почитать на том же ресурсе (в целом процесс не отличается от установки обычной программы).

3) Если не прописана переменная среды CUDA_PATH, то создаем ее с указанием на папку с CUDA:

4) Проверка

инфо о CUDA:

 nvcc -V

+ мониторинг использования ГПУ:

nvidia-smi

5) Далее создаем виртуальное окружение (я делаю это с conda):

conda create -y --prefix D:/dev/envs/pytorch_env python=3.10

6) На сайте PyTorch берем команды инсталляции пакета и исполняем их:

для самой свежей версии

для более старых + установка с conda

В моем случае сработали команды:

для cuda 11.8:

conda install pytorch==2.3.0 torchvision==0.18.0 torchaudio==2.3.0 pytorch-cuda=11.8 -c pytorch -c nvidia

для cuda 10.1:

pip install torch==1.8.1+cu101 torchvision==0.9.1+cu101torchaudio==0.8.1 -f https://download.pytorch.org/whl/torch_stable.html

conda install pytorch torchvision torchaudio cudatoolkit=10.1 -c pytorch

...

7) Проверяем доступность CUDA:

python -c "import torch;print(torch.cuda.is_available())"

8) Напоминаю команды создания kernel-а из действующего окружения для jupyter (подробнее писал тут):

pip install ipykernel

python -m ipykernel install --user --name имя_среды 

Вот как выглядит результат nvidia-smi во время работы скрипта, использующего ГПУ:

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

Начнем с описания самых простых и полезных функций модуля.

getmodule и getfile

getmodule и getfile позволяют получить имя модуля и файла, в которых определен импортированный объект:

import inspect
from fastai.vision.all import *

inspect.getmodule(L)

inspect.getfile(L)

function signature

Для получения описания параметров и значений по умолчанию исследуемой функции обратитесь к signature:

def f(a, b:list=[2]):
  pass

sig = inspect.signature(f)

for name, param in sig.parameters.items():
    print(f"Parameter: {name}")
    print(f"  Default: {param.default}")
    print(f"  Annotation: {param.annotation}")
    print(f"  Kind: {param.kind}")
    print()

frame

Пожалуй, самым полезным объектом модуля является фрейм/блок. Функция currentframe возвращает ссылку на текущий блок кода, у которого есть следующие важные атрибуты:

• f_locals - словарь локальных переменных
• f_globals - словарь глобальных переменных
• f_back - позволяет обратиться к фрейму на уровень выше
• f_code - объект кода, привязанный к фрейму, как правило, описывает функцию, в которой происходит выполнение блока. В свою очередь, имеет атрибут co_varnames, содержащий имена аргументов и локальных переменных;

Ниже демонстрируются эти свойства фреймов:

def f2(a=1):
  frame = inspect.currentframe()
  print(f'f2 local vars - {(frame.f_locals.items())}')
  print(f'f2 func vars and locals - {frame.f_code.co_varnames}')
  print(f'f1 func vars and locals - {frame.f_back.f_code.co_varnames}')
  print(f'f0 func vars and locals - {frame.f_back.f_back.f_code.co_varnames}')

def f1(b):
  f2()

def f0(c=2):
  f1(2)

f0()

store_attr

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

self.a=a

...

Этот шаблонный код, пожалуй, писал каждый питонист хотя бы раз.

class C():

    def __init__(self, a, b):
        frame = inspect.currentframe()
        for k, v in frame.f_locals.items():
          setattr(self, k, v)

item = C(a=3, b=[1, 3])
item.a, item.b

Если делать то же через функцию, то понадобится обратиться к вышестоящему фрейму:

def copy_args():

    frame = inspect.currentframe().f_back
    code = frame.f_code
    args = code.co_varnames

    self = frame.f_locals[args[0]]
    for k, v in frame.f_locals.items():
      setattr(self, k, v)

class C():
    def __init__(self, a, b):
        copy_args()

item = C(a=3, b=[1, 3])
item.a, item.b

Схожий функционал заложен в функции store_attr из модуля fastcore.basics:

from fastcore.basics import store_attr

class C():
    def __init__(self, a, b):
        store_attr()

item = C(a=3, b=[1, 3])
item.a, item.b

Можно часть параметров сохранить через store_attr,перечислив их в скобках через запятую, а часть определить самим:

from fastcore.basics import store_attr

class C():
    def __init__(self, a, b, c):
        store_attr('a, c')
        self.b = a+c

item = C(a=3, b=3, c=5)
item.a, item.b, item.c

stack

Еще посредством функции stack модуля inspect из произвольной строки можно получить информацию о стеке вызовов по аналогии с тем, который выводится при ошибке. Пусть у нас есть модуль:

После вызова f0 мы получим информацию о номере строки текущего фрейма (lineno), функции (function) и файле, в котором она находится (filename). Кроме того, ниже из стека выводится информация о служебных инструментах, которые используются collab-ом:

import sys
sys.path.append('/content/drive/MyDrive/Colab Notebooks')
from my_funcs.module import f0

f0()

"Конфиденциальность — это не только право, это основа свободы." В эпоху цифровых технологий, когда данные становятся всё более ценным ресурсом, защита личной информации приобретает критическое значение. В этой статье я расскажу, как работать с конфиденциальными реквизитами в Python проекте.

load_dotenv

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

Отмечу, что этот способ подойдет, если злоумышленник не имеет доступ к локальным файлам, так как пароль хранится в открытом виде. Продемонстрируем работу на практике. Сначала установим модуль python-dotenv и создадим файл:

!pip install python-dotenv -q

!mkdir pers
!printf ''' USER="user"\n\
PSWD="1pswd1" ''' > pers/.env

!cat .env

Теперь вызовем load_dotenv с указанием пути к файлу:

from dotenv import load_dotenv

load_dotenv('pers/.env')

После этого переменные из файла доступны через окружение:

import os
os.environ['USER'], os.environ['PSWD']

Если все равно страшно хранить реквизиты в открытом файле, можно немного усложнить и, например, хранить часть пароля в файле, а часть добавлять из кода (но это все равно не безопасное решение). В функции ниже прописано такое поведение - при отсутствии создается файл с реквизитами (лучше, чтобы каждый раз при старте сессии, после удалять), при этом первый и последний символы, пароля не хранятся в файле, а добавляются в коде:

from getpass import getpass

def get_creds():

    if not os.path.exists('pers/.env'):
        cred = getpass()
        with open('pers/.env', 'wt') as f_wr:
            f_wr.write(f'PSWD={cred[1:-1]}')

    load_dotenv('pers/.env')
    PSWD = '1'+os.environ['PSWD']+'1'

    return PSWD return PSWD

Удалим файл, старое значением переменной окружения и вызовем функцию:

!rm pers/.env

del os.environ['PSWD']

USER = 'user'
PSWD = get_creds()
PSWD

Пароль правильный и он не хранится в файле полностью:

!cat pers/.env

Если вы делаете проект, который будет запускать лицо с другими реквизитами (например, заказчик), удобно функцию get_creds не показывать, а объявить в отдельно модуле, который не попадет в репозиторий. При этом в отдельном общедоступном py файле прописать переменные USER = 'user', PSWD = get_creds(). Во время передачи проекта указать, что следует задать в переменной PSWD пароль для запуска или для большей безопасности прописать там getpass.

keyring

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

pip install keyring keyrings.alt pycryptodome -q

import keyring
from keyrings.alt.file import EncryptedKeyring

Зададим тип хранилища:

keyring.set_keyring(EncryptedKeyring())

Для записи или обновления пароля можно использовать функцию set_password модуля keyring. Первый параметр - имя сервиса (ключа), к которому привязан пароль:

keyring.set_password('main', 'user', '1pswd1')

Для получения пароля необходимо указать имя сервиса и логин, а функция при первом ее вызове потребует ввод пароля хранилища, который задавался в set_password:

keyring.get_password('main', 'user')

Для получения пути к файлу с зашифрованным паролем можно вызвать атрибут file_path класса EncryptedKeyring:

EncryptedKeyring().file_path

!cat /root/.local/share/python_keyring/crypted_pass.cfg

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

1. keyring.get_keyring() в ячейке
2. keyring diagnose в терминале

«Ошибки — это наука, помогающая нам двигаться вперёд», — говорил Уильям Ченнинг. Визуализация - отличный инструмент, который помогает анализировать данные и выявлять закономерности.

Рассмотрим удобный способ отображения в Python одной из метрик классификации под названием confusion matrix (на русский переводят по-разному - матрица ошибок, неточностей, расхождений или несоответствий).

Сначала загрузим демонстрационный датасет.

import numpy as np
import pandas as pd

from sklearn.datasets import load_iris
from sklearn.linear_model import LogisticRegression

df = load_iris(as_frame=True)['frame']
d = {k:v for k, v in enumerate(load_iris().target_names)}
df['target'] = df['target'].map(d)
df = df.sample(frac=1, random_state=0)

df.head()

Имитируем обучение модели и предсказание, не проводя разбиения на выборки, так как наша цель - показать возможности для визуализации:

model = LogisticRegression(random_state=0, max_iter=1000).fit(df.drop(columns='target'), df.target)

y_p = model.predict(df.drop(columns='target'))

Теперь получим значения ошибок, воспользовавшись функцией confusion_matrix из модуля sklearn.metrics:

from sklearn.metrics import confusion_matrix

labels = np.sort(df['target'].unique())
cm = confusion_matrix(y_true = df['target'], y_pred = y_p,
                      labels=labels,
                      normalize=None)
cm

В строке i и колонке j матрицы располагаются значения, соответствующие количеству объектов класса i, которые предсказаны как j. Значения классов берутся, как метки встретившиеся хотя бы раз в y_true или y_pred упорядоченные по возрастанию. Я намеренно передаю метки явно в параметре labels для демонстрации этого поведения (этим же параметром, можно поменять порядок вывода или указать подмножество/список меток для вывода).

matshow

Визуализацию начнем с низкоуровневых способов matplotlib закончим более быстрыми. Первым кандидатом будет matshow модуля matplotlib.pyplot, которая предоставляет полотно, а элементы потребуется добавлять самостоятельно. Ниже показано, как задать размеры картинки, метки классов, а также дополнить ячейки значениями матрицы через текстовые элементы (большинство настроек разбирались ранее здесь издесь):

import matplotlib.pyplot as plt
fig = plt.figure(figsize=(5, 5))

plt.matshow(cm, alpha=1, cmap='coolwarm', fignum=fig.number)

plt.xticks(range(len(labels)), labels, rotation='vertical')
plt.yticks(range(len(labels)), labels)
plt.gca().xaxis.set_ticks_position('bottom')

for i in range(cm.shape[0]):
    for j in range(cm.shape[1]):
        plt.gca().text(x=j, y=i, s=cm[i][j], ha='center', va='center')

plt.xlabel('Predicted label')
plt.ylabel('True label')

imshow

Функция imshow работает аналогично, только для управления размером не надо явно передавать номер фигуры (как делали выше используя fignum,так как по умолчанию matshow сама создает фигуру):

fig = plt.figure(figsize=(5, 5))

plt.imshow(cm, cmap='viridis')

plt.xticks(range(len(labels)), labels, rotation='vertical')
plt.yticks(range(len(labels)), labels)
plt.gca().xaxis.set_ticks_position('bottom')

for i in range(cm.shape[0]):
    for j in range(cm.shape[1]):
        plt.gca().text(x=j, y=i, s=cm[i][j], ha='center', va='center', color="white")

plt.xlabel('Predicted label')
plt.ylabel('True label')

heatmap

Для отображения с помощью функции heatmap из библиотеки seaborn надо будет внести минимальные правки (закомментировал строки, которые помогут добавить подписи):

import seaborn as sns
plt.figure(figsize=(5,5))

# fmt='.1f'
sns.heatmap(pd.DataFrame(cm, index=labels, columns=labels), annot=True,
            fmt='d', cmap='inferno', cbar=False)

# plt.xticks(range(len(labels)), list(labels), rotation='vertical')
# plt.yticks(range(len(labels)), labels)

# plt.xlabel('Predicted label')
# plt.ylabel('True label')

ConfusionMatrixDisplay

Об удобстве вывода задумались и разработчики scikit-learn, написав класс ConfusionMatrixDisplay. С помощью его методов, например, from_predictions можно сразу нарисовать картинку:

from sklearn.metrics import ConfusionMatrixDisplay


ConfusionMatrixDisplay.from_predictions(y_true=df['target'], y_pred=y_p,
                                              labels=labels, colorbar=False,
                                              xticks_rotation='vertical', im_kw={'cmap':'viridis'});

Однако для более гибкой настройки лучше создать класс с помощью конструктора, инициализировав параметр confusion_matrix нашей матрицей ошибок, а затем вызвать метод plot объекта с настройками отображения:

cmp = ConfusionMatrixDisplay(confusion_matrix=cm, display_labels=labels)

fig, ax = plt.subplots(figsize=(5,5))
cmp.plot(ax=ax, xticks_rotation='vertical', colorbar=False, im_kw={'cmap':'coolwarm'},
         text_kw={'color':"white"})

"Будущее уже здесь, просто оно еще не равномерно распределено" (Уильям Гибсон). В наши дни будущее, где каждому можно получить собственную виртуальную машину и развернуть микросервис, уже наступило. Но все ли знают, как это сделать? В этом материале мы вместе пройдем по шагам, воспользовавшись сервисом cloud.ru.

Создание ключа доступа

Предварительно сгенерируем rsa ключи для входа, используя утилиту ssh-keygen:

Сгенерируется пара ключей rsa в домашнем каталоге:

Создание машины

Теперь надо перенести публичную часть (id_rsa.pub) на сервер. Логинимся удобным способом на cloud.ru, например, по сбер id. Кликаем в левом верхнем углу на прямоугольник с точками и выбираем опцию SSH-ключи:

Выбираем "Создать ключ":

Задаем имя, "загружаем публичный ключ из файла" и жмем создать:

Затем кликаем на прямоугольник из точек в левом верхнем углу -> инфраструктура-> виртуальные машины:

Выбираем "Создать виртуальную машину":

Переходим в опцию "Получить ВМ бесплатно" (на картинке в левом нижнем углу):

Заполняем название, выбираем образ системы (я выбрал ubuntu), указываем имя пользователя и публичный ключ, который недавно создали. Также ставим галочку на "Пароль" и задаем его для входа через пароль (например, в виртуальном терминале в личном кабинете):

Можно сразу подключить публичный IP или потом, чтобы взаимосвязь с машиной происходила через Интернет (это платно, около 150 рублей в месяц, однако на 2 месяца у вас будет 4 тыс. рублей в качестве бонусов). В конце кликаем на "Создать" в самом низу:

И ждем, когда машина будет создана:

Затем можно нажать на троеточие справа и перейти в виртуальную консоль:

После ввода логина и пароля, вы получите доступ в терминал:

Видеоинструкцию по созданию машины можете посмотреть здесь.

Внешний ip

Теперь давайте назначим публичный ip. Перейдите в опцию "Виртуальные машины" (по аналогии, как выше), нажмите на троеточие и выберите соответствующий пункт:

Выберите интерфейс из выпадающего списка и нажмите "Назначить":

Теперь у нас есть ip, и можно для удобства зайти на виртуальную машину по ssh с локальной машины:

ssh имя_пользователя@ip

Так как мы настроили аутентификацию по rsa, пароль вводить не придется.

Микросервис

установка docker

Установим docker, набрав команды (версия для ubuntu):

sudo apt update
sudo apt install docker.io

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

sudo systemctl start docker

После этого желательно добавить текущего пользователя в группу docker, чтобы команды данного инструмента выполнять без привилегий суперпользователя (sudo):

sudo usermod -aG docker al

настройка правил для подключений

Перед запуском контейнера настроим правило для входящего трафика (тут подробнее) от Jupyter lab.

Опять жмем прямоугольник из точек, выбираем раздел "Группы безопасности", нажимаем на группу:

Переходим во вкладку "Правила", жмем "Добавить правило":

и заполняем протокол, порт, ip адрес и маску (для разрешений всем - 0.0.0.0/0):

или так:

Свой ip можно посмотреть тут. Для отдельных пользователей и малых офисов провайдер обычно использует динамические адреса и маску подсети /32, указывающую, что все биты адреса используются для идентификации конкретного устройства. Соответственно, из-за динамической природы адреса ваше правило для одиночного ip при следующем соединении может перестать работать.

запуск сервиса

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

docker run -p 8888:8888 -p 4040:4040 -it --user root jupyter/pyspark-notebook start.sh jupyter lab

Затем на своей машине в браузере вводим:

http://ip:8888/lab?token=a5a6e2d32063017ea28ebe0ea1ab1d928eadea15d16cab50

Напомню, что token высветится в логах запуска контейнера прямо в терминале.

Теперь наслаждаемся результатом:

Визуализация — это язык, который позволяет нам видеть данные и понимать их смысл. Простой и эффективный способ диагностики результатов работы модели на различных объектах заключается в анализе разницы между прогнозами и целями. Он может показать, что в некоторых группах поведение модели имеет особенности (например, склонность к завышению или занижению прогнозов). Для демонстрации того, как строится такая визуализация загрузим набор данных:

from sklearn.datasets import load_diabetes
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

np.random.seed(0)

df, y = load_diabetes(return_X_y=True, as_frame=True)
df['target'] = y
display(df.head())
display(df.shape)

Разобьем датасет на две группы для обучения и оценки:

from sklearn.model_selection import train_test_split

X_tr, X_ts, y_tr, y_ts = train_test_split(df.drop(columns='target').copy(),
                                          df['target'], test_size=0.2)
y_tr.shape[0], y_ts.shape[0]

Теперь построим модель и сделаем предсказания:

from sklearn.linear_model import LinearRegression

model = LinearRegression()

model.fit(X_tr, y_tr)
y_p = model.predict(X_ts)

Аналитический прием, о котором шла речь выше, заключается в визуализации распределения разности между целями и прогнозами (ошибками). Это можно сделать, например, путем построения гистограммы или графика разброса точек с координатами по оси y - ошибки, x - предсказания. Объявим функцию с соответствующими свойствами и вызовем ее:

def plot_residuals(target, predictions, bins_num, figsize=(20, 8), style='seaborn'):

    error = target - predictions
    with plt.style.context(style=style):

      plt.figure(figsize=figsize)
      plt.suptitle(f'Анализ ошибок', fontsize=16)

      plt.subplot(1, 2, 1)
      plt.hist(error, edgecolor='blue', bins=bins_num)
      plt.axvline(x=0, color='black', label='ноль', linestyle='--')
      plt.axvline(x=error.median(), color='red', label='медиана')
      plt.axvline(x=error.mean(), color='orange', label='среднее')
      plt.title(f'Гистограмма ошибок', fontsize=15)
      plt.ylabel('плотность распределения', fontsize=14)
      plt.xlabel('ошибки', fontsize=14)
      plt.legend()

      plt.subplot(1, 2, 2)
      plt.scatter(predictions, error, alpha=0.4)
      plt.axhline(y=0, color='red', label='ноль', linestyle='--')
      plt.title(f'Анализ дисперсии ошибок', fontsize=15)
      plt.ylabel('ошибки', fontsize=14)
      plt.xlabel('предсказания модели', fontsize=14)

plot_residuals(y_ts, y_p, bins_num = 10, figsize=(20, 5), style='bmh')

На графике ошибки распределены равномерно относительно нуля, их среднее и медиана почти совпадают и равны 0.

Аналогичные графики можно построить с библиотекой sklearn (потребуется использовать метод from_predictions класса PredictionErrorDisplay из модуля sklearn.metrics):

from sklearn.metrics import PredictionErrorDisplay

PredictionErrorDisplay.from_predictions(y_ts, y_p)

По оси y можно вывести вместо ошибок реальные значения (цели) против предсказанных по оси x:

PredictionErrorDisplay.from_predictions(y_ts, y_p, kind='actual_vs_predicted')

А теперь для демонстрационных целей добавим выброс в виде новой точки с очень большой целью и снова обучим модель:

model.fit(pd.concat([X_tr, X_tr.iloc[[-1]]]),
          pd.concat([y_tr.to_frame(), pd.Series([1e10]).to_frame('target')])['target'])

y_p = model.predict(X_ts)

plot_residuals(y_ts, y_p, bins_num = 10, figsize=(20, 5), style='bmh')

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

Таким образом, неравномерность распределения ошибок относительно нуля является индикатором того, что модель требует доработки и оптимизации, а причины могут от случая к случаю меняться.

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

Спросим у 4 нейросетей ChatGPT, GigaChat, Алиса Про,Gemini одно и то же, а потом попробуем скорректировать ошибки. Запрос такой:

можно ли методом compare в pandas сравнивать датафреймы с разным количеством строк. Приведи пример кода.

Ответ Алиса Про:

Правильный. Кстати 10 минутами до, спрашивал то же и нейросеть отвечала постоянно неправильно, приводя неработающие куски кода (может мои многократные повторы одного вопроса расценились, как неудовлетворительный результат).

Ответ GigaChat:

Опять ответ верный и противоположный прежнему. Однако я вспомнил, что до этого задавал тот же запрос без "в" между словами compare и pandas. Вот, какой ответ в этом случае:

Уже ответ неверный, однако GigaChat дает нам пример другого полезного метода.

Ответ от ChatGPT:

На удивление у ChatGPT ответ самый худший. Неправильный с неработающим куском кода.

Ответ от Gemini:

Неправильный с неработающим кодом. При этом удивительно, что 10 минут назад выдавала стабильно правильный ответ. Единственный момент - я запустил этот и предыдущие вопросы в рамках одного чата, соответственно, на ответ повлиял контекст. Действительно, открыв новый чат, снова получил правильный ответ.

Резюме:

- 3 нейросети (кроме ChatGPT) результат поменяли на противоположный в течение нескольких минут.

- На результат могут повлиять даже незначительные словесные элементы (в нашем примере - частицы)

- На результат сильно влияет контекст

А теперь проверим, "чувствительность" к промту по исправлению ошибок Gemini и ChatGPT.

можно ли методом compare в pandas сравнивать датафреймы с разным количеством строк. Приведи пример кода. Убедись, что пример работает и в нем нет ошибок.

ChatGPT:

Опять пример неработающий.

Gemini:

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

Отмечу, что проверял и другие промты, нацеленные на исправление ошибок:

"запрос...Если есть в коде ошибки, найди и устрани их"

"запрос...Проверь код в интерпретаторе Python и исправь ошибки"

В целом такими промтами повлиять на результат ChatGPT, а также GigaChat, Алиса Про (когда они давали неверный результат) не удалось. Только с Gemini частично получилось.

Вот еще пример. Зададим вопрос:

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

Из СМИ известно, что в октябре 2019 года.

Алиса Про:

GigaChat:

Сам я точную дату не знаю, но в октябре 2019. То есть с месяцем GigaChat не ошибся. Опять же пару часов назад, он отказывался отвечать на этот вопрос. То есть результаты не стабильны.

ChatGPT:

Gemini:

Получается ChatGPT ошибся. Попробуем улучшающий промт:

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

Интересно, но в данном случае он исправил месяц. При этом более 5 раз до выдавал дату 11 сентября.

Уже после спрашиваю в новом чате в первоначальной формулировке (без дополнения промта просьбой перепроверить факты и даты) ChatGPT отвечает - 25 октября. То есть опять же результаты не стабильны, возможно, коррекция ответа вызвана настойчивыми попытками задать один и тот же вопрос, а возможно эффектом от просьбы перепроверить.

Таким образом, нельзя говорить о стабильном влиянии на результат "перероверяющих" промтов ("проверь факты и даты", "если есть ошибки, найди и устрани их"). Возможно, в отдельных случаях они влияют, но в большинстве - нет.

Моя рекомендация:

- спрашивать у разных нейросетей, выбирая подходящий ответ

- для разных тематик открывать отдельные чаты, чтобы минимизировать влияние контекста.