Skip to content

Memory

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 
141
142
143
144
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 
146
147
148
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

clear_turns 

clear_turns() -> None

Clear all turns from the conversation history.

Source code in packages/ai-parrot/src/parrot/memory/abstract.py 
155
156
157
158
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 
160
161
162
163
164
165
166
167
168
169
170
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 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
@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

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 
194
195
196
197
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 
199
200
201
202
203
204
@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 
206
207
208
209
210
211
@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 
213
214
215
216
@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 
218
219
220
221
222
223
@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 
225
226
227
228
@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 
230
231
232
233
@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 
235
236
237
238
@abstractmethod
async def delete_history(self, user_id: str, session_id: str, chatbot_id: Optional[str] = None) -> bool:
    """Delete a conversation history entirely."""
    pass

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(), chatbot_id: Optional[str] = None)

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 
32
33
34
35
36
37
38
39
40
41
42
43
44
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,
        "chatbot_id": self.chatbot_id,
    }

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 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
@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", {}),
        # Legacy records predate attribution — absent key means "unknown".
        chatbot_id=data.get("chatbot_id"),
    )

from_ai_message classmethod 

from_ai_message(*, user_message: str, response: AIMessage, user_id: str, chatbot_id: str, context_used: Optional[str] = None, turn_id: Optional[str] = None, assistant_text: Optional[str] = None) -> ConversationTurn

Build a turn from the AIMessage a bot round produced.

This is the canonical constructor used by AbstractBot.save_conversation_turn call sites (FEAT-524). Before this existed every bot entry point hand-rolled its own turn with slightly different metadata keys; routing through here gives every persisted turn one shape.

PARAMETER DESCRIPTION
user_message

The user's text for this round.

TYPE: str

response

The final :class:~parrot.models.responses.AIMessage returned by the bot — i.e. after guardrails, redaction and formatting, not the raw client output.

TYPE: AIMessage

user_id

Owner of the conversation.

TYPE: str

chatbot_id

The agent producing the turn. Must be the bot's memory_key_id — attribution and storage key must agree.

TYPE: str

context_used

Optional retrieval context string used this round.

TYPE: Optional[str] DEFAULT: None

turn_id

Explicit turn id. Defaults to response.turn_id, then to a fresh uuid4.

TYPE: Optional[str] DEFAULT: None

assistant_text

Overrides the assistant text taken from response. Used by the streaming partial-save path, where the accumulated text is authoritative and the AIMessage is synthesized after the fact.

TYPE: Optional[str] DEFAULT: None

RETURNS DESCRIPTION
ConversationTurn

A fully populated :class:ConversationTurn.
Source code in packages/ai-parrot/src/parrot/memory/abstract.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
@classmethod
def from_ai_message(
    cls,
    *,
    user_message: str,
    response: "AIMessage",
    user_id: str,
    chatbot_id: str,
    context_used: Optional[str] = None,
    turn_id: Optional[str] = None,
    assistant_text: Optional[str] = None,
) -> "ConversationTurn":
    """Build a turn from the ``AIMessage`` a bot round produced.

    This is the canonical constructor used by
    ``AbstractBot.save_conversation_turn`` call sites (FEAT-524). Before
    this existed every bot entry point hand-rolled its own turn with
    slightly different metadata keys; routing through here gives every
    persisted turn one shape.

    Args:
        user_message: The user's text for this round.
        response: The final :class:`~parrot.models.responses.AIMessage`
            returned by the bot — i.e. *after* guardrails, redaction and
            formatting, not the raw client output.
        user_id: Owner of the conversation.
        chatbot_id: The agent producing the turn. Must be the bot's
            ``memory_key_id`` — attribution and storage key must agree.
        context_used: Optional retrieval context string used this round.
        turn_id: Explicit turn id. Defaults to ``response.turn_id``, then
            to a fresh uuid4.
        assistant_text: Overrides the assistant text taken from
            ``response``. Used by the streaming partial-save path, where
            the accumulated text is authoritative and the ``AIMessage`` is
            synthesized after the fact.

    Returns:
        A fully populated :class:`ConversationTurn`.
    """
    if assistant_text is not None:
        answer = assistant_text
    else:
        answer = getattr(response, "to_text", None)
        if answer is None:
            answer = str(getattr(response, "content", "") or "")

    tool_calls = getattr(response, "tool_calls", None) or []
    usage = getattr(response, "usage", None)

    return cls(
        turn_id=turn_id or getattr(response, "turn_id", None) or str(uuid.uuid4()),
        user_id=user_id,
        user_message=user_message,
        assistant_response=answer,
        context_used=context_used,
        tools_used=[tc.name for tc in tool_calls],
        metadata={
            "model": getattr(response, "model", None),
            "provider": getattr(response, "provider", None),
            "usage": usage.model_dump() if hasattr(usage, "model_dump") else usage,
            "finish_reason": getattr(response, "finish_reason", None),
            "response_time": getattr(response, "response_time", None),
        },
        chatbot_id=chatbot_id,
    )

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 
10
11
12
13
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 
15
16
17
18
19
20
21
22
23
24
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 
26
27
28
29
30
31
32
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. episodic_flush_on_start: Delete persisted episodes (FAISS files + Redis episode cache) when the agent configures, starting each process with a clean episodic slate. Useful while iterating on agent behavior so stale lessons don't reinforce old patterns.

flush_episodic_memory async 

flush_episodic_memory() -> None

Delete persisted episodic memory for this agent.

Removes the FAISS persistence files (episodes.faiss, episodes.jsonl, id_order.json) and every episodic:* Redis cache key scoped to this agent. Safe to call when nothing is persisted. Only covers the FAISS backend — PgVector-backed agents must truncate the episodes table instead.

Source code in packages/ai-parrot/src/parrot/memory/episodic/mixin.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
async def flush_episodic_memory(self) -> None:
    """Delete persisted episodic memory for this agent.

    Removes the FAISS persistence files (``episodes.faiss``,
    ``episodes.jsonl``, ``id_order.json``) and every ``episodic:*`` Redis
    cache key scoped to this agent. Safe to call when nothing is
    persisted. Only covers the FAISS backend — PgVector-backed agents
    must truncate the episodes table instead.
    """
    # 1. FAISS disk persistence
    path = self._resolve_episodic_path()
    if path:
        base = Path(path)
        for fname in ("episodes.faiss", "episodes.jsonl", "id_order.json"):
            target = base / fname
            if target.exists():
                target.unlink()
                logger.info("Episodic flush: removed %s", target)

    # 2. Redis episode cache (episodic:{tenant}:{agent_id}...)
    redis_client = getattr(self, "redis", None)
    if redis_client is not None:
        agent_id = self._get_agent_id()
        pattern = f"episodic:*{agent_id}*"
        try:
            deleted = 0
            async for key in redis_client.scan_iter(match=pattern):
                await redis_client.delete(key)
                deleted += 1
            if deleted:
                logger.info(
                    "Episodic flush: removed %d Redis cache keys (%s)",
                    deleted, pattern,
                )
        except Exception as e:  # non-fatal — cache keys expire via TTL anyway
            logger.warning("Episodic flush: Redis cleanup failed: %s", e)

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 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
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 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
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 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
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 
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
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 
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
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 
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
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 
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
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 
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
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]

mark_consolidated async 

mark_consolidated(episode_ids: list[str], page_id: str) -> int

Mark episodes as consolidated into a dream-cycle brain wiki page.

Thin passthrough over backend.update_metadata() (FEAT-390). Never deletes episodes — existing TTL/compaction keeps pruning. Backends that do not implement update_metadata are tolerated (watermark-only mode): this returns 0 with a WARNING logged.

PARAMETER DESCRIPTION
episode_ids

Episode ids to mark as consolidated.

TYPE: list[str]

page_id

Brain wiki page id the episodes were distilled into.

TYPE: str

RETURNS DESCRIPTION
int

Number of episodes actually updated.
Source code in packages/ai-parrot/src/parrot/memory/episodic/store.py 
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
async def mark_consolidated(
    self,
    episode_ids: list[str],
    page_id: str,
) -> int:
    """Mark episodes as consolidated into a dream-cycle brain wiki page.

    Thin passthrough over ``backend.update_metadata()`` (FEAT-390).
    Never deletes episodes — existing TTL/compaction keeps pruning.
    Backends that do not implement ``update_metadata`` are tolerated
    (watermark-only mode): this returns 0 with a WARNING logged.

    Args:
        episode_ids: Episode ids to mark as consolidated.
        page_id: Brain wiki page id the episodes were distilled into.

    Returns:
        Number of episodes actually updated.
    """
    if not hasattr(self._backend, "update_metadata"):
        logger.warning(
            "Backend %s has no update_metadata(); mark_consolidated() "
            "is a no-op (watermark-only mode)",
            type(self._backend).__name__,
        )
        return 0

    return await self._backend.update_metadata(
        episode_ids, {"consolidated_into": page_id}
    )

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 
608
609
610
611
612
613
614
615
616
617
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 
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
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 
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
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 
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
@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 
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
@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 
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
@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 
37
38
39
40
41
42
43
44
45
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 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
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 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
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 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
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}"

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 
12
13
14
15
16
def __init__(self, base_path: str = "./conversations"):
    super().__init__()
    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 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
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, re-keying a legacy record if needed.

FEAT-524 unified the storage path to {user}/{chatbot}/{session}.json. Histories written before that live at {user}/{session}.json. When chatbot_id is given and the segmented path holds nothing, the legacy path is read once, copied to the segmented path and returned. The legacy file is deliberately left in place so a rollback still finds it.

PARAMETER DESCRIPTION
user_id

Owner of the conversation.

TYPE: str

session_id

Conversation session.

TYPE: str

chatbot_id

Agent path segment. Falsy means "read the legacy path directly" — no re-key is attempted.

TYPE: Optional[str] DEFAULT: None

RETURNS DESCRIPTION
Optional[ConversationHistory]

The history, or None when neither path holds one.
Source code in packages/ai-parrot/src/parrot/memory/file.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
async def get_history(
    self, user_id: str, session_id: str, chatbot_id: Optional[str] = None
) -> Optional[ConversationHistory]:
    """Get a conversation history, re-keying a legacy record if needed.

    FEAT-524 unified the storage path to ``{user}/{chatbot}/{session}.json``.
    Histories written before that live at ``{user}/{session}.json``. When
    ``chatbot_id`` is given and the segmented path holds nothing, the legacy
    path is read once, copied to the segmented path and returned. The legacy
    file is deliberately **left in place** so a rollback still finds it.

    Args:
        user_id: Owner of the conversation.
        session_id: Conversation session.
        chatbot_id: Agent path segment. Falsy means "read the legacy path
            directly" — no re-key is attempted.

    Returns:
        The history, or ``None`` when neither path holds one.
    """
    # One lock acquisition for read-and-maybe-copy: ``self._lock`` is a
    # plain asyncio.Lock (not reentrant), so calling update_history() from
    # in here would deadlock. The write is inlined via _write_history().
    async with self._lock:
        history = await self._read_history(user_id, session_id, chatbot_id)
        if history is not None or not chatbot_id:
            return history

        legacy = await self._read_history(user_id, session_id, None)
        if legacy is None:
            return None

        legacy.chatbot_id = str(chatbot_id)
        await self._write_history(legacy)
        self.logger.info("Re-keyed legacy conversation %s/%s under chatbot %s", user_id, session_id, chatbot_id)
        return legacy

update_history async 

update_history(history: ConversationHistory) -> None

Update a conversation history.

Source code in packages/ai-parrot/src/parrot/memory/file.py 
114
115
116
117
async def update_history(self, history: ConversationHistory) -> None:
    """Update a conversation history."""
    async with self._lock:
        await self._write_history(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/file.py 
119
120
121
122
123
124
125
126
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 
128
129
130
131
132
133
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 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
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 
165
166
167
168
169
170
171
172
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

InMemoryConversation 

InMemoryConversation()

Bases: ConversationMemory

In-memory implementation of conversation memory.

Source code in packages/ai-parrot/src/parrot/memory/mem.py 
 8
 9
10
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 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
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 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
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 
58
59
60
61
62
63
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 
65
66
67
68
69
70
71
72
73
74
75
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 
77
78
79
80
81
82
83
84
85
86
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 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
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 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
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 
13
14
15
16
17
18
19
20
21
22
23
24
25
def __init__(self, redis_url: str = None, key_prefix: str = "conversation", use_hash_storage: bool = True):
    super().__init__()
    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 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
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, re-keying a legacy record if needed.

FEAT-524 unified the storage key to (chatbot, user, session). Histories written before that live under the un-segmented conversation:{user}:{session} key, and these keys carry no TTL, so without a fallback every live conversation would silently restart the first time a bot asked for its history under the new key.

When chatbot_id is given and the segmented key holds nothing, the legacy key is read once, copied under the segmented key and returned. The legacy record is deliberately left in place so a rollback still finds it. The copy is idempotent, so two concurrent first reads racing each other is benign — no locking (spec §7 "Re-key race").

PARAMETER DESCRIPTION
user_id

Owner of the conversation.

TYPE: str

session_id

Conversation session.

TYPE: str

chatbot_id

Agent key segment. Falsy means "read the legacy key directly" — no re-key is attempted.

TYPE: Optional[str] DEFAULT: None

RETURNS DESCRIPTION
Optional[ConversationHistory]

The history, or None when neither key holds one.
Source code in packages/ai-parrot/src/parrot/memory/redis.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
async def get_history(
    self, user_id: str, session_id: str, chatbot_id: Optional[str] = None
) -> Optional[ConversationHistory]:
    """Get a conversation history, re-keying a legacy record if needed.

    FEAT-524 unified the storage key to ``(chatbot, user, session)``.
    Histories written before that live under the un-segmented
    ``conversation:{user}:{session}`` key, and these keys carry no TTL, so
    without a fallback every live conversation would silently restart the
    first time a bot asked for its history under the new key.

    When ``chatbot_id`` is given and the segmented key holds nothing, the
    legacy key is read once, copied under the segmented key and returned.
    The legacy record is deliberately **left in place** so a rollback still
    finds it. The copy is idempotent, so two concurrent first reads racing
    each other is benign — no locking (spec §7 "Re-key race").

    Args:
        user_id: Owner of the conversation.
        session_id: Conversation session.
        chatbot_id: Agent key segment. Falsy means "read the legacy key
            directly" — no re-key is attempted.

    Returns:
        The history, or ``None`` when neither key holds one.
    """
    history = await self._load_history(user_id, session_id, chatbot_id)
    if history is not None or not chatbot_id:
        return history

    legacy = await self._load_history(user_id, session_id, None)
    if legacy is None:
        return None

    legacy.chatbot_id = str(chatbot_id)
    await self.update_history(legacy)
    # Keep list_sessions(user_id, chatbot_id) able to find the migrated
    # session — update_history alone only writes the history key.
    await self.redis.sadd(self._get_user_sessions_key(user_id, chatbot_id), session_id)
    self.logger.info("Re-keyed legacy conversation %s/%s under chatbot %s", user_id, session_id, chatbot_id)
    return legacy

update_history async 

update_history(history: ConversationHistory) -> None

Update a conversation history.

Source code in packages/ai-parrot/src/parrot/memory/redis.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
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 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
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 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
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 
269
270
271
272
273
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 
275
276
277
278
279
280
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 
282
283
284
285
286
287
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 
289
290
291
292
293
294
295
296
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 
299
300
301
302
303
304
305
306
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 
308
309
310
311
312
313
314
315
316
317
318
319
320
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 
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
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 
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
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 
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
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 
462
463
464
465
466
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)

HistoryMessage dataclass 

HistoryMessage(role: Literal['user', 'assistant'], content: str, chatbot_id: Optional[str] = None, turn_id: Optional[str] = None)

A single rendered conversation message, provider-neutral and text-only.

ATTRIBUTE DESCRIPTION
role

Either "user" or "assistant".

TYPE: Literal['user', 'assistant']

content

The message text.

TYPE: str

chatbot_id

The agent that produced the turn this message came from, or None for legacy turns written before attribution existed. When a message is the result of merging several turns, this is the chatbot_id of the first of them.

TYPE: Optional[str]

turn_id

The originating :class:~parrot.memory.ConversationTurn id, under the same "first of a merged run" rule as chatbot_id.

TYPE: Optional[str]

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 
41
42
43
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 = '', semantic_knowledge: 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: ''

semantic_knowledge

Distilled long-term knowledge from the agent's brain wiki (FEAT-390); budgeted at config.brain_weight. Empty by default, so 3-argument call sites behave exactly as before.

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 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def assemble(
    self,
    episodic_warnings: str = "",
    relevant_skills: str = "",
    conversation: str = "",
    semantic_knowledge: 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.
        semantic_knowledge: Distilled long-term knowledge from the
            agent's brain wiki (FEAT-390); budgeted at
            ``config.brain_weight``. Empty by default, so 3-argument
            call sites behave exactly as before.

    Returns:
        MemoryContext with filled sections and accurate token accounting.
    """
    if (
        not episodic_warnings
        and not relevant_skills
        and not conversation
        and not semantic_knowledge
    ):
        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: brain knowledge (gets skills rollover) ---
    brain_budget = int(budget * cfg.brain_weight) + rollover
    filled_brain, brain_tokens = self._fill_section(
        semantic_knowledge, brain_budget
    )

    # --- Priority 4: conversation (gets whatever's left) ---
    conv_budget = budget - episodic_tokens - skills_tokens - brain_tokens
    filled_conv, conv_tokens = self._fill_conversation(conversation, conv_budget)

    total_tokens = episodic_tokens + skills_tokens + brain_tokens + conv_tokens

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

    return MemoryContext(
        episodic_warnings=filled_episodic,
        relevant_skills=filled_skills,
        conversation_summary=filled_conv,
        semantic_knowledge=filled_brain,
        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. enable_brain: Master toggle for the dream cycle (FEAT-390) — episodic-to-wiki brain consolidation. All brain/dream objects are no-ops (never constructed) when False. dream_interval_hours: Hours between dream cycles. dream_importance_threshold: Minimum episode importance eligible for consolidation (episodes with a lesson are always eligible). brain_storage_dir: Directory for the agent's brain wiki; defaults to ~/.parrot/brains/<agent_id>. brain_promote_to_org: Also maintain an org-level brain wiki and promote reinforced pages into it. org_promotion_cycles: Distinct-cycle reinforcement threshold before a page is promoted to the org wiki.

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 
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
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 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
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>"
        )

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

    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, brain: Optional[BrainStore] = None, org_brain: Optional[BrainStore] = 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 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
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,
    brain: Optional[BrainStore] = None,
    org_brain: Optional[BrainStore] = 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.brain = brain
    self.org_brain = org_brain
    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 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
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 
123
124
125
126
127
128
129
130
131
132
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 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
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)
    brain_task = self._get_brain_knowledge(query)

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

    return self._assembler.assemble(
        episodic_warnings=episodic_text,
        relevant_skills=skills_text,
        conversation=conv_text,
        semantic_knowledge=brain_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 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
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
        )

render_history 

render_history(history: Optional[ConversationHistory], *, max_turns: Optional[int] = None, current_chatbot_id: Optional[str] = None, include_other_agents: bool = True, other_agent_label: str = '[agent:{chatbot_id}]') -> List[HistoryMessage]

Render a conversation history into alternating provider-neutral messages.

PARAMETER DESCRIPTION
history

The history to render. None or empty renders to [].

TYPE: Optional[ConversationHistory]

max_turns

Keep only the most recent N turns. None keeps all. Values <= 0 also render to [].

TYPE: Optional[int] DEFAULT: None

current_chatbot_id

The agent doing the asking. Turns whose chatbot_id differs are considered foreign; a turn with chatbot_id is None (legacy, pre-attribution) is always treated as belonging to the current agent. When current_chatbot_id is None no turn is foreign.

TYPE: Optional[str] DEFAULT: None

include_other_agents

When False, foreign turns are dropped entirely. When True (the default) they are kept and their assistant text is prefixed with other_agent_label so the model can tell who said what — the crew/flow shared-history case.

TYPE: bool DEFAULT: True

other_agent_label

Format string for the foreign-turn prefix; receives chatbot_id as its only field.

TYPE: str DEFAULT: '[agent:{chatbot_id}]'

RETURNS DESCRIPTION
List[HistoryMessage]

A list of :class:HistoryMessage with these guarantees:
List[HistoryMessage]
  • roles strictly alternate; the list starts with "user" and ends with "assistant" (or is empty);
List[HistoryMessage]
  • consecutive same-role messages are merged with a blank line;
List[HistoryMessage]
  • turns whose assistant_response is empty or whitespace-only contribute nothing at all — never an empty assistant message;
List[HistoryMessage]
  • history itself is not modified.
Source code in packages/ai-parrot/src/parrot/memory/render.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def render_history(
    history: Optional[ConversationHistory],
    *,
    max_turns: Optional[int] = None,
    current_chatbot_id: Optional[str] = None,
    include_other_agents: bool = True,
    other_agent_label: str = "[agent:{chatbot_id}]",
) -> List[HistoryMessage]:
    """Render a conversation history into alternating provider-neutral messages.

    Args:
        history: The history to render. ``None`` or empty renders to ``[]``.
        max_turns: Keep only the most recent ``N`` turns. ``None`` keeps all.
            Values ``<= 0`` also render to ``[]``.
        current_chatbot_id: The agent doing the asking. Turns whose
            ``chatbot_id`` differs are considered *foreign*; a turn with
            ``chatbot_id is None`` (legacy, pre-attribution) is always treated
            as belonging to the current agent. When ``current_chatbot_id`` is
            ``None`` no turn is foreign.
        include_other_agents: When ``False``, foreign turns are dropped
            entirely. When ``True`` (the default) they are kept and their
            assistant text is prefixed with ``other_agent_label`` so the model
            can tell who said what — the crew/flow shared-history case.
        other_agent_label: Format string for the foreign-turn prefix; receives
            ``chatbot_id`` as its only field.

    Returns:
        A list of :class:`HistoryMessage` with these guarantees:

        * roles strictly alternate; the list starts with ``"user"`` and ends
          with ``"assistant"`` (or is empty);
        * consecutive same-role messages are merged with a blank line;
        * turns whose ``assistant_response`` is empty or whitespace-only
          contribute nothing at all — never an empty assistant message;
        * ``history`` itself is not modified.
    """
    if history is None or not history.turns:
        return []

    if max_turns is not None:
        if max_turns <= 0:
            return []
        turns = history.turns[-max_turns:]
    else:
        turns = history.turns

    rendered: List[HistoryMessage] = []
    for turn in turns:
        assistant_response = turn.assistant_response or ""
        if not assistant_response.strip():
            # An assistant message with no text is not a usable turn: it would
            # either be rejected by the provider or teach the model to answer
            # with silence. Skip the whole turn so alternation stays intact.
            continue

        is_foreign = (
            current_chatbot_id is not None and turn.chatbot_id is not None and turn.chatbot_id != current_chatbot_id
        )
        if is_foreign and not include_other_agents:
            continue
        if is_foreign:
            label = other_agent_label.format(chatbot_id=turn.chatbot_id)
            assistant_response = f"{label} {assistant_response}"

        _append(
            rendered,
            HistoryMessage(
                role="user",
                content=turn.user_message or "",
                chatbot_id=turn.chatbot_id,
                turn_id=turn.turn_id,
            ),
        )
        _append(
            rendered,
            HistoryMessage(
                role="assistant",
                content=assistant_response,
                chatbot_id=turn.chatbot_id,
                turn_id=turn.turn_id,
            ),
        )

    return rendered