Пошаговое руководство по созданию автономных систем извлечения информации из памяти.
Делиться
Каждый вызов LLM — это новое начало. Если вы явно не предоставите информацию из предыдущих сессий, модель не имеет встроенного ощущения непрерывности между запросами или сессиями. Такая безсостоятельная архитектура отлично подходит для параллельной обработки и безопасности, но представляет собой огромную проблему для чат-приложений, требующих персонализации на уровне пользователя.
Если ваш чат-бот каждый раз при входе в систему воспринимает пользователя как незнакомца, как он сможет генерировать персонализированные ответы?
В этой статье мы с нуля создадим простую систему памяти, вдохновленную популярной архитектурой Mem0.
Если не указано иное, все представленные здесь иллюстрации созданы мной, автором.
Цель этой статьи — ознакомить читателей с управлением памятью как с проблемой контекстного проектирования . В конце статьи вы также найдете:
- Ссылка на GitHub , содержащая полный проект по работе с памятью; вы можете разместить его на своем сервере.
- Подробный видеоурок на YouTube , в котором концепции разбираются построчно.
Память как проблема контекстного проектирования
Контекстная инженерия — это метод наполнения контекста LLM всей необходимой информацией для выполнения задачи. На мой взгляд, память — одна из самых сложных и интересных задач контекстной инженерии.
Изучение работы с памятью знакомит вас (как разработчика) с некоторыми из наиболее важных методов, необходимых практически для всех задач контекстной инженерии, а именно:
- Извлечение структурированной информации из необработанных текстовых потоков.
- Подведение итогов
- Векторные базы данных
- Генерация запросов и поиск сходства
- Постобработка запросов и переранжирование
- Вызов агентского инструмента
И многое другое.
Поскольку мы создаём слой памяти с нуля, нам придётся применить все эти методы! Читайте дальше.
Архитектура высокого уровня
На первый взгляд, система должна уметь выполнять четыре задачи: извлечение , встраивание, получение и сопровождение . Давайте рассмотрим основные планы, прежде чем приступать к реализации.
Компоненты
• Извлечение : Извлекает потенциальные атомарные ячейки памяти из текущих сообщений пользовательского помощника.
• Векторная база данных : Встраивает извлеченные фактоиды в непрерывные векторы и сохраняет их в векторной базе данных.
• Извлечение информации : Когда пользователь задает вопрос, мы генерируем запрос с использованием LLM и извлекаем воспоминания, похожие на этот запрос.
• Поддержание : Используя цикл ReAct (рассуждение и действие), агент принимает решение о добавлении, обновлении, удалении или бездействии в зависимости от хода и противоречий с существующими фактами.
Важно отметить, что каждый из описанных выше шагов должен быть необязательным. Если агенту LLM не требуется доступ к предыдущим воспоминаниям для ответа на вопрос, он вообще не должен пытаться искать информацию в нашей векторной базе данных.
Стратегия заключается в том, чтобы предоставить магистру права все необходимые инструменты для выполнения задач, а также четкие инструкции по использованию каждого инструмента, и полагаться на интеллектуальные способности магистра права в автономном использовании этих инструментов!
Давайте посмотрим, как это работает на практике!
2) Извлечение информации из памяти с помощью DSPy: от транскрипта до интересных фактов
В этом разделе давайте разработаем надежный этап извлечения информации, который преобразует стенограммы разговоров в набор атомарных, классифицированных фактоидов.
Что мы извлекаем и почему это важно
Цель состоит в создании хранилища в памяти, представляющего собой постоянную базу данных, основанную на векторном представлении данных, для каждого пользователя.
Что такое «хорошая» память?
Краткая, самодостаточная величина — атомная единица, которую можно закрепить и впоследствии извлечь с высокой точностью.
С помощью DSPy извлечение структурированной информации очень просто. Рассмотрим приведенный ниже фрагмент кода.
- Мы определяем сигнатуру DSPy под названием MemoryExtract.
- Входными данными для этой подписи (обозначенной как InputField) являются текст транскрипта.
- Ожидаемый результат (обозначенный как OutputField) — это список строк, содержащих каждый фактоид.
Ввод контекстной строки, вывод списка строк из памяти.
# … другие импорты import dspy from pydantic import BaseModel class MemoryExtract(dspy.Signature): «»» Извлекает релевантную информацию из разговора. Воспоминания — это атомарные независимые фактоиды, которые мы должны узнать о пользователе. Если стенограмма не содержит никакой информации, которую стоит извлекать, возвращает пустой список. «»» transcript: str = dspy.InputField() memories: list[str] = dspy.OutputField() memory_extractor = dspy.Predict(MemoryExtract)
В DSPy строка документации сигнатуры используется в качестве системной подсказки. Мы можем настроить строку документации, чтобы явно указать, какую информацию LLM будет извлекать из диалога.
Наконец, для извлечения воспоминаний мы передаем историю разговоров в средство извлечения памяти в виде строки JSON. Ознакомьтесь с фрагментом кода ниже.
async def extract_memories_from_messages(messages): transcript = json.dumps(messages) with dspy.context(lm=dspy.LM(model=MODEL_NAME)): out = await memory_extractor.acall(transcript=transcript) return out.memories # возвращает список воспоминаний
Вот и всё! Давайте запустим код с фиктивным диалогом и посмотрим, что получится.
if __name__ == «__main__»: messages = [ { «role»: «user», «content»: «Мне нравится кофе» }, { «role»: «assistant», «content»: «Понял!» }, { «role»: «user», «content»: «На самом деле, нет, мне больше нравится чай. Мне также нравится футбол» } ] memories = asyncio.run(extract_memories_from_messages(messages)) print(memories) ''' Вывод: [ «Пользователь раньше любил чай, но теперь нет», «Пользователь любит кофе», «Пользователь любит футбол» ] '''
Как видите, мы можем извлекать независимые факты из разговоров. Что это значит?
Мы можем сохранить извлеченные факты в базе данных, которая находится вне сеанса чата.
Если вас заинтересовала технология DSPy, ознакомьтесь со статьей «Context Engineering with DSPy», в которой более подробно рассматривается эта концепция. Или посмотрите видео ниже.
Встраивание извлеченных воспоминаний
Таким образом, мы можем извлекать воспоминания из разговоров. Далее, давайте внедрим их, чтобы в конечном итоге сохранить в векторной базе данных.
В этом проекте мы будем использовать QDrant в качестве векторной базы данных — у них есть отличный бесплатный тариф, который работает очень быстро и поддерживает дополнительные функции, такие как гибридная фильтрация (где вы можете передавать атрибутивные фильтры, подобные SQL-запросу «where», в ваш векторный запрос).
Выбор модели встраивания и фиксация размерности.
Для обеспечения оптимального соотношения цены, скорости и высокого качества при работе с короткими фактами мы выбираем параметр text-embedding-3-small . Размер вектора мы устанавливаем на 64, что уменьшает объем занимаемого места и ускоряет поиск, оставаясь при этом достаточно выразительным для кратких воспоминаний. Этот гиперпараметр мы можем настроить позже в соответствии с нашими потребностями.
client = openai.AsyncClient() async def generate_embeddings(strings: list[str]): out = await client.embeddings.create( input=strings, model=»text-embedding-3-small», dimensions=64 ) embeddings = [item.embedding for item in out.data] return embeddings
Для вставки данных в QDrant давайте сначала создадим наши базы данных и индекс по user_id. Это позволит нам быстро фильтровать записи по пользователям.
from qdrant_client import AsyncQdrantClient COLLECTION_NAME = «memories» async def create_memory_collection(): if not (await client.collection_exists(COLLECTION_NAME)): await client.create_collection( collection_name=COLLECTION_NAME, vectors_config=VectorParams(size=64, distance=Distance.DOT), ) await client.create_payload_index( collection_name=COLLECTION_NAME, field_name=»user_id», field_schema=models.PayloadSchemaType.INTEGER )
Мне нравится определять контракты с помощью Pydantic в самом начале, чтобы другие модули знали структуру выходных данных этих функций.
from pydantic import BaseModel class EmbeddedMemory(BaseModel): user_id: int memory_text: str date: str embedding: list[float] class RetrievedMemory(BaseModel): point_id: str user_id: int memory_text: str date: str score: float
Далее напишем вспомогательные функции для вставки, удаления и обновления памяти.
async def insert_memories(memories: list[EmbeddedMemory]): «»» Получив список воспоминаний, вставьте их в базу данных «»» await client.upsert( collection_name=COLLECTION_NAME, points=[ models.PointStruct( id=uuid4().hex, payload={ «user_id»: memory.user_id, «memory_text»: memory.memory_text, «date»: memory.date }, vector=memory.embedding ) for memory in memories ] ) async def delete_records(point_ids): «»» Удалите список идентификаторов точек из базы данных «»» await client.delete( collection_name=COLLECTION_NAME, points_selector=models.PointIdsList( points=point_ids ) )
Аналогично, давайте напишем функцию для поиска. Она принимает вектор поиска и идентификатор пользователя и извлекает ближайших соседей к этому вектору.
from qdrant_client.models import Distance, Filter, models async def search_memories( search_vector: list[float], user_id: int, topk_neighbors=5 ): # Фильтрация по user_id must_conditions: list[models.Condition] = [ models.FieldCondition( key=»user_id», match=models.MatchValue(value=user_id) ) ] outs = await client.query_points( collection_name=COLLECTION_NAME, query=search_vector, with_payload=True, query_filter=Filter(must=must_conditions), score_threshold=0.1, limit=topk_neighbors ) return [ convert_retrieved_records(point) for point in outs.points if point is not None ]
Обратите внимание, как мы можем устанавливать гибридные фильтры запросов, например, фильтр models.MatchValue. Создание индекса по user_id позволяет быстро выполнять эти запросы к нашим данным. Вы можете расширить эту идею, включив в нее теги категорий, диапазоны дат и любые другие метаданные, которые важны для вашего приложения. Просто убедитесь, что вы создали индекс для более быстрой обработки данных.
В следующей главе мы подключим этот уровень хранения к нашему агентскому циклу, используя сигнатуры DSPy и ReAct (рассуждение и действие).
Извлечение памяти
В этом разделе мы создаём удобный интерфейс поиска, который извлекает наиболее релевантные воспоминания каждого пользователя для данного хода.
Наш алгоритм прост – мы создадим чат-бота, который будет вызывать инструменты. На каждом этапе агент получает расшифровку разговора и должен сгенерировать ответ. Давайте определим сигнатуру DSPy.
class ResponseGenerator(dspy.Signature): «»» Вам будет предоставлена стенограмма предыдущего разговора между пользователем и агентом ИИ. А также последний вопрос пользователя. У вас есть возможность поискать предыдущие воспоминания в векторной базе данных, чтобы получить необходимый контекст, если это требуется. Если вы не можете найти ответ на вопрос пользователя в стенограмме или из собственных внутренних знаний, используйте предоставленные вызовы инструмента поиска для поиска информации. Вы должны вывести окончательный ответ, а также решить, нужно ли записать последнее взаимодействие в базу данных памяти. Новые воспоминания предназначены для хранения новой информации, предоставленной пользователем. Новые воспоминания должны создаваться, когда ПОЛЬЗОВАТЕЛЬ предоставляет новую информацию. Это не для сохранения информации об ИИ или помощнике. «»» transcript: list[dict] = dspy.InputField() question: str = dspy.InputField() response: str = dspy.OutputField() save_memory: bool = dspy.OutputField(description= «True, если необходимо создать новую запись в памяти для последнего взаимодействия» )
Документация (docstring) сигнатуры dspy служит дополнительными инструкциями, которые мы передаем в LLM, чтобы помочь ему выбрать необходимые действия. Также обратите внимание на флаг save_memory, который мы обозначили как OutputField. Мы также просим LLM выводить информацию о необходимости сохранения новой памяти в связи с последним взаимодействием с ответом.
Нам также необходимо решить вопрос о том, как получать релевантные воспоминания в контекст агента. Один из вариантов — всегда выполнять функцию search_memories, но с этим связаны две серьезные проблемы:
- Не на все вопросы пользователей требуется вспоминать информацию из памяти.
- Хотя функция search_memories ожидает вектор поиска, не всегда очевидно, «какой текст нам следует встроить». Это может быть вся стенограмма, или только последнее сообщение пользователя, или же преобразование контекста текущей беседы.
К счастью, мы можем использовать вызов инструмента по умолчанию. Когда агент считает, что ему не хватает контекста для выполнения запроса, он может вызвать инструмент, чтобы получить соответствующие воспоминания, связанные с контекстом разговора. В DSPy инструменты можно создавать, просто написав обычную функцию Python с документацией. LLM считывает эту документацию, чтобы решить, когда и как вызвать этот инструмент.
async def fetch_similar_memories(search_text: str): «»» Поиск воспоминаний в векторной базе данных, если для диалога требуется дополнительный контекст. Аргументы: — search_text : Строка для встраивания и выполнения поиска векторного сходства «»» search_vector = (await generate_embeddings([search_text]))[0] memories = await search_memories(search_vector, user_id=user_id) memories_str = [ f»id={m_.id}ntext={m_.text}ncreated_at={m_.date}» for m_ in memories ] return { «memories»: memories_str }
Обратите внимание, что мы отслеживаем идентификатор пользователя извне и используем его из нашего источника достоверной информации, не запрашивая у LLM его генерацию. Это гарантирует изоляцию в контексте текущей сессии чата.
Далее создадим агента ReAct с помощью DSPy. ReAct расшифровывается как «Рассуждение и действие». По сути, агент LLM анализирует данные (в данном случае, историю разговора), рассуждает на их основе, а затем действует.
Действие может заключаться в том, чтобы получить ответ напрямую или попытаться сначала вспомнить что-то.
response_generator = dspy.ReAct( ResponseGenerator, tools=[fetch_similar_memories], max_iters=4 )
В агентном потоке политика DSPy ReAct может сформировать краткий поисковый текст на основе текущего хода и известной задачи. Агент ReAct может вызывать функцию fetch_similar_memories до 4 раз для поиска воспоминаний, прежде чем ему потребуется ответить на вопрос пользователя.
Другие стратегии поиска
Помимо поиска по сходству, вы также можете выбрать другие стратегии поиска. Вот несколько идей:
- Поиск по ключевым словам – изучите такие алгоритмы, как BM-25 или TF-IDF.
- Фильтрация по категориям – Если принудительно задать для каждого объекта памяти четкую метаинформацию (например, «еда», «спорт», «привычки»), агент сможет генерировать запросы для поиска по этим конкретным подкатегориям, а не по всему стеку памяти.
- Временные запросы — позволяют агенту получать записи за определенные временные диапазоны!
Эти варианты во многом зависят от конкретного применения.
Независимо от выбранной стратегии получения данных, как только инструмент получит ответы LLM, агент сгенерирует ответы на основе полученных данных! Помните, он также выводит флаг save_memory? Мы можем запустить нашу собственную логику обновления, когда он будет установлен в значение true.
out = await response_generator.acall( transcript=past_messages, question=question, ) response = out.response # ответ save_memory = out.save_memory # решение LLM о сохранении памяти past_messages.extend( [ {«role»: «user», «content»: question}, {«role»: «assistant», «content»: response}, ] ) # обновить стек диалога if (save_memory): # Обновлять память только если LLM выводит этот флаг как true update_result = await update_memories( user_id=user_id, messages=past_messages, )
Давайте посмотрим, как работает этап обновления.
Обслуживание памяти
Память — это не просто журнал записей. Это постоянно развивающийся массив информации. Некоторые воспоминания следует удалить, потому что они больше не актуальны. Другие воспоминания необходимо обновить, потому что изменились условия окружающего мира.
Например, предположим, у нас есть воспоминание о том, что «пользователь любит чай», и мы только что узнали, что «пользователь ненавидит чай». Вместо того чтобы создавать совершенно новое воспоминание, мы должны удалить старое и создать новое.
Когда агент генератора ответов решает сохранить новые воспоминания, мы будем использовать отдельный агентский поток для определения способа обновления. Агент обновления памяти получает на вход новые воспоминания и список похожих на состояние разговора воспоминаний.
…. # если save_memory равно True response = await update_memories_agent( user_id=user_id, existing_memories=similar_memories, messages=messages )
После принятия решения об обновлении базы данных памяти агент управления памятью может выполнить четыре логических действия:
• add_memory(text) : Вставляет совершенно новый атомарный фактоид. Он вычисляет новое векторное представление и записывает запись для текущего пользователя. Перед вставкой также должна быть применена логика дедупликации.
• update_memory(id, updated_text) : Заменяет текст в существующей памяти. Удаляет старую точку, вставляет новый текст и повторно вставляет его под тем же пользователем, при необходимости сохраняя или корректируя категории. Это стандартный способ обработки уточнений или исправлений.
• delete_memories(ids): Удаляет одну или несколько воспоминаний, которые больше недействительны из-за противоречий или устаревания.
• no_op() : Явно ничего не делает, если агент обслуживания решает, что новая память неактуальна или уже полностью занята в состоянии базы данных.
Эта архитектура снова вдохновлена исследовательской работой Mem0.
Приведённый ниже код демонстрирует интеграцию этих инструментов в агент DSPy ReAct со структурированной сигнатурой и циклом выбора инструментов.
class MemoryWithIds(BaseModel): memory_id: int memory_text: str class UpdateMemorySignature(dspy.Signature): «»» Вам будет предоставлен диалог между пользователем и ассистентом, а также несколько похожих воспоминаний из базы данных. Ваша цель — решить, как объединить новые воспоминания в базе данных с существующими. Действия: — ДОБАВИТЬ: добавить новые воспоминания в базу данных как новые воспоминания. — ОБНОВИТЬ: обновить существующее воспоминание более подробной информацией. — УДАЛИТЬ: удалить из базы данных элементы памяти, которые больше не нужны из-за новой информации. — НЕТ: никаких действий не требуется. Если никаких действий не требуется, вы можете закончить. Меньше думайте и выполняйте действия. «»» messages: list[dict] = dspy.InputField() existing_memories: list[MemoryWithIds] = dspy.InputField() summary: str = dspy.OutputField( description=»Кратко опишите, что вы сделали. Очень коротко (менее 10 слов)» )
Далее, давайте напишем инструменты, необходимые нашему агенту обслуживания. Нам нужны функции для добавления, удаления и обновления памяти, а также фиктивная функция no_op, которую LLM может вызывать, когда хочет «перейти».
async def update_memories_agent( user_id: int, messages: list[dict], existing_memories: list[RetrievedMemory] ): def get_point_id_from_memory_id(memory_id): return existing_memories[memory_id].point_id async def add_memory(memory_ext: str) -> str: «»» Добавить новую память в базу данных. «»» embeddings = await generate_embeddings( [memory_text] ) await insert_memories( memories = [ EmbeddedMemory( user_id=user_id, memory_text=memory_text, date=datetime.now().strftime(«%Y-%m-%d %H:%m»), embedding=embeddings[0] ) ] ) return f»Память: '{memory_text}' была добавлена в БД» async def update(memory_id: int, updated_memory_text: str, ): «»» Обновление memory_id для использования updated_memory_text. Аргументы: memory_id: целочисленный индекс памяти для замены. updated_memory_text: простой атомарный фактоид для замены старой памяти новой. «»» point_id = get_point_id_from_memory_id(memory_id) await delete_records([point_id]) embeddings = await generate_embeddings( [updated_memory_text] ) await insert_memories( memories = [ EmbeddedMemory( user_id=user_id, memory_text=updated_memory_text, categories=categories, date=datetime.now().strftime(«%Y-%m-%d %H:%m»), embedding=embeddings[0] ) ] ) return «Память {memory_id} обновлена до: '{updated_memory_text}'» async def noop(): «»» Вызовите эту функцию, если никаких действий не требуется»» return «Действий не выполнено» async def delete(memory_ids: list[int]): «»» Удалите эти memory_ids из базы данных»»» await delete_records(memory_ids) return «Память {memory_ids} удалена» memory_updater = dspy.ReAct( UpdateMemorySignature, tools=[add_memory, update, delete, noop], max_iters=3 ) out = await memory_updater.acall( messages=messages, existing_memories=memory_ids )
Вот и всё! В зависимости от того, какое действие выберет агент ReAct, мы можем просто вставить, удалить, обновить или проигнорировать новые воспоминания. Ниже вы можете увидеть простой пример того, как всё выглядит при запуске кода.
Полная версия кода также включает дополнительные функции, такие как метаданные для точного поиска, которые я не рассматривал в этой статье, чтобы сделать её более понятной для начинающих. Обязательно ознакомьтесь с репозиторием GitHub по ссылке ниже или посмотрите обучающее видео на YouTube, чтобы изучить весь проект!
Что дальше?
Полную версию видеоурока, в котором более подробно рассказывается о создании агентов памяти, можно посмотреть здесь.
Репозиторий с кодом можно найти здесь: https://github.com/avbiswas/mem0-dspy
В этом руководстве были объяснены основные элементы системы памяти. Вот несколько идей о том, как расширить эту концепцию:
- Система графовой памяти — вместо векторной базы данных, храните воспоминания в графовой базе данных. Это означает, что ваши модули dspy должны извлекать триплеты вместо плоских строк для представления воспоминаний.
- Метаданные — Помимо текста, добавьте дополнительные фильтры атрибутов. Например, вы можете сгруппировать все воспоминания, связанные с «едой». Это позволит агентам LLM запрашивать определенные теги при получении воспоминаний, вместо того чтобы запрашивать все воспоминания сразу.
- Оптимизация подсказок для каждого пользователя : вы можете хранить важную информацию в своей базе данных в памяти и напрямую внедрять ее в системную подсказку. Эти данные передаются в каждое сообщение как данные сессии в памяти.
- Файловые системы : Еще один распространенный подход, который набирает популярность, — это поиск информации на основе файлов. Основные принципы остаются теми же, что мы обсуждали здесь, но вместо векторной базы данных можно использовать файловую систему. Вставка и обновление записей осуществляется путем записи файлов .md. А запросы обычно включают дополнительные этапы индексирования или просто использование таких инструментов, как поиск по регулярным выражениям или grep.
Мой Patreon:
https://www.patreon.com/NeuralBreakdownwithAVB
Мой канал на YouTube:
https://www.youtube.com/@avb_fj
Подписывайтесь на меня в Твиттере:
https://x.com/neural_avb
Читайте мои статьи:
https://towardsdatascience.com/author/neural-avb/
Источник: towardsdatascience.com
