Memory

ConversationMemory

ConversationMemory(debug: bool = False)

Bases: ABC

Abstract base class for conversation memory storage.

Source code in packages/ai-parrot/src/parrot/memory/abstract.py

def __init__(self, debug: bool = False):
    self.logger = logging.getLogger(
        f"parrot.Memory.{self.__class__.__name__}"
    )
    self._json = JSONContent()
    self.debug = debug

create_history abstractmethod async

create_history(user_id: str, session_id: str, metadata: Optional[Dict[str, Any]] = None, chatbot_id: Optional[str] = None) -> ConversationHistory

Create a new conversation history.

Source code in packages/ai-parrot/src/parrot/memory/abstract.py

@abstractmethod
async def create_history(
    self,
    user_id: str,
    session_id: str,
    metadata: Optional[Dict[str, Any]] = None,
    chatbot_id: Optional[str] = None
) -> ConversationHistory:
    """Create a new conversation history."""
    pass

get_history abstractmethod async

get_history(user_id: str, session_id: str, chatbot_id: Optional[str] = None) -> Optional[ConversationHistory]

Get a conversation history.

Source code in packages/ai-parrot/src/parrot/memory/abstract.py

@abstractmethod
async def get_history(
    self,
    user_id: str,
    session_id: str,
    chatbot_id: Optional[str] = None
) -> Optional[ConversationHistory]:
    """Get a conversation history."""
    pass

update_history abstractmethod async

update_history(history: ConversationHistory) -> None

Update a conversation history.

Source code in packages/ai-parrot/src/parrot/memory/abstract.py

@abstractmethod
async def update_history(self, history: ConversationHistory) -> None:
    """Update a conversation history."""
    pass

add_turn abstractmethod async

add_turn(user_id: str, session_id: str, turn: ConversationTurn, chatbot_id: Optional[str] = None) -> None

Add a turn to the conversation.

Source code in packages/ai-parrot/src/parrot/memory/abstract.py

@abstractmethod
async def add_turn(
    self,
    user_id: str,
    session_id: str,
    turn: ConversationTurn,
    chatbot_id: Optional[str] = None
) -> None:
    """Add a turn to the conversation."""
    pass

clear_history abstractmethod async

clear_history(user_id: str, session_id: str, chatbot_id: Optional[str] = None) -> None

Clear a conversation history.

Source code in packages/ai-parrot/src/parrot/memory/abstract.py

@abstractmethod
async def clear_history(
    self,
    user_id: str,
    session_id: str,
    chatbot_id: Optional[str] = None
) -> None:
    """Clear a conversation history."""
    pass

list_sessions abstractmethod async

list_sessions(user_id: str, chatbot_id: Optional[str] = None) -> List[str]

List all session IDs for a user.

Source code in packages/ai-parrot/src/parrot/memory/abstract.py

@abstractmethod
async def list_sessions(
    self,
    user_id: str,
    chatbot_id: Optional[str] = None
) -> List[str]:
    """List all session IDs for a user."""
    pass

delete_history abstractmethod async

delete_history(user_id: str, session_id: str, chatbot_id: Optional[str] = None) -> bool

Delete a conversation history entirely.

Source code in packages/ai-parrot/src/parrot/memory/abstract.py

@abstractmethod
async def delete_history(
    self,
    user_id: str,
    session_id: str,
    chatbot_id: Optional[str] = None
) -> bool:
    """Delete a conversation history entirely."""
    pass

ConversationHistory dataclass

ConversationHistory(session_id: str, user_id: str, chatbot_id: Optional[str] = None, turns: List[ConversationTurn] = list(), created_at: datetime = datetime.now(), updated_at: datetime = datetime.now(), metadata: Dict[str, Any] = dict())

Manages conversation history for a session - replaces ConversationSession.

add_turn

add_turn(turn: ConversationTurn) -> None

Add a new turn to the conversation history.

Source code in packages/ai-parrot/src/parrot/memory/abstract.py

def add_turn(self, turn: ConversationTurn) -> None:
    """Add a new turn to the conversation history."""
    self.turns.append(turn)
    self.updated_at = datetime.now()

get_recent_turns

get_recent_turns(count: int = 5) -> List[ConversationTurn]

Get the most recent turns for context.

Source code in packages/ai-parrot/src/parrot/memory/abstract.py

def get_recent_turns(self, count: int = 5) -> List[ConversationTurn]:
    """Get the most recent turns for context."""
    return self.turns[-count:] if count > 0 else self.turns

get_messages_for_api

get_messages_for_api(model: str = 'claude') -> List[Dict[str, Any]]

Convert turns to API message format for LLM Model.

Source code in packages/ai-parrot/src/parrot/memory/abstract.py

def get_messages_for_api(self, model: str = 'claude') -> List[Dict[str, Any]]:
    """Convert turns to API message format for LLM Model."""
    messages = []
    if model == 'claude':
        # Claude expects messages in a specific format
        for turn in self.turns:
            messages.append({
                "role": "user",
                "content": turn.user_message
            })
            messages.append({
                "role": "assistant",
                "content": turn.assistant_response
            })
    else:
        # Default format for other models
        # This can be adjusted based on the specific model requirements
        for turn in self.turns:
            # Add user message
            messages.append({
                "role": "user",
                "content": [{"type": "text", "text": turn.user_message}]
            })
            # Add assistant response
            messages.append({
                "role": "assistant",
                "content": [{"type": "text", "text": turn.assistant_response}]
            })
    return messages

clear_turns

clear_turns() -> None

Clear all turns from the conversation history.

Source code in packages/ai-parrot/src/parrot/memory/abstract.py

def clear_turns(self) -> None:
    """Clear all turns from the conversation history."""
    self.turns.clear()
    self.updated_at = datetime.now()

to_dict

to_dict() -> Dict[str, Any]

Serialize conversation history to dictionary.

Source code in packages/ai-parrot/src/parrot/memory/abstract.py

def to_dict(self) -> Dict[str, Any]:
    """Serialize conversation history to dictionary."""
    return {
        'session_id': self.session_id,
        'user_id': self.user_id,
        'chatbot_id': self.chatbot_id,
        'turns': [turn.to_dict() for turn in self.turns],
        'created_at': self.created_at.isoformat(),
        'updated_at': self.updated_at.isoformat(),
        'metadata': self.metadata
    }

from_dict classmethod

from_dict(data: Dict[str, Any]) -> ConversationHistory

Deserialize conversation history from dictionary.

Source code in packages/ai-parrot/src/parrot/memory/abstract.py

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'ConversationHistory':
    """Deserialize conversation history from dictionary."""
    history = cls(
        session_id=data['session_id'],
        user_id=data['user_id'],
        chatbot_id=data.get('chatbot_id'),
        created_at=datetime.fromisoformat(data['created_at']),
        updated_at=datetime.fromisoformat(data['updated_at']),
        metadata=data.get('metadata', {})
    )

    for turn_data in data.get('turns', []):
        turn = ConversationTurn.from_dict(turn_data)
        history.turns.append(turn)

    return history

ConversationTurn dataclass

ConversationTurn(turn_id: str, user_id: str, user_message: str, assistant_response: str, context_used: Optional[str] = None, tools_used: List[str] = list(), timestamp: datetime = datetime.now(), metadata: Dict[str, Any] = dict())

Represents a single turn in a conversation.

to_dict

to_dict() -> Dict[str, Any]

Serialize turn to dictionary.

Source code in packages/ai-parrot/src/parrot/memory/abstract.py

def to_dict(self) -> Dict[str, Any]:
    """Serialize turn to dictionary."""
    return {
        'turn_id': self.turn_id,
        'user_id': self.user_id,
        'user_message': self.user_message,
        'assistant_response': self.assistant_response,
        'context_used': self.context_used,
        'tools_used': self.tools_used,
        'timestamp': self.timestamp.isoformat(),
        'metadata': self.metadata
    }

from_dict classmethod

from_dict(data: Dict[str, Any]) -> ConversationTurn

Deserialize turn from dictionary.

Source code in packages/ai-parrot/src/parrot/memory/abstract.py

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'ConversationTurn':
    """Deserialize turn from dictionary."""
    return cls(
        turn_id=data['turn_id'],
        user_id=data['user_id'],
        user_message=data['user_message'],
        assistant_response=data['assistant_response'],
        context_used=data.get('context_used'),
        tools_used=data.get('tools_used', []),
        timestamp=datetime.fromisoformat(data['timestamp']),
        metadata=data.get('metadata', {})
    )

InMemoryConversation

InMemoryConversation()

Bases: ConversationMemory

In-memory implementation of conversation memory.

Source code in packages/ai-parrot/src/parrot/memory/mem.py

def __init__(self):
    super().__init__()
    self._histories: Dict[str, Dict[str, Dict[str, ConversationHistory]]] = {}

create_history async

create_history(user_id: str, session_id: str, metadata: Optional[Dict[str, Any]] = None, chatbot_id: Optional[str] = None) -> ConversationHistory

Create a new conversation history.

Source code in packages/ai-parrot/src/parrot/memory/mem.py

async def create_history(
    self,
    user_id: str,
    session_id: str,
    metadata: Optional[Dict[str, Any]] = None,
    chatbot_id: Optional[str] = None
) -> ConversationHistory:
    """Create a new conversation history."""
    chatbot_key = self._get_chatbot_key(chatbot_id)
    self._histories.setdefault(user_id, {})
    self._histories[user_id].setdefault(chatbot_key, {})

    history = ConversationHistory(
        session_id=session_id,
        user_id=user_id,
        chatbot_id=chatbot_id,
        metadata=metadata or {}
    )

    self._histories[user_id][chatbot_key][session_id] = history
    return history

get_history async

get_history(user_id: str, session_id: str, chatbot_id: Optional[str] = None) -> Optional[ConversationHistory]

Get a conversation history.

Source code in packages/ai-parrot/src/parrot/memory/mem.py

async def get_history(
    self,
    user_id: str,
    session_id: str,
    chatbot_id: Optional[str] = None
) -> Optional[ConversationHistory]:
    """Get a conversation history."""
    user_histories = self._histories.get(user_id, {})
    result: Optional[ConversationHistory] = None
    if chatbot_id is not None:
        chatbot_key = self._get_chatbot_key(chatbot_id)
        result = user_histories.get(chatbot_key, {}).get(session_id)
    else:
        for histories in user_histories.values():
            if session_id in histories:
                result = histories[session_id]
                break
    if result and self.debug:
        self.logger.debug(f"DEBUG: History has {len(result.turns)} turns")
    return result

update_history async

update_history(history: ConversationHistory) -> None

Update a conversation history.

Source code in packages/ai-parrot/src/parrot/memory/mem.py

async def update_history(self, history: ConversationHistory) -> None:
    """Update a conversation history."""
    chatbot_key = self._get_chatbot_key(history.chatbot_id)
    self._histories.setdefault(history.user_id, {})
    self._histories[history.user_id].setdefault(chatbot_key, {})
    self._histories[history.user_id][chatbot_key][history.session_id] = history

add_turn async

add_turn(user_id: str, session_id: str, turn: ConversationTurn, chatbot_id: Optional[str] = None) -> None

Add a turn to the conversation.

Source code in packages/ai-parrot/src/parrot/memory/mem.py

async def add_turn(
    self,
    user_id: str,
    session_id: str,
    turn: ConversationTurn,
    chatbot_id: Optional[str] = None
) -> None:
    """Add a turn to the conversation."""
    history = await self.get_history(user_id, session_id, chatbot_id)
    if history:
        history.add_turn(turn)

clear_history async

clear_history(user_id: str, session_id: str, chatbot_id: Optional[str] = None) -> None

Clear a conversation history.

Source code in packages/ai-parrot/src/parrot/memory/mem.py

async def clear_history(
    self,
    user_id: str,
    session_id: str,
    chatbot_id: Optional[str] = None
) -> None:
    """Clear a conversation history."""
    history = await self.get_history(user_id, session_id, chatbot_id)
    if history:
        history.clear_turns()

list_sessions async

list_sessions(user_id: str, chatbot_id: Optional[str] = None) -> List[str]

List all session IDs for a user.

Source code in packages/ai-parrot/src/parrot/memory/mem.py

async def list_sessions(
    self,
    user_id: str,
    chatbot_id: Optional[str] = None
) -> List[str]:
    """List all session IDs for a user."""
    user_histories = self._histories.get(user_id, {})
    if chatbot_id is not None:
        chatbot_key = self._get_chatbot_key(chatbot_id)
        return list(user_histories.get(chatbot_key, {}).keys())
    sessions: List[str] = []
    seen = set()
    for histories in user_histories.values():
        for session in histories.keys():
            if session not in seen:
                seen.add(session)
                sessions.append(session)
    return sessions

delete_history async

delete_history(user_id: str, session_id: str, chatbot_id: Optional[str] = None) -> bool

Delete a conversation history entirely.

Source code in packages/ai-parrot/src/parrot/memory/mem.py

async def delete_history(
    self,
    user_id: str,
    session_id: str,
    chatbot_id: Optional[str] = None
) -> bool:
    """Delete a conversation history entirely."""
    if user_id not in self._histories:
        return False

    if chatbot_id is not None:
        chatbot_key = self._get_chatbot_key(chatbot_id)
        histories = self._histories[user_id].get(chatbot_key)
        if histories and session_id in histories:
            del histories[session_id]
            return True
        return False

    removed = False
    for histories in self._histories[user_id].values():
        if session_id in histories:
            del histories[session_id]
            removed = True
            break
    return removed

RedisConversation

RedisConversation(redis_url: str = None, key_prefix: str = 'conversation', use_hash_storage: bool = True)

Bases: ConversationMemory

Redis-based conversation memory with proper encoding handling.

Source code in packages/ai-parrot/src/parrot/memory/redis.py

def __init__(
    self,
    redis_url: str = None,
    key_prefix: str = "conversation",
    use_hash_storage: bool = True
):
    self.redis_url = redis_url or REDIS_HISTORY_URL
    self.key_prefix = key_prefix
    self.use_hash_storage = use_hash_storage
    self.redis = Redis.from_url(
        self.redis_url,
        decode_responses=True,
        encoding="utf-8",
        socket_connect_timeout=5,
        socket_timeout=5,
        retry_on_timeout=True
    )

create_history async

create_history(user_id: str, session_id: str, metadata: Optional[Dict[str, Any]] = None, chatbot_id: Optional[str] = None) -> ConversationHistory

Create a new conversation history.

Source code in packages/ai-parrot/src/parrot/memory/redis.py

async def create_history(
    self,
    user_id: str,
    session_id: str,
    metadata: Optional[Dict[str, Any]] = None,
    chatbot_id: Optional[str] = None,
) -> ConversationHistory:
    """Create a new conversation history."""
    history = ConversationHistory(
        session_id=session_id,
        user_id=user_id,
        chatbot_id=chatbot_id,
        metadata=metadata or {}
    )

    if self.use_hash_storage:
        # Method 1: Using Redis Hash (RECOMMENDED for objects)
        key = self._get_key(user_id, session_id, chatbot_id)
        history_dict = history.to_dict()

        # Store each field separately in a hash
        mapping = {
            'session_id': str(history_dict['session_id']),
            'user_id': str(history_dict['user_id']),
            'turns': self._serialize_data(history_dict['turns']),
            'created_at': history_dict['created_at'],
            'updated_at': history_dict['updated_at'],
            'metadata': self._serialize_data(history_dict['metadata'])
        }
        if chatbot_id:
            mapping['chatbot_id'] = str(chatbot_id)
    else:
        # Method 2: Using simple key-value storage
        key = self._get_key(user_id, session_id, chatbot_id)
        serialized_data = self._serialize_data(history.to_dict())
        await self.redis.set(key, serialized_data)

    # Add to user sessions set
    await self.redis.sadd(
        self._get_user_sessions_key(user_id, chatbot_id),
        session_id
    )
    return history

get_history async

get_history(user_id: str, session_id: str, chatbot_id: Optional[str] = None) -> Optional[ConversationHistory]

Get a conversation history.

Source code in packages/ai-parrot/src/parrot/memory/redis.py

async def get_history(
    self,
    user_id: str,
    session_id: str,
    chatbot_id: Optional[str] = None
) -> Optional[ConversationHistory]:
    """Get a conversation history."""
    key = self._get_key(user_id, session_id, chatbot_id)

    if self.use_hash_storage:
        # Method 1: Get from Redis Hash
        data = await self.redis.hgetall(key)
        if not data:
            return None

        try:
            # Reconstruct the history dict
            history_dict = {
                'session_id': data.get('session_id', session_id),
                'user_id': data.get('user_id', user_id),
                'chatbot_id': data.get('chatbot_id', chatbot_id),
                'turns': self._deserialize_data(data.get('turns', '[]')),
                'created_at': data.get('created_at', datetime.now().isoformat()),
                'updated_at': data.get('updated_at', datetime.now().isoformat()),
                'metadata': self._deserialize_data(data.get('metadata', '{}'))
            }
            return ConversationHistory.from_dict(history_dict)
        except (KeyError, ValueError) as e:
            print(f"Error deserializing conversation history: {e}")
            return None
    else:
        # Method 2: Get from simple key-value
        data = await self.redis.get(key)
        if data:
            try:
                history_dict = self._deserialize_data(data)
                if history_dict is not None and chatbot_id and not history_dict.get('chatbot_id'):
                    history_dict['chatbot_id'] = chatbot_id
                return ConversationHistory.from_dict(history_dict)
            except (ValueError, KeyError) as e:
                print(f"Error deserializing conversation history: {e}")
                return None
        return None

update_history async

update_history(history: ConversationHistory) -> None

Update a conversation history.

Source code in packages/ai-parrot/src/parrot/memory/redis.py

async def update_history(self, history: ConversationHistory) -> None:
    """Update a conversation history."""
    key = self._get_key(history.user_id, history.session_id, history.chatbot_id)

    if self.use_hash_storage:
        # Method 1: Update Redis Hash
        history_dict = history.to_dict()
        mapping = {
            'session_id': history_dict['session_id'],
            'user_id': history_dict['user_id'],
            'turns': self._serialize_data(history_dict['turns']),
            'created_at': history_dict['created_at'],
            'updated_at': history_dict['updated_at'],
            'metadata': self._serialize_data(history_dict['metadata'])
        }
        if history_dict.get('chatbot_id') is not None:
            mapping['chatbot_id'] = history_dict['chatbot_id']
        await self.redis.hset(key, mapping=mapping)
    else:
        # Method 2: Update simple key-value
        serialized_data = self._serialize_data(history.to_dict())
        await self.redis.set(key, serialized_data)

add_turn async

add_turn(user_id: str, session_id: str, turn: ConversationTurn, chatbot_id: Optional[str] = None) -> None

Add a turn to the conversation efficiently.

Source code in packages/ai-parrot/src/parrot/memory/redis.py

async def add_turn(
    self,
    user_id: str,
    session_id: str,
    turn: ConversationTurn,
    chatbot_id: Optional[str] = None
) -> None:
    """Add a turn to the conversation efficiently."""
    if self.use_hash_storage:
        # Optimized: Only update the turns field
        key = self._get_key(user_id, session_id, chatbot_id)

        # Get current turns
        current_turns_data = await self.redis.hget(key, 'turns')
        if current_turns_data:
            turns = self._deserialize_data(current_turns_data)
        else:
            turns = []

        # Add new turn
        turns.append(turn.to_dict())

        # Update only the turns and updated_at fields
        mapping = {
            'turns': self._serialize_data(turns),
            'updated_at': datetime.now().isoformat()
        }
        if chatbot_id is not None:
            mapping['chatbot_id'] = str(chatbot_id)
        await self.redis.hset(key, mapping=mapping)
    else:
        # Fallback to full history update
        history = await self.get_history(user_id, session_id, chatbot_id)
        if history:
            history.add_turn(turn)
            await self.update_history(history)

clear_history async

clear_history(user_id: str, session_id: str, chatbot_id: Optional[str] = None) -> None

Clear a conversation history.

Source code in packages/ai-parrot/src/parrot/memory/redis.py

async def clear_history(
    self,
    user_id: str,
    session_id: str,
    chatbot_id: Optional[str] = None
) -> None:
    """Clear a conversation history."""
    if self.use_hash_storage:
        # Optimized: Only clear turns
        key = self._get_key(user_id, session_id, chatbot_id)
        # Reset turns to empty list and update updated_at
        mapping = {
            'turns': self._serialize_data([]),
            'updated_at': datetime.now().isoformat()
        }
        if chatbot_id is not None:
            mapping['chatbot_id'] = str(chatbot_id)
        await self.redis.hset(key, mapping=mapping)
    else:
        history = await self.get_history(user_id, session_id, chatbot_id)
        if history:
            history.clear_turns()
            await self.update_history(history)

list_sessions async

list_sessions(user_id: str, chatbot_id: Optional[str] = None) -> List[str]

List all session IDs for a user.

Source code in packages/ai-parrot/src/parrot/memory/redis.py

async def list_sessions(
    self,
    user_id: str,
    chatbot_id: Optional[str] = None
) -> List[str]:
    """List all session IDs for a user."""
    sessions = await self.redis.smembers(
        self._get_user_sessions_key(user_id, chatbot_id)
    )
    # Since decode_responses=True, sessions should already be strings
    return list(sessions)

delete_history async

delete_history(user_id: str, session_id: str, chatbot_id: Optional[str] = None) -> bool

Delete a conversation history entirely.

Source code in packages/ai-parrot/src/parrot/memory/redis.py

async def delete_history(
    self,
    user_id: str,
    session_id: str,
    chatbot_id: Optional[str] = None
) -> bool:
    """Delete a conversation history entirely."""
    key = self._get_key(user_id, session_id, chatbot_id)
    result = await self.redis.delete(key)
    await self.redis.srem(
        self._get_user_sessions_key(user_id, chatbot_id),
        session_id
    )
    return result > 0

close async

close()

Close the Redis connection.

Source code in packages/ai-parrot/src/parrot/memory/redis.py

async def close(self):
    """Close the Redis connection."""
    try:
        await self.redis.close()
    except Exception as e:
        self.logger.error(f"Error closing Redis connection: {e}")

ping async

ping() -> bool

Test Redis connection.

Source code in packages/ai-parrot/src/parrot/memory/redis.py

async def ping(self) -> bool:
    """Test Redis connection."""
    try:
        await self.redis.ping()
        return True
    except Exception as e:
        self.logger.error(f"Error pinging Redis: {e}")
        return False

get_raw_data async

get_raw_data(user_id: str, session_id: str, chatbot_id: Optional[str] = None) -> Optional[Dict]

Get raw data from Redis for debugging.

Source code in packages/ai-parrot/src/parrot/memory/redis.py

async def get_raw_data(
    self,
    user_id: str,
    session_id: str,
    chatbot_id: Optional[str] = None
) -> Optional[Dict]:
    """Get raw data from Redis for debugging."""
    key = self._get_key(user_id, session_id, chatbot_id)

    if self.use_hash_storage:
        return await self.redis.hgetall(key)
    data = await self.redis.get(key)
    return {"raw_data": data} if data else None

debug_conversation async

debug_conversation(user_id: str, session_id: str, chatbot_id: Optional[str] = None) -> Dict[str, Any]

Debug method to inspect conversation data.

Source code in packages/ai-parrot/src/parrot/memory/redis.py

async def debug_conversation(
    self,
    user_id: str,
    session_id: str,
    chatbot_id: Optional[str] = None
) -> Dict[str, Any]:
    """Debug method to inspect conversation data."""
    raw_data = await self.get_raw_data(user_id, session_id, chatbot_id)
    history = await self.get_history(user_id, session_id, chatbot_id)

    return {
        "raw_data": raw_data,
        "parsed_history": history.to_dict() if history else None,
        "turns_count": len(history.turns) if history else 0,
        "storage_method": "hash" if self.use_hash_storage else "string"
    }

list_sessions_by_chatbot async

list_sessions_by_chatbot(chatbot_id: str, user_id: Optional[str] = None) -> List[str]

List all sessions for a specific chatbot.

PARAMETER	DESCRIPTION
chatbot_id	The chatbot identifier TYPE: str
user_id	Optional user filter TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
List[str]	List of session IDs

Source code in packages/ai-parrot/src/parrot/memory/redis.py

async def list_sessions_by_chatbot(
    self,
    chatbot_id: str,
    user_id: Optional[str] = None
) -> List[str]:
    """List all sessions for a specific chatbot.

    Args:
        chatbot_id: The chatbot identifier
        user_id: Optional user filter

    Returns:
        List of session IDs
    """
    if user_id:
        # Get sessions for specific user and chatbot
        return await self.list_sessions(user_id, chatbot_id)

    # Get all sessions for this chatbot across all users
    pattern = f"{self.key_prefix}:{chatbot_id}:*"
    sessions = []
    cursor = 0

    while True:
        cursor, keys = await self.redis.scan(
            cursor,
            match=pattern,
            count=100
        )
        for key in keys:
            # Extract session_id from key
            # Format: conversation:chatbot_id:user_id:session_id
            parts = key.split(':')
            if len(parts) >= 4:
                sessions.append(parts[3])

        if cursor == 0:
            break

    return sessions

get_chatbot_stats async

get_chatbot_stats(chatbot_id: str) -> Dict[str, Any]

Get statistics for a specific chatbot.

RETURNS	DESCRIPTION
Dict[str, Any]	Dictionary with conversation counts, active users, etc.

Source code in packages/ai-parrot/src/parrot/memory/redis.py

async def get_chatbot_stats(self, chatbot_id: str) -> Dict[str, Any]:
    """Get statistics for a specific chatbot.

    Returns:
        Dictionary with conversation counts, active users, etc.
    """
    pattern = f"{self.key_prefix}:{chatbot_id}:*"
    total_conversations = 0
    total_turns = 0
    unique_users = set()
    cursor = 0

    while True:
        cursor, keys = await self.redis.scan(
            cursor,
            match=pattern,
            count=100
        )
        total_conversations += len(keys)

        for key in keys:
            # Extract user_id
            parts = key.split(':')
            if len(parts) >= 3:
                unique_users.add(parts[2])

            # Count turns
            if self.use_hash_storage:
                turns_data = await self.redis.hget(key, 'turns')
                if turns_data:
                    turns = self._deserialize_data(turns_data)
                    total_turns += len(turns)

        if cursor == 0:
            break

    return {
        'chatbot_id': chatbot_id,
        'total_conversations': total_conversations,
        'total_turns': total_turns,
        'unique_users': len(unique_users),
        'avg_turns_per_conversation': total_turns / total_conversations if total_conversations > 0 else 0
    }

delete_all_chatbot_conversations async

delete_all_chatbot_conversations(chatbot_id: str, user_id: Optional[str] = None) -> int

Delete all conversations for a chatbot.

PARAMETER	DESCRIPTION
chatbot_id	The chatbot identifier TYPE: str
user_id	Optional user filter TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
int	Number of conversations deleted

Source code in packages/ai-parrot/src/parrot/memory/redis.py

async def delete_all_chatbot_conversations(
    self,
    chatbot_id: str,
    user_id: Optional[str] = None
) -> int:
    """Delete all conversations for a chatbot.

    Args:
        chatbot_id: The chatbot identifier
        user_id: Optional user filter

    Returns:
        Number of conversations deleted
    """
    if user_id:
        pattern = f"{self.key_prefix}:{chatbot_id}:{user_id}:*"
    else:
        pattern = f"{self.key_prefix}:{chatbot_id}:*"

    deleted_count = 0
    cursor = 0

    while True:
        cursor, keys = await self.redis.scan(
            cursor,
            match=pattern,
            count=100
        )

        if keys:
            deleted_count += await self.redis.delete(*keys)

        if cursor == 0:
            break

    # Also clean up session sets
    if user_id:
        await self.redis.delete(self._get_user_sessions_key(user_id, chatbot_id))
    else:
        # Clean all user session sets for this chatbot
        session_pattern = f"{self.key_prefix}_sessions:{chatbot_id}:*"
        cursor = 0
        while True:
            cursor, session_keys = await self.redis.scan(
                cursor,
                match=session_pattern,
                count=100
            )
            if session_keys:
                await self.redis.delete(*session_keys)
            if cursor == 0:
                break

    return deleted_count

get_chatbot_users async

get_chatbot_users(chatbot_id: str) -> List[str]

Get all users who have interacted with a chatbot.

Source code in packages/ai-parrot/src/parrot/memory/redis.py

async def get_chatbot_users(self, chatbot_id: str) -> List[str]:
    """Get all users who have interacted with a chatbot."""
    users_key = f"{self.key_prefix}_index:chatbot_users:{chatbot_id}"
    users = await self.redis.smembers(users_key)
    return list(users)

FileConversationMemory

FileConversationMemory(base_path: str = './conversations')

Bases: ConversationMemory

File-based implementation of conversation memory.

Source code in packages/ai-parrot/src/parrot/memory/file.py

def __init__(self, base_path: str = "./conversations"):
    self.base_path = Path(base_path)
    self.base_path.mkdir(exist_ok=True)
    self._lock = asyncio.Lock()

create_history async

create_history(user_id: str, session_id: str, metadata: Optional[Dict[str, Any]] = None, chatbot_id: Optional[str] = None) -> ConversationHistory

Create a new conversation history.

Source code in packages/ai-parrot/src/parrot/memory/file.py

async def create_history(
    self,
    user_id: str,
    session_id: str,
    metadata: Optional[Dict[str, Any]] = None,
    chatbot_id: Optional[str] = None
) -> ConversationHistory:
    """Create a new conversation history."""
    async with self._lock:
        history = ConversationHistory(
            session_id=session_id,
            user_id=user_id,
            chatbot_id=chatbot_id,
            metadata=metadata or {}
        )

        file_path = self._get_file_path(user_id, session_id, chatbot_id)
        with open(file_path, 'w', encoding='utf-8') as f:
            json.dump(history.to_dict(), f, indent=2, ensure_ascii=False, default=str)

        return history

get_history async

get_history(user_id: str, session_id: str, chatbot_id: Optional[str] = None) -> Optional[ConversationHistory]

Get a conversation history.

Source code in packages/ai-parrot/src/parrot/memory/file.py

async def get_history(
    self,
    user_id: str,
    session_id: str,
    chatbot_id: Optional[str] = None
) -> Optional[ConversationHistory]:
    """Get a conversation history."""
    async with self._lock:
        file_path = self._get_file_path(user_id, session_id, chatbot_id)
        if not file_path.exists():
            return None

        try:
            async with aiofiles.open(file_path, 'r', encoding='utf-8') as f:
                content = await f.read()
            data = json.loads(content)
            return ConversationHistory.from_dict(data)
        except (TypeError, KeyError, ValueError):
            return None

update_history async

update_history(history: ConversationHistory) -> None

Update a conversation history.

Source code in packages/ai-parrot/src/parrot/memory/file.py

async def update_history(self, history: ConversationHistory) -> None:
    """Update a conversation history."""
    async with self._lock:
        file_path = self._get_file_path(
            history.user_id,
            history.session_id,
            history.chatbot_id
        )
        async with aiofiles.open(file_path, 'w', encoding='utf-8') as f:
            await f.write(json.dumps(history.to_dict(), indent=2, ensure_ascii=False, default=str))

add_turn async

add_turn(user_id: str, session_id: str, turn: ConversationTurn, chatbot_id: Optional[str] = None) -> None

Add a turn to the conversation.

Source code in packages/ai-parrot/src/parrot/memory/file.py

async def add_turn(
    self,
    user_id: str,
    session_id: str,
    turn: ConversationTurn,
    chatbot_id: Optional[str] = None
) -> None:
    """Add a turn to the conversation."""
    history = await self.get_history(user_id, session_id, chatbot_id)
    if history:
        history.add_turn(turn)
        await self.update_history(history)

clear_history async

clear_history(user_id: str, session_id: str, chatbot_id: Optional[str] = None) -> None

Clear a conversation history.

Source code in packages/ai-parrot/src/parrot/memory/file.py

async def clear_history(
    self,
    user_id: str,
    session_id: str,
    chatbot_id: Optional[str] = None
) -> None:
    """Clear a conversation history."""
    history = await self.get_history(user_id, session_id, chatbot_id)
    if history:
        history.clear_turns()
        await self.update_history(history)

list_sessions async

list_sessions(user_id: str, chatbot_id: Optional[str] = None) -> List[str]

List all session IDs for a user.

Source code in packages/ai-parrot/src/parrot/memory/file.py

async def list_sessions(
    self,
    user_id: str,
    chatbot_id: Optional[str] = None
) -> List[str]:
    """List all session IDs for a user."""
    async with self._lock:
        base_user_dir = self.base_path / str(user_id)
        if not base_user_dir.exists():
            return []

        sessions: List[str] = []
        seen = set()
        if chatbot_id is None:
            for file_path in base_user_dir.glob("*.json"):
                if file_path.stem not in seen:
                    seen.add(file_path.stem)
                    sessions.append(file_path.stem)
            for subdir in base_user_dir.iterdir():
                if subdir.is_dir():
                    for file_path in subdir.glob("*.json"):
                        if file_path.stem not in seen:
                            seen.add(file_path.stem)
                            sessions.append(file_path.stem)
        else:
            chatbot_dir = base_user_dir / str(chatbot_id)
            if chatbot_dir.exists():
                for file_path in chatbot_dir.glob("*.json"):
                    if file_path.stem not in seen:
                        seen.add(file_path.stem)
                        sessions.append(file_path.stem)

        return sessions

delete_history async

delete_history(user_id: str, session_id: str, chatbot_id: Optional[str] = None) -> bool

Delete a conversation history entirely.

Source code in packages/ai-parrot/src/parrot/memory/file.py

async def delete_history(
    self,
    user_id: str,
    session_id: str,
    chatbot_id: Optional[str] = None
) -> bool:
    """Delete a conversation history entirely."""
    async with self._lock:
        file_path = self._get_file_path(user_id, session_id, chatbot_id)
        if file_path.exists():
            file_path.unlink()
            return True
        return False

AnswerMemory

AnswerMemory(agent_id: str)

Store and retrieve question/answer interactions by turn identifier.

Source code in packages/ai-parrot/src/parrot/memory/agent.py

def __init__(self, agent_id: str) -> None:
    self.agent_id = agent_id
    self._interactions: Dict[str, Dict[str, Any]] = {self.agent_id: {}}
    self._lock = asyncio.Lock()

store_interaction async

store_interaction(turn_id: str, question: str, answer: Any) -> None

Persist a question/answer pair under the provided turn identifier.

Source code in packages/ai-parrot/src/parrot/memory/agent.py

async def store_interaction(self, turn_id: str, question: str, answer: Any) -> None:
    """Persist a question/answer pair under the provided turn identifier."""
    if not turn_id:
        raise ValueError("turn_id is required to store an interaction")

    async with self._lock:
        self._interactions[self.agent_id][turn_id] = {
            "question": question,
            "answer": answer,
        }

get async

get(turn_id: str) -> Optional[Dict[str, Any]]

Retrieve a stored interaction by turn identifier.

Source code in packages/ai-parrot/src/parrot/memory/agent.py

async def get(self, turn_id: str) -> Optional[Dict[str, Any]]:
    """Retrieve a stored interaction by turn identifier."""
    if not turn_id:
        return None

    async with self._lock:
        return self._interactions.get(self.agent_id, {}).get(turn_id)

EpisodicMemoryMixin

Mixin that adds automatic episodic memory to bots.

Provides hooks that bot implementations call at appropriate points in their ask() flow. The mixin is opt-in — bots that don't inherit it are completely unaffected.

Configuration attributes (override in subclass or set via kwargs): enable_episodic_memory: Master toggle for the mixin. episodic_backend: Backend type ("pgvector" or "faiss"). episodic_dsn: PostgreSQL DSN for PgVector backend. episodic_faiss_path: Persistence path for FAISS backend. episodic_schema: PostgreSQL schema name. episodic_reflection_enabled: Whether to generate reflections. episodic_inject_warnings: Whether to inject failure warnings pre-LLM. episodic_max_warnings: Maximum warnings to inject. episodic_trivial_tools: Tools to skip when recording.

EpisodicMemoryStore

EpisodicMemoryStore(backend: AbstractEpisodeBackend, embedding_provider: EpisodeEmbeddingProvider | None = None, reflection_engine: ReflectionEngine | None = None, redis_cache: EpisodeRedisCache | None = None, default_ttl_days: int = 90, importance_scorer: ImportanceScorer | None = None, recall_strategy: RecallStrategy | None = None)

Main orchestrator for episodic memory operations.

Coordinates a backend (PgVector, FAISS, or RedisVector), an optional embedding provider (sentence-transformers), an optional reflection engine (LLM + heuristic), and an optional Redis cache for fast recent/failure lookups.

Pluggable strategies: - importance_scorer: When provided, used to compute episode importance in record_episode() instead of the inline heuristic. Must satisfy the ImportanceScorer protocol. - recall_strategy: When provided, used in recall_similar() instead of calling backend.search_similar() directly. Must satisfy the RecallStrategy protocol.

When neither is provided, behavior is identical to the pre-FEAT-075 implementation (no breaking changes).

PARAMETER	DESCRIPTION
backend	Storage backend (PgVector, FAISS, or RedisVector). TYPE: AbstractEpisodeBackend
embedding_provider	Optional embedding provider for semantic search. TYPE: EpisodeEmbeddingProvider | None DEFAULT: None
reflection_engine	Optional reflection engine for lesson extraction. TYPE: ReflectionEngine | None DEFAULT: None
redis_cache	Optional Redis cache for hot episodes. TYPE: EpisodeRedisCache | None DEFAULT: None
default_ttl_days	Default time-to-live for episodes (0 = no expiry). TYPE: int DEFAULT: 90
importance_scorer	Optional pluggable importance scorer. TYPE: ImportanceScorer | None DEFAULT: None
recall_strategy	Optional pluggable recall strategy. TYPE: RecallStrategy | None DEFAULT: None

Source code in packages/ai-parrot/src/parrot/memory/episodic/store.py

def __init__(
    self,
    backend: AbstractEpisodeBackend,
    embedding_provider: EpisodeEmbeddingProvider | None = None,
    reflection_engine: ReflectionEngine | None = None,
    redis_cache: EpisodeRedisCache | None = None,
    default_ttl_days: int = 90,
    importance_scorer: ImportanceScorer | None = None,
    recall_strategy: RecallStrategy | None = None,
) -> None:
    self._backend = backend
    self._embedding = embedding_provider
    self._reflection = reflection_engine
    self._cache = redis_cache
    self._default_ttl_days = default_ttl_days
    self._importance_scorer = importance_scorer
    self._recall_strategy = recall_strategy

record_episode async

record_episode(namespace: MemoryNamespace, situation: str, action_taken: str, outcome: EpisodeOutcome, outcome_details: str | None = None, error_type: str | None = None, error_message: str | None = None, category: EpisodeCategory = EpisodeCategory.TOOL_EXECUTION, importance: int | None = None, related_tools: list[str] | None = None, related_entities: list[str] | None = None, metadata: dict[str, Any] | None = None, generate_reflection: bool = True, ttl_days: int | None = None) -> EpisodicMemory

Record a new episode with auto-enrichment.

Auto-computes importance, generates reflection (if engine available), embeds text for similarity search, stores in backend, and caches.

PARAMETER	DESCRIPTION
namespace	Scoping dimensions for this episode. TYPE: MemoryNamespace
situation	What the agent was trying to do. TYPE: str
action_taken	What action was executed. TYPE: str
outcome	The result classification. TYPE: EpisodeOutcome
outcome_details	Additional outcome description. TYPE: str | None DEFAULT: None
error_type	Error category if applicable. TYPE: str | None DEFAULT: None
error_message	Detailed error message. TYPE: str | None DEFAULT: None
category	Episode classification. TYPE: EpisodeCategory DEFAULT: TOOL_EXECUTION
importance	Override auto-computed importance (1-10). TYPE: int | None DEFAULT: None
related_tools	Tools involved in this episode. TYPE: list[str] | None DEFAULT: None
related_entities	Entities referenced. TYPE: list[str] | None DEFAULT: None
metadata	Additional key-value metadata. TYPE: dict[str, Any] | None DEFAULT: None
generate_reflection	Whether to generate reflection via engine. TYPE: bool DEFAULT: True
ttl_days	Override default TTL. 0 = no expiry. TYPE: int | None DEFAULT: None

RETURNS	DESCRIPTION
EpisodicMemory	The stored EpisodicMemory with all enrichments.

Source code in packages/ai-parrot/src/parrot/memory/episodic/store.py

async def record_episode(
    self,
    namespace: MemoryNamespace,
    situation: str,
    action_taken: str,
    outcome: EpisodeOutcome,
    outcome_details: str | None = None,
    error_type: str | None = None,
    error_message: str | None = None,
    category: EpisodeCategory = EpisodeCategory.TOOL_EXECUTION,
    importance: int | None = None,
    related_tools: list[str] | None = None,
    related_entities: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    generate_reflection: bool = True,
    ttl_days: int | None = None,
) -> EpisodicMemory:
    """Record a new episode with auto-enrichment.

    Auto-computes importance, generates reflection (if engine available),
    embeds text for similarity search, stores in backend, and caches.

    Args:
        namespace: Scoping dimensions for this episode.
        situation: What the agent was trying to do.
        action_taken: What action was executed.
        outcome: The result classification.
        outcome_details: Additional outcome description.
        error_type: Error category if applicable.
        error_message: Detailed error message.
        category: Episode classification.
        importance: Override auto-computed importance (1-10).
        related_tools: Tools involved in this episode.
        related_entities: Entities referenced.
        metadata: Additional key-value metadata.
        generate_reflection: Whether to generate reflection via engine.
        ttl_days: Override default TTL. 0 = no expiry.

    Returns:
        The stored EpisodicMemory with all enrichments.
    """
    # Auto-compute importance (deferred until after episode is built if scorer)
    if importance is None and self._importance_scorer is None:
        importance = _auto_importance(outcome, error_type)
    elif importance is None:
        # Use inline heuristic as default; scorer overrides after build
        importance = _auto_importance(outcome, error_type)

    is_failure = outcome in (EpisodeOutcome.FAILURE, EpisodeOutcome.TIMEOUT)

    # Compute TTL
    effective_ttl = ttl_days if ttl_days is not None else self._default_ttl_days
    expires_at = None
    if effective_ttl > 0:
        expires_at = datetime.now(timezone.utc) + timedelta(days=effective_ttl)

    # Generate reflection
    reflection = None
    lesson_learned = None
    suggested_action = None
    if generate_reflection and self._reflection is not None:
        try:
            result = await self._reflection.reflect(
                situation, action_taken, outcome, error_message
            )
            reflection = result.reflection
            lesson_learned = result.lesson_learned
            suggested_action = result.suggested_action
        except Exception as e:
            logger.warning("Reflection generation failed: %s", e)

    # Build episode
    episode = EpisodicMemory(
        tenant_id=namespace.tenant_id,
        agent_id=namespace.agent_id,
        user_id=namespace.user_id,
        session_id=namespace.session_id,
        room_id=namespace.room_id,
        crew_id=namespace.crew_id,
        situation=situation,
        action_taken=action_taken,
        outcome=outcome,
        outcome_details=outcome_details,
        error_type=error_type,
        error_message=error_message,
        reflection=reflection,
        lesson_learned=lesson_learned,
        suggested_action=suggested_action,
        category=category,
        importance=importance,
        is_failure=is_failure,
        related_tools=related_tools or [],
        related_entities=related_entities or [],
        expires_at=expires_at,
        metadata=metadata or {},
    )

    # Apply importance scorer if provided (overrides inline heuristic)
    if self._importance_scorer is not None:
        try:
            raw_score = self._importance_scorer.score(episode)
            # Normalize [0.0, 1.0] → [1, 10] integer scale
            episode.importance = max(1, min(10, round(raw_score * 10)))
        except Exception as e:
            logger.warning("ImportanceScorer.score() failed: %s", e)

    # Generate embedding
    if self._embedding is not None:
        try:
            text = EpisodeEmbeddingProvider.get_searchable_text(episode)
            episode.embedding = await self._embedding.embed(text)
        except Exception as e:
            logger.warning("Embedding generation failed: %s", e)

    # Store in backend
    await self._backend.store(episode)

    # Cache in Redis
    if self._cache is not None:
        await self._cache.cache_episode(namespace, episode)

    logger.debug(
        "Recorded episode %s: %s → %s",
        episode.episode_id,
        category.value,
        outcome.value,
    )
    return episode

record_tool_episode async

record_tool_episode(namespace: MemoryNamespace, tool_name: str, tool_args: dict[str, Any], tool_result: Any, user_query: str | None = None) -> EpisodicMemory

Record an episode from a tool execution.

Extracts episode fields from a ToolResult and delegates to record_episode().

PARAMETER	DESCRIPTION
namespace	Scoping dimensions. TYPE: MemoryNamespace
tool_name	Name of the tool that was called. TYPE: str
tool_args	Arguments passed to the tool. TYPE: dict[str, Any]
tool_result	The ToolResult from the tool execution. TYPE: Any
user_query	The original user query that triggered the tool. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
EpisodicMemory	The stored episode.

Source code in packages/ai-parrot/src/parrot/memory/episodic/store.py

async def record_tool_episode(
    self,
    namespace: MemoryNamespace,
    tool_name: str,
    tool_args: dict[str, Any],
    tool_result: Any,
    user_query: str | None = None,
) -> EpisodicMemory:
    """Record an episode from a tool execution.

    Extracts episode fields from a ToolResult and delegates to
    record_episode().

    Args:
        namespace: Scoping dimensions.
        tool_name: Name of the tool that was called.
        tool_args: Arguments passed to the tool.
        tool_result: The ToolResult from the tool execution.
        user_query: The original user query that triggered the tool.

    Returns:
        The stored episode.
    """
    # Build situation
    situation = user_query or f"Tool execution: {tool_name}"

    # Summarize args (truncate long values)
    summarized = {
        k: str(v)[:100] for k, v in (tool_args or {}).items()
    }
    action_taken = f"Called {tool_name}({json.dumps(summarized)})"

    # Map outcome
    success = getattr(tool_result, "success", True)
    error = getattr(tool_result, "error", None)
    status = getattr(tool_result, "status", "success")

    if not success:
        outcome = EpisodeOutcome.FAILURE
    elif status == "timeout":
        outcome = EpisodeOutcome.TIMEOUT
    elif status == "partial":
        outcome = EpisodeOutcome.PARTIAL
    else:
        outcome = EpisodeOutcome.SUCCESS

    # Extract error details
    error_type = None
    error_message = None
    if error:
        error_message = str(error)[:500]
        # Try to classify error type
        error_lower = error_message.lower()
        for etype in _KNOWN_ERROR_TYPES:
            if etype in error_lower:
                error_type = etype
                break

    outcome_details = None
    result_val = getattr(tool_result, "result", None)
    if result_val is not None:
        outcome_details = str(result_val)[:200]

    return await self.record_episode(
        namespace=namespace,
        situation=situation,
        action_taken=action_taken,
        outcome=outcome,
        outcome_details=outcome_details,
        error_type=error_type,
        error_message=error_message,
        category=EpisodeCategory.TOOL_EXECUTION,
        related_tools=[tool_name],
    )

record_crew_episode async

record_crew_episode(namespace: MemoryNamespace, crew_result: Any, flow_description: str, per_agent: bool = True) -> list[EpisodicMemory]

Record episodes from a crew execution.

Creates a crew-level episode and optionally per-agent episodes.

PARAMETER	DESCRIPTION
namespace	Scoping dimensions (should include crew_id). TYPE: MemoryNamespace
crew_result	The result from AgentCrew execution. TYPE: Any
flow_description	Description of the workflow. TYPE: str
per_agent	If True, create one episode per participating agent. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
list[EpisodicMemory]	List of all created episodes.

Source code in packages/ai-parrot/src/parrot/memory/episodic/store.py

async def record_crew_episode(
    self,
    namespace: MemoryNamespace,
    crew_result: Any,
    flow_description: str,
    per_agent: bool = True,
) -> list[EpisodicMemory]:
    """Record episodes from a crew execution.

    Creates a crew-level episode and optionally per-agent episodes.

    Args:
        namespace: Scoping dimensions (should include crew_id).
        crew_result: The result from AgentCrew execution.
        flow_description: Description of the workflow.
        per_agent: If True, create one episode per participating agent.

    Returns:
        List of all created episodes.
    """
    episodes = []

    # Crew-level episode
    success = getattr(crew_result, "success", True)
    outcome = EpisodeOutcome.SUCCESS if success else EpisodeOutcome.FAILURE
    error = getattr(crew_result, "error", None)

    crew_ep = await self.record_episode(
        namespace=namespace,
        situation=f"Crew execution: {flow_description}",
        action_taken=f"Ran crew workflow with agents",
        outcome=outcome,
        error_message=str(error)[:500] if error else None,
        category=EpisodeCategory.WORKFLOW_PATTERN,
        importance=6,
    )
    episodes.append(crew_ep)

    # Per-agent episodes
    if per_agent:
        agent_results = getattr(crew_result, "agent_results", None) or {}
        for agent_id, agent_result in agent_results.items():
            agent_success = getattr(agent_result, "success", True)
            agent_outcome = (
                EpisodeOutcome.SUCCESS if agent_success else EpisodeOutcome.FAILURE
            )
            agent_ns = MemoryNamespace(
                tenant_id=namespace.tenant_id,
                agent_id=agent_id,
                crew_id=namespace.crew_id,
            )
            agent_error = getattr(agent_result, "error", None)
            agent_ep = await self.record_episode(
                namespace=agent_ns,
                situation=f"Agent participation in crew: {flow_description}",
                action_taken=f"Executed assigned task in crew workflow",
                outcome=agent_outcome,
                error_message=str(agent_error)[:500] if agent_error else None,
                category=EpisodeCategory.WORKFLOW_PATTERN,
                importance=5,
            )
            episodes.append(agent_ep)

    return episodes

recall_similar async

recall_similar(query: str, namespace: MemoryNamespace, top_k: int = 5, score_threshold: float = 0.3, category: EpisodeCategory | None = None, include_failures_only: bool = False) -> list[EpisodeSearchResult]

Recall episodes similar to a query.

Embeds the query and performs vector similarity search with namespace filtering.

PARAMETER	DESCRIPTION
query	Natural language query to search for. TYPE: str
namespace	Scoping dimensions for filtering. TYPE: MemoryNamespace
top_k	Maximum results. TYPE: int DEFAULT: 5
score_threshold	Minimum similarity score (0-1). TYPE: float DEFAULT: 0.3
category	Optional category filter (applied post-search). TYPE: EpisodeCategory | None DEFAULT: None
include_failures_only	If True, only return failure episodes. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
list[EpisodeSearchResult]	List of episodes ranked by similarity score.

Source code in packages/ai-parrot/src/parrot/memory/episodic/store.py

async def recall_similar(
    self,
    query: str,
    namespace: MemoryNamespace,
    top_k: int = 5,
    score_threshold: float = 0.3,
    category: EpisodeCategory | None = None,
    include_failures_only: bool = False,
) -> list[EpisodeSearchResult]:
    """Recall episodes similar to a query.

    Embeds the query and performs vector similarity search with
    namespace filtering.

    Args:
        query: Natural language query to search for.
        namespace: Scoping dimensions for filtering.
        top_k: Maximum results.
        score_threshold: Minimum similarity score (0-1).
        category: Optional category filter (applied post-search).
        include_failures_only: If True, only return failure episodes.

    Returns:
        List of episodes ranked by similarity score.
    """
    if self._embedding is None:
        logger.warning("No embedding provider; recall_similar unavailable")
        return []

    embedding = await self._embedding.embed(query)
    ns_filter = namespace.build_filter()
    effective_top_k = top_k * 2 if category else top_k  # over-fetch for post-filter

    if self._recall_strategy is not None:
        # Use pluggable recall strategy
        results = await self._recall_strategy.search(
            query=query,
            query_embedding=embedding,
            backend=self._backend,
            namespace_filter=ns_filter,
            top_k=effective_top_k,
            score_threshold=score_threshold,
            include_failures_only=include_failures_only,
        )
    else:
        # Default: direct backend call (unchanged behavior)
        results = await self._backend.search_similar(
            embedding=embedding,
            namespace_filter=ns_filter,
            top_k=effective_top_k,
            score_threshold=score_threshold,
            include_failures_only=include_failures_only,
        )

    if category is not None:
        results = [r for r in results if r.category == category]

    return results[:top_k]

get_failure_warnings async

get_failure_warnings(namespace: MemoryNamespace, current_query: str | None = None, max_warnings: int = 5) -> str

Generate injectable warning text from past failures.

Combines semantically similar failures (if query provided) with recent failures, deduplicates, and formats as text suitable for system prompt injection.

PARAMETER	DESCRIPTION
namespace	Scoping dimensions. TYPE: MemoryNamespace
current_query	Current user query for semantic matching. TYPE: str | None DEFAULT: None
max_warnings	Maximum number of warnings. TYPE: int DEFAULT: 5

RETURNS	DESCRIPTION
str	Formatted warning text (empty string if no failures found).

Source code in packages/ai-parrot/src/parrot/memory/episodic/store.py

async def get_failure_warnings(
    self,
    namespace: MemoryNamespace,
    current_query: str | None = None,
    max_warnings: int = 5,
) -> str:
    """Generate injectable warning text from past failures.

    Combines semantically similar failures (if query provided)
    with recent failures, deduplicates, and formats as text
    suitable for system prompt injection.

    Args:
        namespace: Scoping dimensions.
        current_query: Current user query for semantic matching.
        max_warnings: Maximum number of warnings.

    Returns:
        Formatted warning text (empty string if no failures found).
    """
    failures: dict[str, EpisodicMemory] = {}

    # Semantic search for similar failures
    if current_query and self._embedding is not None:
        try:
            similar = await self.recall_similar(
                query=current_query,
                namespace=namespace,
                top_k=max_warnings,
                include_failures_only=True,
            )
            for ep in similar:
                failures[ep.episode_id] = ep
        except Exception as e:
            logger.warning("Failure recall failed: %s", e)

    # Recent failures from backend
    try:
        recent_failures = await self._backend.get_failures(
            agent_id=namespace.agent_id,
            tenant_id=namespace.tenant_id,
            limit=max_warnings,
        )
        for ep in recent_failures:
            if ep.episode_id not in failures:
                failures[ep.episode_id] = ep
    except Exception as e:
        logger.warning("get_failures failed: %s", e)

    if not failures:
        return ""

    # Sort by importance DESC, then recency
    sorted_failures = sorted(
        failures.values(),
        key=lambda ep: (ep.importance, ep.created_at.timestamp()),
        reverse=True,
    )[:max_warnings]

    # Format warnings
    lines_mistakes = []
    lines_success = []

    for ep in sorted_failures:
        tool_info = f" (tool: {', '.join(ep.related_tools)})" if ep.related_tools else ""
        if ep.lesson_learned:
            lines_mistakes.append(
                f"- {ep.situation[:100]}{tool_info} — {ep.lesson_learned}"
            )
        elif ep.error_message:
            lines_mistakes.append(
                f"- {ep.situation[:100]}{tool_info} — {ep.error_message[:100]}"
            )
        if ep.suggested_action:
            lines_success.append(f"- {ep.suggested_action}")

    parts = []
    if lines_mistakes:
        parts.append("MISTAKES TO AVOID:")
        parts.extend(lines_mistakes)
    if lines_success:
        parts.append("SUGGESTED APPROACHES:")
        parts.extend(lines_success)

    return "\n".join(parts)

get_user_preferences async

get_user_preferences(namespace: MemoryNamespace, limit: int = 10) -> list[EpisodicMemory]

Get user preference episodes.

PARAMETER	DESCRIPTION
namespace	Must include user_id for meaningful results. TYPE: MemoryNamespace
limit	Maximum preferences to return. TYPE: int DEFAULT: 10

RETURNS	DESCRIPTION
list[EpisodicMemory]	List of USER_PREFERENCE category episodes.

Source code in packages/ai-parrot/src/parrot/memory/episodic/store.py

async def get_user_preferences(
    self,
    namespace: MemoryNamespace,
    limit: int = 10,
) -> list[EpisodicMemory]:
    """Get user preference episodes.

    Args:
        namespace: Must include user_id for meaningful results.
        limit: Maximum preferences to return.

    Returns:
        List of USER_PREFERENCE category episodes.
    """
    ns_filter = namespace.build_filter()
    ns_filter["category"] = EpisodeCategory.USER_PREFERENCE.value

    return await self._backend.get_recent(
        namespace_filter=ns_filter,
        limit=limit,
    )

get_room_context async

get_room_context(namespace: MemoryNamespace, limit: int = 10, categories: list[EpisodeCategory] | None = None) -> list[EpisodicMemory]

Get recent episodes for a room.

PARAMETER	DESCRIPTION
namespace	Must include room_id for room scoping. TYPE: MemoryNamespace
limit	Maximum episodes to return. TYPE: int DEFAULT: 10
categories	Optional category filter (applied post-retrieval). TYPE: list[EpisodeCategory] | None DEFAULT: None

RETURNS	DESCRIPTION
list[EpisodicMemory]	List of recent room-scoped episodes.

Source code in packages/ai-parrot/src/parrot/memory/episodic/store.py

async def get_room_context(
    self,
    namespace: MemoryNamespace,
    limit: int = 10,
    categories: list[EpisodeCategory] | None = None,
) -> list[EpisodicMemory]:
    """Get recent episodes for a room.

    Args:
        namespace: Must include room_id for room scoping.
        limit: Maximum episodes to return.
        categories: Optional category filter (applied post-retrieval).

    Returns:
        List of recent room-scoped episodes.
    """
    ns_filter = namespace.build_filter()

    episodes = await self._backend.get_recent(
        namespace_filter=ns_filter,
        limit=limit * 2 if categories else limit,
    )

    if categories:
        cat_values = {c.value for c in categories}
        episodes = [
            ep for ep in episodes if ep.category.value in cat_values
        ]

    return episodes[:limit]

cleanup_expired async

cleanup_expired() -> int

Delete expired episodes from the backend.

RETURNS	DESCRIPTION
int	Number of episodes deleted.

Source code in packages/ai-parrot/src/parrot/memory/episodic/store.py

async def cleanup_expired(self) -> int:
    """Delete expired episodes from the backend.

    Returns:
        Number of episodes deleted.
    """
    count = await self._backend.delete_expired()
    if count > 0:
        logger.info("Cleaned up %d expired episodes", count)
    return count

compact_namespace async

compact_namespace(namespace: MemoryNamespace, keep_top_n: int = 100, keep_all_failures: bool = True) -> int

Compact a namespace by keeping only the most important episodes.

Retains the top-N episodes by importance score, plus all failure episodes if keep_all_failures is True. Deletes the rest.

PARAMETER	DESCRIPTION
namespace	The namespace to compact. TYPE: MemoryNamespace
keep_top_n	Number of top episodes to retain. TYPE: int DEFAULT: 100
keep_all_failures	If True, retain all failure episodes. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
int	Number of episodes deleted.

Source code in packages/ai-parrot/src/parrot/memory/episodic/store.py

async def compact_namespace(
    self,
    namespace: MemoryNamespace,
    keep_top_n: int = 100,
    keep_all_failures: bool = True,
) -> int:
    """Compact a namespace by keeping only the most important episodes.

    Retains the top-N episodes by importance score, plus all failure
    episodes if keep_all_failures is True. Deletes the rest.

    Args:
        namespace: The namespace to compact.
        keep_top_n: Number of top episodes to retain.
        keep_all_failures: If True, retain all failure episodes.

    Returns:
        Number of episodes deleted.
    """
    ns_filter = namespace.build_filter()

    # Get all episodes in namespace
    total = await self._backend.count(ns_filter)
    if total <= keep_top_n:
        return 0

    # Get all episodes sorted by importance
    all_episodes = await self._backend.get_recent(
        namespace_filter=ns_filter,
        limit=total,
    )

    # Sort by importance DESC
    all_episodes.sort(key=lambda ep: ep.importance, reverse=True)

    # Determine which to keep
    keep_ids = set()
    for ep in all_episodes[:keep_top_n]:
        keep_ids.add(ep.episode_id)

    if keep_all_failures:
        for ep in all_episodes:
            if ep.is_failure:
                keep_ids.add(ep.episode_id)

    # Delete the rest (not directly supported by backend protocol,
    # so we rebuild by deleting individually if the backend supports it,
    # or log a warning)
    deleted = 0
    for ep in all_episodes:
        if ep.episode_id not in keep_ids:
            # Mark as expired for next cleanup
            ep.expires_at = datetime.now(timezone.utc) - timedelta(seconds=1)
            deleted += 1

    # Trigger cleanup to remove the newly-expired episodes
    if deleted > 0:
        actual = await self._backend.delete_expired()
        logger.info(
            "Compacted namespace %s: marked %d for deletion, cleaned %d",
            namespace.scope_label,
            deleted,
            actual,
        )

    # Invalidate cache for this namespace
    if self._cache is not None:
        await self._cache.invalidate(namespace)

    return deleted

export_episodes async

export_episodes(namespace: MemoryNamespace, limit: int = 1000) -> str

Export episodes as JSONL string.

PARAMETER	DESCRIPTION
namespace	The namespace to export. TYPE: MemoryNamespace
limit	Maximum episodes to export. TYPE: int DEFAULT: 1000

RETURNS	DESCRIPTION
str	JSONL-formatted string of episodes.

Source code in packages/ai-parrot/src/parrot/memory/episodic/store.py

async def export_episodes(
    self,
    namespace: MemoryNamespace,
    limit: int = 1000,
) -> str:
    """Export episodes as JSONL string.

    Args:
        namespace: The namespace to export.
        limit: Maximum episodes to export.

    Returns:
        JSONL-formatted string of episodes.
    """
    ns_filter = namespace.build_filter()
    episodes = await self._backend.get_recent(
        namespace_filter=ns_filter,
        limit=limit,
    )

    lines = [json.dumps(ep.to_dict()) for ep in episodes]
    return "\n".join(lines)

create_pgvector async classmethod

create_pgvector(dsn: str, schema: str = 'parrot_memory', table: str = 'episodic_memory', embedding_provider: EpisodeEmbeddingProvider | None = None, reflection_engine: ReflectionEngine | None = None, redis_cache: EpisodeRedisCache | None = None, recall_strategy: RecallStrategy | None = None, importance_scorer: ImportanceScorer | None = None, **kwargs: Any) -> 'EpisodicMemoryStore'

Create a store with PgVector backend.

PARAMETER	DESCRIPTION
dsn	PostgreSQL connection string. TYPE: str
schema	PostgreSQL schema name. TYPE: str DEFAULT: 'parrot_memory'
table	Table name. TYPE: str DEFAULT: 'episodic_memory'
embedding_provider	Optional embedding provider. TYPE: EpisodeEmbeddingProvider | None DEFAULT: None
reflection_engine	Optional reflection engine. TYPE: ReflectionEngine | None DEFAULT: None
redis_cache	Optional Redis cache. TYPE: EpisodeRedisCache | None DEFAULT: None
recall_strategy	Optional recall strategy. TYPE: RecallStrategy | None DEFAULT: None
importance_scorer	Optional importance scorer. TYPE: ImportanceScorer | None DEFAULT: None
**kwargs	Additional kwargs for EpisodicMemoryStore. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
'EpisodicMemoryStore'	Configured EpisodicMemoryStore with PgVector backend.

Source code in packages/ai-parrot/src/parrot/memory/episodic/store.py

@classmethod
async def create_pgvector(
    cls,
    dsn: str,
    schema: str = "parrot_memory",
    table: str = "episodic_memory",
    embedding_provider: EpisodeEmbeddingProvider | None = None,
    reflection_engine: ReflectionEngine | None = None,
    redis_cache: EpisodeRedisCache | None = None,
    recall_strategy: RecallStrategy | None = None,
    importance_scorer: ImportanceScorer | None = None,
    **kwargs: Any,
) -> "EpisodicMemoryStore":
    """Create a store with PgVector backend.

    Args:
        dsn: PostgreSQL connection string.
        schema: PostgreSQL schema name.
        table: Table name.
        embedding_provider: Optional embedding provider.
        reflection_engine: Optional reflection engine.
        redis_cache: Optional Redis cache.
        recall_strategy: Optional recall strategy.
        importance_scorer: Optional importance scorer.
        **kwargs: Additional kwargs for EpisodicMemoryStore.

    Returns:
        Configured EpisodicMemoryStore with PgVector backend.
    """
    from .backends.pgvector import PgVectorBackend

    backend = PgVectorBackend(dsn=dsn, schema=schema, table=table)
    await backend.configure()

    return cls(
        backend=backend,
        embedding_provider=embedding_provider,
        reflection_engine=reflection_engine,
        redis_cache=redis_cache,
        recall_strategy=recall_strategy,
        importance_scorer=importance_scorer,
        **kwargs,
    )

create_redis_vector async classmethod

create_redis_vector(redis_url: str, index_name: str = 'idx:episodes', embedding_dim: int = 384, namespace: MemoryNamespace | None = None, embedding_provider: EpisodeEmbeddingProvider | None = None, reflection_engine: ReflectionEngine | None = None, redis_cache: EpisodeRedisCache | None = None, recall_strategy: RecallStrategy | None = None, importance_scorer: ImportanceScorer | None = None, **kwargs: Any) -> 'EpisodicMemoryStore'

Create a store with RedisVectorBackend.

Requires Redis Stack with RediSearch module enabled.

PARAMETER	DESCRIPTION
redis_url	Redis connection URL (e.g., redis://localhost:6379). TYPE: str
index_name	RediSearch index name. TYPE: str DEFAULT: 'idx:episodes'
embedding_dim	Dimension of embedding vectors. TYPE: int DEFAULT: 384
namespace	Optional namespace for default scoping. TYPE: MemoryNamespace | None DEFAULT: None
embedding_provider	Optional embedding provider. TYPE: EpisodeEmbeddingProvider | None DEFAULT: None
reflection_engine	Optional reflection engine. TYPE: ReflectionEngine | None DEFAULT: None
redis_cache	Optional Redis cache (separate from vector backend). TYPE: EpisodeRedisCache | None DEFAULT: None
recall_strategy	Optional recall strategy. TYPE: RecallStrategy | None DEFAULT: None
importance_scorer	Optional importance scorer. TYPE: ImportanceScorer | None DEFAULT: None
**kwargs	Additional kwargs for EpisodicMemoryStore. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
'EpisodicMemoryStore'	Configured EpisodicMemoryStore with RedisVectorBackend.

Source code in packages/ai-parrot/src/parrot/memory/episodic/store.py

@classmethod
async def create_redis_vector(
    cls,
    redis_url: str,
    index_name: str = "idx:episodes",
    embedding_dim: int = 384,
    namespace: MemoryNamespace | None = None,
    embedding_provider: EpisodeEmbeddingProvider | None = None,
    reflection_engine: ReflectionEngine | None = None,
    redis_cache: EpisodeRedisCache | None = None,
    recall_strategy: RecallStrategy | None = None,
    importance_scorer: ImportanceScorer | None = None,
    **kwargs: Any,
) -> "EpisodicMemoryStore":
    """Create a store with RedisVectorBackend.

    Requires Redis Stack with RediSearch module enabled.

    Args:
        redis_url: Redis connection URL (e.g., ``redis://localhost:6379``).
        index_name: RediSearch index name.
        embedding_dim: Dimension of embedding vectors.
        namespace: Optional namespace for default scoping.
        embedding_provider: Optional embedding provider.
        reflection_engine: Optional reflection engine.
        redis_cache: Optional Redis cache (separate from vector backend).
        recall_strategy: Optional recall strategy.
        importance_scorer: Optional importance scorer.
        **kwargs: Additional kwargs for EpisodicMemoryStore.

    Returns:
        Configured EpisodicMemoryStore with RedisVectorBackend.
    """
    from .backends.redis_vector import RedisVectorBackend

    backend = RedisVectorBackend(
        redis_url=redis_url,
        index_name=index_name,
        embedding_dim=embedding_dim,
    )
    await backend.configure()

    return cls(
        backend=backend,
        embedding_provider=embedding_provider,
        reflection_engine=reflection_engine,
        redis_cache=redis_cache,
        recall_strategy=recall_strategy,
        importance_scorer=importance_scorer,
        **kwargs,
    )

create_faiss async classmethod

create_faiss(persistence_path: str | None = None, dimension: int = 384, max_episodes: int = 10000, embedding_provider: EpisodeEmbeddingProvider | None = None, reflection_engine: ReflectionEngine | None = None, redis_cache: EpisodeRedisCache | None = None, **kwargs: Any) -> 'EpisodicMemoryStore'

Create a store with FAISS backend.

PARAMETER	DESCRIPTION
persistence_path	Directory for disk persistence (None = in-memory only). TYPE: str | None DEFAULT: None
dimension	Embedding vector dimension. TYPE: int DEFAULT: 384
max_episodes	Maximum episodes in the FAISS index. TYPE: int DEFAULT: 10000
embedding_provider	Optional embedding provider. TYPE: EpisodeEmbeddingProvider | None DEFAULT: None
reflection_engine	Optional reflection engine. TYPE: ReflectionEngine | None DEFAULT: None
redis_cache	Optional Redis cache. TYPE: EpisodeRedisCache | None DEFAULT: None
**kwargs	Additional kwargs for EpisodicMemoryStore. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
'EpisodicMemoryStore'	Configured EpisodicMemoryStore with FAISS backend.

Source code in packages/ai-parrot/src/parrot/memory/episodic/store.py

@classmethod
async def create_faiss(
    cls,
    persistence_path: str | None = None,
    dimension: int = 384,
    max_episodes: int = 10000,
    embedding_provider: EpisodeEmbeddingProvider | None = None,
    reflection_engine: ReflectionEngine | None = None,
    redis_cache: EpisodeRedisCache | None = None,
    **kwargs: Any,
) -> "EpisodicMemoryStore":
    """Create a store with FAISS backend.

    Args:
        persistence_path: Directory for disk persistence (None = in-memory only).
        dimension: Embedding vector dimension.
        max_episodes: Maximum episodes in the FAISS index.
        embedding_provider: Optional embedding provider.
        reflection_engine: Optional reflection engine.
        redis_cache: Optional Redis cache.
        **kwargs: Additional kwargs for EpisodicMemoryStore.

    Returns:
        Configured EpisodicMemoryStore with FAISS backend.
    """
    from .backends.faiss import FAISSBackend

    backend = FAISSBackend(
        dimension=dimension,
        persistence_path=persistence_path,
        max_episodes=max_episodes,
    )
    await backend.configure()

    return cls(
        backend=backend,
        embedding_provider=embedding_provider,
        reflection_engine=reflection_engine,
        redis_cache=redis_cache,
        **kwargs,
    )

EpisodicMemoryToolkit

EpisodicMemoryToolkit(store: EpisodicMemoryStore, namespace: MemoryNamespace, **kwargs)

Bases: AbstractToolkit

Toolkit exposing episodic memory as agent-callable tools.

Provides three tools for LLM agents: - search_episodic_memory: Semantic search over past experiences. - record_lesson: Explicitly record a lesson for future reference. - get_warnings: Retrieve relevant past failure warnings.

PARAMETER	DESCRIPTION
store	The EpisodicMemoryStore instance. TYPE: EpisodicMemoryStore
namespace	The namespace scope for all operations. TYPE: MemoryNamespace

Source code in packages/ai-parrot/src/parrot/memory/episodic/tools.py

def __init__(
    self,
    store: EpisodicMemoryStore,
    namespace: MemoryNamespace,
    **kwargs,
) -> None:
    self._store = store
    self._namespace = namespace
    super().__init__(**kwargs)

search_episodic_memory async

search_episodic_memory(query: str, top_k: int = 5, failures_only: bool = False) -> str

Search past agent experiences by semantic similarity.

Use this tool to recall what happened in similar past situations, including lessons learned and outcomes.

PARAMETER	DESCRIPTION
query	What to search for in past experiences. TYPE: str
top_k	Maximum number of results to return. TYPE: int DEFAULT: 5
failures_only	If True, only return past failures and mistakes. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
str	Formatted list of relevant past experiences with lessons learned.

Source code in packages/ai-parrot/src/parrot/memory/episodic/tools.py

async def search_episodic_memory(
    self,
    query: str,
    top_k: int = 5,
    failures_only: bool = False,
) -> str:
    """Search past agent experiences by semantic similarity.

    Use this tool to recall what happened in similar past situations,
    including lessons learned and outcomes.

    Args:
        query: What to search for in past experiences.
        top_k: Maximum number of results to return.
        failures_only: If True, only return past failures and mistakes.

    Returns:
        Formatted list of relevant past experiences with lessons learned.
    """
    try:
        results = await self._store.recall_similar(
            query=query,
            namespace=self._namespace,
            top_k=top_k,
            include_failures_only=failures_only,
        )

        if not results:
            return "No relevant past experiences found."

        lines = []
        for i, ep in enumerate(results, 1):
            outcome_icon = (
                "FAIL" if ep.is_failure else "OK"
            )
            line = f"{i}. [{outcome_icon}] {ep.situation[:120]}"
            line += f"\n   Action: {ep.action_taken[:120]}"
            if ep.lesson_learned:
                line += f"\n   Lesson: {ep.lesson_learned}"
            if ep.suggested_action:
                line += f"\n   Suggestion: {ep.suggested_action}"
            line += f"\n   (score: {ep.score:.2f}, importance: {ep.importance})"
            lines.append(line)

        return f"Found {len(results)} relevant experiences:\n\n" + "\n\n".join(lines)

    except Exception as e:
        logger.warning("search_episodic_memory failed: %s", e)
        return f"Error searching episodic memory: {e}"

record_lesson async

record_lesson(situation: str, lesson: str, category: str = 'decision', importance: int = 5) -> str

Explicitly record a lesson learned for future reference.

Use this when you discover something important that should be remembered for similar future situations.

PARAMETER	DESCRIPTION
situation	What was happening when the lesson was learned. TYPE: str
lesson	The concise lesson or insight to remember. TYPE: str
category	Type of lesson (decision, user_preference, workflow_pattern). TYPE: str DEFAULT: 'decision'
importance	How important this lesson is (1-10, default 5). TYPE: int DEFAULT: 5

RETURNS	DESCRIPTION
str	Confirmation that the lesson was recorded.

Source code in packages/ai-parrot/src/parrot/memory/episodic/tools.py

async def record_lesson(
    self,
    situation: str,
    lesson: str,
    category: str = "decision",
    importance: int = 5,
) -> str:
    """Explicitly record a lesson learned for future reference.

    Use this when you discover something important that should be
    remembered for similar future situations.

    Args:
        situation: What was happening when the lesson was learned.
        lesson: The concise lesson or insight to remember.
        category: Type of lesson (decision, user_preference, workflow_pattern).
        importance: How important this lesson is (1-10, default 5).

    Returns:
        Confirmation that the lesson was recorded.
    """
    try:
        # Map category string to enum
        category_map = {
            "decision": EpisodeCategory.DECISION,
            "user_preference": EpisodeCategory.USER_PREFERENCE,
            "workflow_pattern": EpisodeCategory.WORKFLOW_PATTERN,
            "tool_execution": EpisodeCategory.TOOL_EXECUTION,
            "error_recovery": EpisodeCategory.ERROR_RECOVERY,
            "query_resolution": EpisodeCategory.QUERY_RESOLUTION,
            "handoff": EpisodeCategory.HANDOFF,
        }
        cat = category_map.get(category.lower(), EpisodeCategory.DECISION)

        episode = await self._store.record_episode(
            namespace=self._namespace,
            situation=situation,
            action_taken=f"Recorded lesson: {lesson[:200]}",
            outcome=EpisodeOutcome.SUCCESS,
            category=cat,
            importance=max(1, min(10, importance)),
            generate_reflection=False,
        )
        # Set the lesson directly since we skipped reflection
        episode.lesson_learned = lesson
        episode.reflection = f"Agent explicitly recorded: {lesson}"

        return (
            f"Lesson recorded (id: {episode.episode_id[:8]}). "
            f"Category: {cat.value}, importance: {importance}."
        )

    except Exception as e:
        logger.warning("record_lesson failed: %s", e)
        return f"Error recording lesson: {e}"

get_warnings async

get_warnings(context: str = '') -> str

Get warnings about past mistakes relevant to the current task.

Use this before attempting actions that might fail, to check if similar attempts have failed before.

PARAMETER	DESCRIPTION
context	Description of what you're about to do (for relevance matching). TYPE: str DEFAULT: ''

RETURNS	DESCRIPTION
str	Formatted warnings about past failures and successful approaches,
str	or a message if no relevant warnings exist.

Source code in packages/ai-parrot/src/parrot/memory/episodic/tools.py

async def get_warnings(
    self,
    context: str = "",
) -> str:
    """Get warnings about past mistakes relevant to the current task.

    Use this before attempting actions that might fail, to check
    if similar attempts have failed before.

    Args:
        context: Description of what you're about to do (for relevance matching).

    Returns:
        Formatted warnings about past failures and successful approaches,
        or a message if no relevant warnings exist.
    """
    try:
        warnings = await self._store.get_failure_warnings(
            namespace=self._namespace,
            current_query=context if context else None,
        )

        if not warnings:
            return "No relevant warnings from past experiences."

        return warnings

    except Exception as e:
        logger.warning("get_warnings failed: %s", e)
        return f"Error retrieving warnings: {e}"

ContextAssembler

ContextAssembler(config: MemoryConfig | None = None)

Assembles context from multiple sources within a token budget.

Priority order (highest first): 1. Episodic failure warnings — critical for avoiding past mistakes 2. Relevant skills — applicable knowledge 3. Conversation history — recent turns (truncated from oldest)

Each section gets a weight-based allocation from the total budget. Unused budget from empty sections rolls forward to the next priority.

PARAMETER	DESCRIPTION
config	Optional MemoryConfig; defaults to MemoryConfig() if omitted. TYPE: MemoryConfig | None DEFAULT: None

Example

assembler = ContextAssembler(MemoryConfig(max_context_tokens=2000)) ctx = assembler.assemble( episodic_warnings="Don't call X without auth", relevant_skills="Use get_schema tool", conversation="User: hello\nAssistant: hi", ) print(ctx.tokens_used)

Source code in packages/ai-parrot/src/parrot/memory/unified/context.py

def __init__(self, config: MemoryConfig | None = None) -> None:
    self.config = config or MemoryConfig()
    self._max_tokens = int(self.config.max_context_tokens * _HEADROOM)

assemble

assemble(episodic_warnings: str = '', relevant_skills: str = '', conversation: str = '') -> MemoryContext

Assemble context respecting token budget.

Sections are filled in priority order. Any budget left over from an empty (or smaller-than-allocation) section is carried forward and added to the remaining sections' budgets proportionally.

PARAMETER	DESCRIPTION
episodic_warnings	Past failure lessons from episodic memory. TYPE: str DEFAULT: ''
relevant_skills	Applicable skills from the skill registry. TYPE: str DEFAULT: ''
conversation	Recent conversation turns. TYPE: str DEFAULT: ''

RETURNS	DESCRIPTION
MemoryContext	MemoryContext with filled sections and accurate token accounting.

Source code in packages/ai-parrot/src/parrot/memory/unified/context.py

def assemble(
    self,
    episodic_warnings: str = "",
    relevant_skills: str = "",
    conversation: str = "",
) -> MemoryContext:
    """Assemble context respecting token budget.

    Sections are filled in priority order.  Any budget left over from an
    empty (or smaller-than-allocation) section is carried forward and
    added to the remaining sections' budgets proportionally.

    Args:
        episodic_warnings: Past failure lessons from episodic memory.
        relevant_skills: Applicable skills from the skill registry.
        conversation: Recent conversation turns.

    Returns:
        MemoryContext with filled sections and accurate token accounting.
    """
    if not episodic_warnings and not relevant_skills and not conversation:
        logger.debug("ContextAssembler: all inputs empty, returning empty context")
        return MemoryContext(tokens_budget=self.config.max_context_tokens)

    budget = self._max_tokens
    cfg = self.config

    # --- Priority 1: episodic warnings ---
    episodic_budget = int(budget * cfg.episodic_weight)
    filled_episodic, episodic_tokens = self._fill_section(
        episodic_warnings, episodic_budget
    )
    rollover = episodic_budget - episodic_tokens

    # --- Priority 2: relevant skills (gets episodic rollover) ---
    skills_budget = int(budget * cfg.skill_weight) + rollover
    filled_skills, skills_tokens = self._fill_section(
        relevant_skills, skills_budget
    )
    rollover = skills_budget - skills_tokens

    # --- Priority 3: conversation (gets remaining rollover) ---
    conv_budget = budget - episodic_budget - int(budget * cfg.skill_weight) + rollover
    # Simpler: whatever is left from the total after the first two sections
    conv_budget = budget - episodic_tokens - skills_tokens
    filled_conv, conv_tokens = self._fill_conversation(conversation, conv_budget)

    total_tokens = episodic_tokens + skills_tokens + conv_tokens

    logger.debug(
        "ContextAssembler: assembled %d tokens "
        "(episodic=%d, skills=%d, conv=%d, budget=%d)",
        total_tokens,
        episodic_tokens,
        skills_tokens,
        conv_tokens,
        self._max_tokens,
    )

    return MemoryContext(
        episodic_warnings=filled_episodic,
        relevant_skills=filled_skills,
        conversation_summary=filled_conv,
        tokens_used=total_tokens,
        tokens_budget=self.config.max_context_tokens,
    )

LongTermMemoryMixin

Single opt-in mixin for long-term memory in any bot/agent.

Provides unified episodic + skill + conversation memory without requiring the bot to manage individual subsystems.

MRO note: place before AbstractBot (or Agent) in the class definition so this mixin's methods take priority in the resolution order:

class MyAgent(LongTermMemoryMixin, Agent):
    enable_long_term_memory = True

Configuration attributes (override in the subclass or via kwargs): enable_long_term_memory: Master toggle — all methods are no-ops when False. episodic_inject_warnings: Retrieve past failure warnings. episodic_auto_record: Record interactions to episodic memory. episodic_max_warnings: Maximum failure warnings per context. skill_inject_context: Retrieve relevant skills into context. skill_auto_extract: Auto-extract skills from successful interactions. skill_expose_tools: Register skill tools with the agent's tool manager. skill_max_context: Maximum skills per context. memory_max_context_tokens: Total token budget for assembled context.

get_memory_context async

get_memory_context(query: str, user_id: str, session_id: str) -> str

Return assembled memory context as an injectable prompt string.

PARAMETER	DESCRIPTION
query	Current user query for semantic retrieval. TYPE: str
user_id	User identifier for conversation history. TYPE: str
session_id	Session identifier for conversation history. TYPE: str

RETURNS	DESCRIPTION
str	Formatted multi-section string ready for system prompt injection,
str	or empty string when memory is disabled or not configured.

Source code in packages/ai-parrot/src/parrot/memory/unified/mixin.py

async def get_memory_context(
    self,
    query: str,
    user_id: str,
    session_id: str,
) -> str:
    """Return assembled memory context as an injectable prompt string.

    Args:
        query: Current user query for semantic retrieval.
        user_id: User identifier for conversation history.
        session_id: Session identifier for conversation history.

    Returns:
        Formatted multi-section string ready for system prompt injection,
        or empty string when memory is disabled or not configured.
    """
    if not self.enable_long_term_memory or self._memory_manager is None:
        return ""

    try:
        ctx = await self._memory_manager.get_context_for_query(
            query=query,
            user_id=user_id,
            session_id=session_id,
        )
        return ctx.to_prompt_string()
    except Exception as exc:  # noqa: BLE001
        logger.warning("get_memory_context failed: %s", exc)
        return ""

MemoryConfig

Bases: BaseModel

Configuration for UnifiedMemoryManager.

Controls which subsystems are enabled, token budget allocation, and per-subsystem limits.

MemoryContext

Bases: BaseModel

Assembled context from all memory subsystems.

Holds the text sections retrieved from episodic memory, skill registry, and conversation history, along with token accounting for budget enforcement.

to_prompt_string

to_prompt_string() -> str

Format as injectable system prompt sections.

Only non-empty sections are included. Each section is wrapped in descriptive XML tags so the LLM can distinguish memory types.

RETURNS	DESCRIPTION
str	Formatted string ready for injection into a system prompt.

Source code in packages/ai-parrot/src/parrot/memory/unified/models.py

def to_prompt_string(self) -> str:
    """Format as injectable system prompt sections.

    Only non-empty sections are included.  Each section is wrapped
    in descriptive XML tags so the LLM can distinguish memory types.

    Returns:
        Formatted string ready for injection into a system prompt.
    """
    sections: list[str] = []

    if self.episodic_warnings:
        sections.append(
            "<past_failures_to_avoid>\n"
            f"{self.episodic_warnings}\n"
            "</past_failures_to_avoid>"
        )

    if self.relevant_skills:
        sections.append(
            "<relevant_skills>\n"
            f"{self.relevant_skills}\n"
            "</relevant_skills>"
        )

    if self.conversation_summary:
        sections.append(
            "<recent_conversation>\n"
            f"{self.conversation_summary}\n"
            "</recent_conversation>"
        )

    return "\n\n".join(sections)

UnifiedMemoryManager

UnifiedMemoryManager(namespace: MemoryNamespace, conversation_memory: Optional[ConversationMemory] = None, episodic_store: Optional[EpisodicMemoryStore] = None, skill_registry: Optional[Any] = None, config: Optional[MemoryConfig] = None, cross_domain_router: Optional[CrossDomainRouter] = None)

Coordinates episodic memory, skill registry, and conversation memory.

All retrieval in get_context_for_query runs concurrently via asyncio.gather. Subsystems that are None are silently skipped.

When cross_domain_router is provided, get_context_for_query also queries episodic memories from relevant agent namespaces identified by the router. Cross-domain results are labeled and appended to the episodic warnings text. Cross-domain failures never break the main retrieval flow.

PARAMETER	DESCRIPTION
namespace	Scoping dimensions for episodic memory queries. TYPE: MemoryNamespace
conversation_memory	Optional conversation history store. TYPE: Optional[ConversationMemory] DEFAULT: None
episodic_store	Optional episodic memory store. TYPE: Optional[EpisodicMemoryStore] DEFAULT: None
skill_registry	Optional skill registry (duck-typed via SkillRegistry protocol). TYPE: Optional[Any] DEFAULT: None
config	Optional memory configuration; defaults to MemoryConfig(). TYPE: Optional[MemoryConfig] DEFAULT: None
cross_domain_router	Optional router for multi-agent memory sharing. TYPE: Optional[CrossDomainRouter] DEFAULT: None

Example

manager = UnifiedMemoryManager( namespace=MemoryNamespace(agent_id="my-agent"), episodic_store=store, cross_domain_router=router, ) ctx = await manager.get_context_for_query("user query", "u1", "s1") prompt += ctx.to_prompt_string()

Source code in packages/ai-parrot/src/parrot/memory/unified/manager.py

def __init__(
    self,
    namespace: MemoryNamespace,
    conversation_memory: Optional[ConversationMemory] = None,
    episodic_store: Optional[EpisodicMemoryStore] = None,
    skill_registry: Optional[Any] = None,
    config: Optional[MemoryConfig] = None,
    cross_domain_router: Optional[CrossDomainRouter] = None,
) -> None:
    self.namespace = namespace
    self.conversation = conversation_memory
    self.episodic = episodic_store
    self.skills = skill_registry
    self.config = config or MemoryConfig()
    self._assembler = ContextAssembler(self.config)
    self._cross_domain_router = cross_domain_router
    self.logger = logging.getLogger(f"{__name__}.{self.__class__.__name__}")

configure async

configure(**kwargs: Any) -> None

Initialise all non-None subsystems.

Calls configure(**kwargs) on each subsystem that exposes it. Subsystems that lack a configure method are silently skipped.

PARAMETER	DESCRIPTION
**kwargs	Forwarded verbatim to each subsystem's configure. TYPE: Any DEFAULT: {}

Source code in packages/ai-parrot/src/parrot/memory/unified/manager.py

async def configure(self, **kwargs: Any) -> None:
    """Initialise all non-None subsystems.

    Calls ``configure(**kwargs)`` on each subsystem that exposes it.
    Subsystems that lack a ``configure`` method are silently skipped.

    Args:
        **kwargs: Forwarded verbatim to each subsystem's ``configure``.
    """
    for name, subsystem in self._subsystems():
        method = getattr(subsystem, "configure", None)
        if callable(method):
            self.logger.debug("Configuring subsystem: %s", name)
            await method(**kwargs)

cleanup async

cleanup() -> None

Release resources held by all non-None subsystems.

Calls cleanup() on each subsystem that exposes it.

Source code in packages/ai-parrot/src/parrot/memory/unified/manager.py

async def cleanup(self) -> None:
    """Release resources held by all non-None subsystems.

    Calls ``cleanup()`` on each subsystem that exposes it.
    """
    for name, subsystem in self._subsystems():
        method = getattr(subsystem, "cleanup", None)
        if callable(method):
            self.logger.debug("Cleaning up subsystem: %s", name)
            await method()

get_context_for_query async

get_context_for_query(query: str, user_id: str, session_id: str) -> MemoryContext

Retrieve and assemble context from all memory subsystems.

All three retrieval calls run concurrently via asyncio.gather. If a subsystem is None or raises an exception its section is returned as an empty string.

PARAMETER	DESCRIPTION
query	Current user query — used for semantic similarity search. TYPE: str
user_id	User identifier for conversation history lookup. TYPE: str
session_id	Session identifier for conversation history lookup. TYPE: str

RETURNS	DESCRIPTION
MemoryContext	MemoryContext assembled within the configured token budget.

Source code in packages/ai-parrot/src/parrot/memory/unified/manager.py

async def get_context_for_query(
    self,
    query: str,
    user_id: str,
    session_id: str,
) -> MemoryContext:
    """Retrieve and assemble context from all memory subsystems.

    All three retrieval calls run concurrently via ``asyncio.gather``.
    If a subsystem is ``None`` or raises an exception its section is
    returned as an empty string.

    Args:
        query: Current user query — used for semantic similarity search.
        user_id: User identifier for conversation history lookup.
        session_id: Session identifier for conversation history lookup.

    Returns:
        ``MemoryContext`` assembled within the configured token budget.
    """
    episodic_task = self._get_episodic_warnings(query)
    skills_task = self._get_relevant_skills(query)
    conversation_task = self._get_conversation(user_id, session_id)

    episodic_text, skills_text, conv_text = await asyncio.gather(
        episodic_task,
        skills_task,
        conversation_task,
    )

    return self._assembler.assemble(
        episodic_warnings=episodic_text,
        relevant_skills=skills_text,
        conversation=conv_text,
    )

record_interaction async

record_interaction(query: str, response: Any, tool_calls: list[Any], user_id: str, session_id: str) -> None

Record a completed interaction to episodic and conversation memory.

This method is exception-safe: any error is logged at WARNING level and execution continues. It is designed to be called fire-and-forget after returning a response to the user.

PARAMETER	DESCRIPTION
query	The user's original query. TYPE: str
response	The agent's response object (content extracted as str). TYPE: Any
tool_calls	List of tool call objects from the interaction. TYPE: list[Any]
user_id	User identifier. TYPE: str
session_id	Session identifier. TYPE: str

Source code in packages/ai-parrot/src/parrot/memory/unified/manager.py

async def record_interaction(
    self,
    query: str,
    response: Any,
    tool_calls: list[Any],
    user_id: str,
    session_id: str,
) -> None:
    """Record a completed interaction to episodic and conversation memory.

    This method is exception-safe: any error is logged at WARNING level
    and execution continues.  It is designed to be called fire-and-forget
    after returning a response to the user.

    Args:
        query: The user's original query.
        response: The agent's response object (content extracted as str).
        tool_calls: List of tool call objects from the interaction.
        user_id: User identifier.
        session_id: Session identifier.
    """
    try:
        if self.episodic is not None:
            await self._record_episodic(
                query, response, tool_calls, user_id, session_id
            )
    except Exception as exc:  # noqa: BLE001
        self.logger.warning(
            "record_interaction: episodic recording failed — %s", exc
        )