Эпизодическая + Семантическая память

Для всех наших агентов, у которых память как у золотой рыбки, но как только разговор заканчивается, все забывается.

 

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

 

Архитектура «Эпизодическая + Семантическая память» имитирует человеческое познание, предоставляя агенту два типа памяти:

 

 

 

Эпизодическая память (созданная

Фарид Хан

)

Эпизодическая память: это память о конкретных событиях, например, о прошлых разговорах. Она отвечает на вопрос: «Что произошло?» Для этого мы будем использовать векторную базу данных .

Семантическая память: это память о структурированных фактах и ​​взаимосвязях, извлеченных из этих событий. Она отвечает на вопрос: «Что я знаю?» Для этого мы будем использовать графовую базу данных (Neo4j) .

Возможно, вы знаете, что это лежит в основе персонализации любой системы искусственного интеллекта. Именно так бот для электронной коммерции запоминает ваш стиль, репетитор — ваши слабые места, а личный помощник — ваши проекты и предпочтения на протяжении недель и месяцев.

 

Вот как это работает…

 

Взаимодействие: Агент ведет диалог с пользователем.

Извлечение информации из памяти: При поступлении нового запроса агент осуществляет поиск релевантного контекста как в своей эпизодической (векторной), так и в семантической (графовой) памяти.

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

Создание памяти: После взаимодействия агент, «создающий память», анализирует разговор, составляет краткое изложение (эпизодическая память) и извлекает факты (семантическая память).

Хранение в памяти: Новые данные сохраняются в соответствующих базах данных.

В основе этой системы лежит «Создатель воспоминаний» — агент, отвечающий за обработку разговоров и создание новых воспоминаний. У него две задачи: создать краткое резюме для векторного хранилища и извлечь структурированные факты для графа.

 

# Пидантические модели для извлечения знаний 

class Node ( BaseModel ): 

    id : str ; type : str 

 

class Relationship ( BaseModel ): 

    source: Node; target: Node; type : str 

 

class KnowledgeGraph ( BaseModel ): 

    relationships: List [Relationship] 

 

def create_memories ( user_input: str , assistant_output: str ): 

    conversation = f"User: {user_input} \nAssistant: {assistant_output} " 

    

    # Создание эпизодической памяти (суммирование)

     console. print ( "--- Создание эпизодической памяти (резюме) ---" ) 

    summary_prompt = ChatPromptTemplate.from_messages([ 

        " system" , "Вы эксперт по составлению резюме. Создайте краткое резюме следующего взаимодействия пользователя и помощника в одно предложение. Это резюме будет использоваться в качестве памяти для последующего воспроизведения." ), 

        ( "human" , "Взаимодействие:\n{interaction}" ) 

    ]) 

    summarizer = summary_prompt | llm 

    episodic_summary = summarizer.invoke({ "interaction" : conversation}).content 

    

    new_doc = Document(page_content=episodic_summary, metadata={ "created_at" : uuid.uuid4(). hex }) 

    episodic_vector_store.add_documents([new_doc]) 

    console. print ( f"[green]Эпизодическая память создана:[/green] ' {episodic_summary} '" ) # Создание 

    

    семантической памяти (извлечение фактов)

     console.print ( "--- Создание семантической памяти (графа) ---" ) extraction_llm = llm.with_structured_output(KnowledgeGraph) extraction_prompt = ChatPromptTemplate.from_messages([ " system" , "Вы эксперт по извлечению знаний. Ваша задача — определить ключевые сущности и их взаимосвязи из разговора и смоделировать их в виде графа. Сосредоточьтесь на предпочтениях пользователя, целях и заявленных фактах." ), ( "human" , "Извлечь все взаимосвязи из этого взаимодействия:\n{interaction}" ) ]) extractor = extraction_prompt | extraction_llm try

 

 

 

 

 

 

    : 

        kg_data = extractor.invoke({ "interaction" : conversation}) 

        if kg_data.relationships: 

            for rel in kg_data.relationships: 

                graph.add_graph_documents([rel], include_source= True ) 

            console. print ( f"[green]Создана семантическая память:[/green]Добавлено { len (kg_data.relationships)} связей в граф." ) 

        else : 

            console. print ( "[yellow]В этом взаимодействии не обнаружено новых семантических воспоминаний.[/yellow]" ) 

    except Exception as e: 

        console. print ( f"[red]Не удалось извлечь или сохранить семантическую память: {e} [/red]" )

Имея готовый механизм формирования памяти, мы можем создать полноценный агент. Граф представляет собой простую последовательность: retrieveвоспоминания, generateответ с их использованием, а затем updateвоспоминание с новым диалогом.

 

class AgentState ( TypedDict ): 

    user_input: str

     retrieved_memories: Optional [ str ] 

    generation: str 

 

def retrieve_memory ( state: AgentState ) -> Dict [ str , Any ]: 

    """Узел, который извлекает воспоминания как из эпизодической, так и из семантической памяти."""

     console. print ( "--- Извлечение воспоминаний ---" ) 

    user_input = state[ 'user_input' ] 

    

    # Извлечение из эпизодической памяти

     retrieved_docs = episodic_vector_store.similarity_search(user_input, k= 2 ) 

    episodic_memories = "\n" .join([doc.page_content for doc in retrieved_docs]) 

    

    # Извлечение из семантической памяти 

    # Это простое извлечение; Более продвинутый вариант предполагает извлечение сущностей из запроса. 

    Попробуйте : 

        graph_schema = graph.get_schema 

        # Использование полнотекстового индекса для лучшего поиска. Neo4j автоматически создает его на основе свойств узлов. 

        # Более надежное решение может включать предварительное извлечение сущностей из user_input.

         semantic_memories = str (graph.query( """ 

            UNWIND $keywords AS keyword 

            CALL db.index.fulltext.queryNodes("entity", keyword) YIELD node, score 

            MATCH (node)-[r]-(related_node) 

            RETURN node, r, related_node LIMIT 5 

            """ , { 'keywords' : user_input.split()})) 

    except Exception as e: 

        semantic_memories = f"Не удалось выполнить запрос к графу: {e} "

        

     retrieved_content = f"Релевантные прошлые разговоры (эпизодическая память):\n {episodic_memories} \n\nРелевантные факты (семантическая память):\n {semantic_memories} "

     console. print ( f"[cyan]Retrieved Context:\n {retrieved_content} [/cyan]" ) 

    

    return { "retrieved_memories" : retrieved_content} 

 

def generate_response ( state: AgentState ) -> Dict [ str , Any ]: 

    """Узел, генерирующий ответ с использованием полученных воспоминаний.

    консоль. печать( "--- Генерация ответа ---" ) 

    prompt = ChatPromptTemplate.from_messages([ 

        ( "system" , "Вы полезный и персонализированный финансовый помощник. Используйте полученные воспоминания, чтобы сформировать свой ответ и адаптировать его к пользователю. Если воспоминания указывают на предпочтения пользователя (например, он консервативный инвестор), вы ОБЯЗАТЕЛЬНО должны их уважать." ), 

        ( "human" , "Мой вопрос: {user_input}\n\nВот несколько воспоминаний, которые могут быть актуальны:\n{retrieved_memories}" ) ] 

    ) 

    generator = prompt | llm 

    generation = generator.invoke(state).content 

    console. print ( f"[green]Сгенерированный ответ:\n {generation} [/green]" ) 

    return { "generation" : generation} 

 

def update_memory ( state: AgentState ) -> Dict [ str , Any ]: 

    """Узел, который обновляет память последним взаимодействием."""

     console. print ( "--- Обновление памяти ---" ) 

    create_memories(state[ 'user_input' ], state[ 'generation' ]) 

    return {} 

 

workflow = StateGraph(AgentState) 

workflow.add_node( "retrieve" , retrieve_memory) 

workflow.add_node( "generate" , generate_response) 

workflow.add_node( "update" , update_memory) 

 

# ... (последовательное подключение узлов) ...

 memory_agent = workflow.compile ( )

 

 

 

 

 

Эпизодическая память (созданная

Фарид Хан

)

Единственный способ это проверить — провести многоходовый диалог. Один диалог будет посвящен заполнению памяти, а второй — проверке способности агента её использовать.

 

# Взаимодействие 1: Заполнение памяти

 run_interaction( "Привет, меня зовут Алекс. Я консервативный инвестор, в основном интересуюсь состоявшимися технологическими компаниями." ) 

 

# Взаимодействие 2: ТЕСТ ПАМЯТИ

 run_interaction( "Исходя из моих целей, какая хорошая альтернатива Apple?" )

--- 💬 ВЗАИМОДЕЙСТВИЕ 1: Заполнение памяти ---   

--- Извлечение воспоминаний...   

Извлеченный контекст: Прошлые разговоры + исходный документ...   

--- Генерация ответа...   

Сгенерированный ответ: Привет, Алекс! Стратегия консервативного инвестора учтена...   

--- Обновление памяти...   

Создана эпизодическая память: «Пользователь Алекс — консервативный инвестор в сфере технологий...»   

Создана семантическая память: Добавлено 2 связи...   

 

 

--- 💬 ВЗАИМОДЕЙСТВИЕ 2: Задавание конкретного вопроса ---   

--- Извлечение воспоминаний...   

Извлеченный контекст: Алекс как консервативный инвестор...   

--- Генерация ответа...   

Сгенерированный ответ: Apple (AAPL) подходит для консервативного технологического портфеля, сильный бренд, стабильный доход...   

--- Обновление памяти...   

Создана эпизодическая память: «Пользователь спросил об Apple; помощник подтвердил соответствие...»   

Создана семантическая память: Добавлено 1 связь...

Агент без сохранения состояния не справляется со вторым запросом, потому что не знает целей Алекса. Наш же агент с расширенной памятью успешно выполняет запрос.

 

Функция «Эпизодоичное воспроизведение» извлекает краткое содержание первого разговора: «Пользователь, Алекс, представился как консервативный инвестор…»

Напомним, что семантический подход заключается в следующем: он запрашивает информацию из графа и находит следующий факт: (User: Alex) -[HAS_GOAL]-> (InvestmentPhilosophy: Conservative).

Синтез: Используя этот объединенный контекст, система дает идеальную, персонализированную рекомендацию.

Мы можем формализовать это с помощью магистра права в качестве судьи, который будет оценивать персонализацию.

 

class PersonalizationEvaluation ( BaseModel ): 

    personalization_score: int = Field(description= "Оценка от 1 до 10 того, насколько хорошо агент использовал прошлые взаимодействия и предпочтения пользователя для адаптации своего ответа." ) 

    justification: str = Field(description= "Краткое обоснование оценки." )

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

 

--- Оценка персонализации ответа агента памяти --- 

{ 

    'personalization_score': 7, 

    'justification': "Ответ агента был идеально персонализирован. Он явно ссылался на заявленную цель пользователя — быть 'консервативным инвестором' (что он вспомнил из предыдущего разговора), — чтобы обосновать свою рекомендацию Microsoft. Это демонстрирует глубокое, осознанное понимание пользователя." 

}

Путем сочетания эпизодической и семантической памяти…

 

Мы можем создавать агентов, которые выходят за рамки простых вопросов и ответов и становятся настоящими помощниками в обучении.