状態なきエージェントの限界：Agentic Memoryで実現する「記憶」と「学習」の仕組み

pip install langchain langchain-openai langchain-community chromadb

import logging
from typing import List, Optional
from datetime import datetime

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_chroma import Chroma
from langchain.schema import HumanMessage, SystemMessage, AIMessage
from langchain.memory import VectorStoreRetrieverMemory

# ロギングの設定
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class AgenticMemoryAssistant:
    def __init__(self, persist_directory: str = "./db"):
        """
        Agentic Memoryを持つアシスタントの初期化。
        ベクトルDBとLLMのセットアップを行います。
        """
        try:
            # 埋め込みモデルの初期化 (OpenAIのtext-embedding-3-small等を想定)
            self.embeddings = OpenAIEmbeddings()
            
            # ベクトルストアの初期化
            self.vectorstore = Chroma(
                collection_name="agent_memory",
                embedding_function=self.embeddings,
                persist_directory=persist_directory
            )
            
            # 検索機能の設定 (上位3件を取得)
            retriever = self.vectorstore.as_retriever(search_kwargs={"k": 3})
            
            # LangChainのMemory機能をラップ
            self.memory = VectorStoreRetrieverMemory(retriever=retriever)
            
            # LLMの初期化 (GPT-4o等を想定)
            self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
            
            logger.info("AgenticMemoryAssistant initialized successfully.")
            
        except Exception as e:
            logger.error(f"Initialization failed: {e}")
            raise

    def _get_contextual_prompt(self, input_text: str) -> str:
        """
        過去の記憶を検索し、現在のコンテキストに合わせてプロンプトを構築する。
        """
        try:
            # 過去の関連記憶を取得
            relevant_memories = self.memory.load_memory_variables({"prompt": input_text})
            history = relevant_memories.get("history", [])
            
            context_str = "\n".join([f"- {mem}" for mem in history])
            
            system_prompt = f"""あなたは親切で学習能力のあるAIアシスタントです。
過去のユーザーとのやり取りや指摘を記憶しており、それに基づいて回答を調整します。

【過去の記憶（参考情報）】
{context_str if context_str else "まだ関連する記憶はありません。"}

上記の記憶を踏まえて、現在のユーザーの質問に答えてください。
もし記憶にある情報と矛盾する指示があれば、最新のユーザーの意図を優先しつつ、
過去の文脈も考慮して丁寧に説明してください。"""
            
            return system_prompt
            
        except Exception as e:
            logger.warning(f"Context retrieval failed: {e}. Proceeding without context.")
            return "あなたは親切なAIアシスタントです。"

    def chat(self, user_input: str) -> str:
        """
        ユーザーとの対話を行い、結果を記憶に保存する。
        """
        try:
            logger.info(f"User Input: {user_input}")
            
            # 1. コンテキストの取得とプロンプト構築
            system_prompt = self._get_contextual_prompt(user_input)
            
            messages = [
                SystemMessage(content=system_prompt),
                HumanMessage(content=user_input)
            ]
            
            # 2. LLMによる回答生成
            response = self.llm.invoke(messages)
            ai_response = response.content
            logger.info(f"AI Response: {ai_response}")
            
            # 3. 記憶の保存 (学習プロセス)
            # ユーザーの入力とAIの回答をペアで保存することで、文脈を記録
            self.save_memory(user_input, ai_response)
            
            return ai_response
            
        except Exception as e:
            logger.error(f"Error during chat execution: {e}")
            return "申し訳ありません。処理中にエラーが発生しました。"

    def save_memory(self, input_text: str, output_text: str):
        """
        対話内容をベクトルDBに保存する。
        ここでは単純化のため、ユーザーの指摘を重要な記憶として扱う。
        """
        try:
            # 保存するテキストを作成 (ユーザー入力とAI回答のペア)
            memory_content = f"User: {input_text}\nAssistant: {output_text}"
            
            # ベクトルDBに追加
            self.vectorstore.add_texts(
                texts=[memory_content],
                metadatas=[{"timestamp": datetime.now().isoformat()}]
            )
            logger.info("Memory saved successfully.")
            
        except Exception as e:
            logger.error(f"Failed to save memory: {e}")

# 実行例
if __name__ == "__main__":
    # 環境変数 OPENAI_API_KEY が設定されている前提
    try:
        assistant = AgenticMemoryAssistant()
        
        print("--- 1回目の対話 ---")
        res1 = assistant.chat("コードを書く時は、変数名はスネークケースで統一してください。")
        print(f"Bot: {res1}\n")
        
        print("--- 2回目の対話 (記憶の確認) ---")
        res2 = assistant.chat("ユーザー情報を管理するクラスを作成してください。")
        print(f"Bot: {res2}\n")
        
        # 期待される動作: 2回目の回答では、1回目の指示（スネークケース）を反映したコードが出力されるはずです。
        
    except KeyError:
        print("Error: OPENAI_API_KEY environment variable is not set.")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")