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

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

get_recent_turns ¶

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

Get the most recent turns for context.

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

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

clear_turns ¶

clear_turns() -> None

Clear all turns from the conversation history.

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

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

to_dict ¶

to_dict() -> Dict[str, Any]

Serialize conversation history to dictionary.

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

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

from_dict `classmethod` ¶

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

Deserialize conversation history from dictionary.

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

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

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

    return history

ConversationMemory ¶

ConversationMemory(debug: bool = False, *, token_counter: Optional[TokenCounter] = None, omission_store: Optional[OmissionStore] = None, normalize: bool = True, oversize_tool_tokens: int = 2000)

Bases: ABC

Abstract base class for conversation memory storage.

FEAT-525 turns add_turn into a concrete template method (the FEAT-391 "concrete public, abstract private" pattern): every writer — bot, ChatStorage cold tier, voice transcripts — gets Stage 0 (normalization), Stage 0.5 (token counting) and write-time oversize offload for free. Backends implement the abstract :meth:_store_turn only, persisting the turn and (when given) the updated metadata["compaction"] in one write.

Initialize the memory.

PARAMETER	DESCRIPTION
`debug`	Enables verbose per-history debug logging. TYPE: `bool` DEFAULT: `False`
`token_counter`	The counter used for Stage 0.5. Defaults to :func:`parrot.memory.compaction.tokens.get_default_counter` on first use (lazy — never resolved unless needed). TYPE: `Optional[TokenCounter]` DEFAULT: `None`
`omission_store`	The store oversized tool outputs are offloaded to. Backends normally pass their own default; falling back to an :class:`InMemoryOmissionStore` here is a safety net, not the intended configuration. TYPE: `Optional[OmissionStore]` DEFAULT: `None`
`normalize`	When `False`, disables Stage 0 for this instance only; Stage 0.5 (token counting) stays always-on. TYPE: `bool` DEFAULT: `True`
`oversize_tool_tokens`	Write-time offload threshold, in tokens (same default as :class:`~parrot.memory.compaction.models.ContextBudget`). TYPE: `int` DEFAULT: `2000`

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

def __init__(
    self,
    debug: bool = False,
    *,
    token_counter: Optional["TokenCounter"] = None,
    omission_store: Optional[OmissionStore] = None,
    normalize: bool = True,
    oversize_tool_tokens: int = 2_000,
) -> None:
    """Initialize the memory.

    Args:
        debug: Enables verbose per-history debug logging.
        token_counter: The counter used for Stage 0.5. Defaults to
            :func:`parrot.memory.compaction.tokens.get_default_counter`
            on first use (lazy — never resolved unless needed).
        omission_store: The store oversized tool outputs are offloaded
            to. Backends normally pass their own default; falling back
            to an :class:`InMemoryOmissionStore` here is a safety net,
            not the intended configuration.
        normalize: When ``False``, disables Stage 0 for this instance
            only; Stage 0.5 (token counting) stays always-on.
        oversize_tool_tokens: Write-time offload threshold, in tokens
            (same default as :class:`~parrot.memory.compaction.models.ContextBudget`).
    """
    self.logger = logging.getLogger(f"parrot.Memory.{self.__class__.__name__}")
    self._json = JSONContent()
    self.debug = debug
    self._token_counter = token_counter
    self._omission_store = omission_store
    self._normalize = normalize
    self._oversize_tool_tokens = oversize_tool_tokens

token_counter `property` ¶

token_counter: TokenCounter

The Stage 0.5 token counter, resolved lazily on first use.

omission_store `property` ¶

omission_store: OmissionStore

The store oversized tool outputs are offloaded to.

Backends should set self._omission_store in their own __init__ to a store sharing their connection/root. This fallback exists so the property never raises, but an :class:InMemoryOmissionStore built here is unset-and-forget: it is process-local and lost on restart.

omission_key ¶

omission_key(user_id: str, session_id: str, chatbot_id: Optional[str]) -> str

Compose the omission-store scoping key for one session.

PARAMETER	DESCRIPTION
`user_id`	Owner of the conversation. TYPE: `str`
`session_id`	Conversation session. TYPE: `str`
`chatbot_id`	Agent attribution; `None` becomes `"_default"`. TYPE: `Optional[str]`

RETURNS	DESCRIPTION
`str`	`"{chatbot_id}:{user_id}:{session_id}"`.

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

def omission_key(self, user_id: str, session_id: str, chatbot_id: Optional[str]) -> str:
    """Compose the omission-store scoping key for one session.

    Args:
        user_id: Owner of the conversation.
        session_id: Conversation session.
        chatbot_id: Agent attribution; ``None`` becomes ``"_default"``.

    Returns:
        ``"{chatbot_id}:{user_id}:{session_id}"``.
    """
    return f"{chatbot_id or '_default'}:{user_id}:{session_id}"

create_history `abstractmethod` `async` ¶

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

Create a new conversation history.

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

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

get_history `abstractmethod` `async` ¶

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

Get a conversation history.

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

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

update_history `abstractmethod` `async` ¶

update_history(history: ConversationHistory) -> None

Update a conversation history.

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

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

add_turn `async` ¶

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

Persist one turn: normalize, count, offload oversized outputs, write once.

Concrete template method (FEAT-525): normalizes (Stage 0, unless normalize=False), counts tokens (Stage 0.5, always-on), offloads any tool output above oversize_tool_tokens to the omission store with a short preview left in the turn, then delegates the single backend write to :meth:_store_turn. When compaction is given, folds it into the persisted metadata["compaction"] state (calibration EWMA, boundary, stage2_needed) in the same write.

PARAMETER	DESCRIPTION
`user_id`	Owner of the conversation. TYPE: `str`
`session_id`	Conversation session. TYPE: `str`
`turn`	The turn to persist. Not mutated when `normalize=True` (the default) — Stage 0 rebinds `turn` to a new, normalized copy first, so the token-count and oversize- offload steps that follow only touch that copy. When the caller opts out with `normalize=False`, that defensive copy is skipped and this method mutates the caller's own `turn`/`ToolInvocation` objects in place (stamps `schema_version`/`token_count`, and rewrites `inv.output`/`output_chars`/`omitted` for any oversize output) — do not reuse a `turn` object across calls after opting out of normalization. TYPE: `ConversationTurn`
`chatbot_id`	Agent attribution. TYPE: `Optional[str]` DEFAULT: `None`
`compaction`	The bot's commit for this round, or `None` for writers that do not participate in the budget round-trip (e.g. a partial-save on error, or the `ChatStorage` tier). TYPE: `Optional[CompactionCommit]` DEFAULT: `None`

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

async def add_turn(
    self,
    user_id: str,
    session_id: str,
    turn: ConversationTurn,
    chatbot_id: Optional[str] = None,
    *,
    compaction: Optional[CompactionCommit] = None,
) -> None:
    """Persist one turn: normalize, count, offload oversized outputs, write once.

    Concrete template method (FEAT-525): normalizes (Stage 0, unless
    ``normalize=False``), counts tokens (Stage 0.5, always-on),
    offloads any tool output above ``oversize_tool_tokens`` to the
    omission store with a short preview left in the turn, then
    delegates the single backend write to :meth:`_store_turn`. When
    ``compaction`` is given, folds it into the persisted
    ``metadata["compaction"]`` state (calibration EWMA, boundary,
    ``stage2_needed``) in the same write.

    Args:
        user_id: Owner of the conversation.
        session_id: Conversation session.
        turn: The turn to persist. Not mutated when ``normalize=True``
            (the default) — Stage 0 rebinds ``turn`` to a new,
            normalized copy first, so the token-count and oversize-
            offload steps that follow only touch that copy. When the
            caller opts out with ``normalize=False``, that defensive
            copy is skipped and this method mutates the caller's own
            ``turn``/``ToolInvocation`` objects in place (stamps
            ``schema_version``/``token_count``, and rewrites
            ``inv.output``/``output_chars``/``omitted`` for any
            oversize output) — do not reuse a ``turn`` object across
            calls after opting out of normalization.
        chatbot_id: Agent attribution.
        compaction: The bot's commit for this round, or ``None`` for
            writers that do not participate in the budget round-trip
            (e.g. a partial-save on error, or the ``ChatStorage`` tier).
    """
    from .compaction.tokens import count_turn, needs_recount

    counter = self.token_counter
    if self._normalize:
        from .compaction.normalize import normalize_turn

        turn = normalize_turn(turn)
    if needs_recount(turn, counter):
        turn.token_count = count_turn(turn, counter)

    key = self.omission_key(user_id, session_id, chatbot_id)
    offloaded = False
    for inv in turn.tool_invocations:
        if inv.output and "output" not in inv.omitted and counter.count(inv.output) > self._oversize_tool_tokens:
            cid = await self.omission_store.put(key, inv.output, turn_id=turn.turn_id)
            inv.output_chars = len(inv.output)
            inv.output = _preview(inv.output)
            inv.omitted["output"] = cid
            offloaded = True
    if offloaded:
        turn.token_count = count_turn(turn, counter)

    turn.schema_version = 2

    state: Optional[Dict[str, Any]] = None
    if compaction is not None:
        prev = await self._get_compaction_state(user_id, session_id, chatbot_id)
        state = apply_commit(
            CompactionState.from_dict(prev) if prev else None,
            compaction,
            counter.name,
            _provider_prompt_tokens(turn),
        ).to_dict()

    await self._store_turn(user_id, session_id, turn, chatbot_id, compaction_state=state)

report_usage `async` ¶

report_usage(user_id: str, session_id: str, *, estimated_prompt_tokens: int, provider_prompt_tokens: Optional[int], chatbot_id: Optional[str] = None) -> None

Fold one (estimate, provider) observation into the calibration state, without writing a turn.

Standalone counterpart to the calibration folded into :meth:add_turn via compaction= — used by partial-save paths (e.g. ask_stream on error) and tests that need to update calibration independently of a turn write.

PARAMETER	DESCRIPTION
`user_id`	Owner of the conversation. TYPE: `str`
`session_id`	Conversation session. TYPE: `str`
`estimated_prompt_tokens`	The bot's own token estimate for the round. TYPE: `int`
`provider_prompt_tokens`	The provider-reported prompt token count for the same round, when available. TYPE: `Optional[int]`
`chatbot_id`	Agent attribution. TYPE: `Optional[str]` DEFAULT: `None`

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

async def report_usage(
    self,
    user_id: str,
    session_id: str,
    *,
    estimated_prompt_tokens: int,
    provider_prompt_tokens: Optional[int],
    chatbot_id: Optional[str] = None,
) -> None:
    """Fold one (estimate, provider) observation into the calibration state, without writing a turn.

    Standalone counterpart to the calibration folded into
    :meth:`add_turn` via ``compaction=`` — used by partial-save paths
    (e.g. ``ask_stream`` on error) and tests that need to update
    calibration independently of a turn write.

    Args:
        user_id: Owner of the conversation.
        session_id: Conversation session.
        estimated_prompt_tokens: The bot's own token estimate for the
            round.
        provider_prompt_tokens: The provider-reported prompt token
            count for the same round, when available.
        chatbot_id: Agent attribution.
    """
    history = await self.get_history(user_id, session_id, chatbot_id)
    if history is None:
        return
    prev = history.metadata.get("compaction")
    state = CompactionState.from_dict(prev) if prev else CompactionState(tokenizer=self.token_counter.name)
    state = apply_usage(state, estimated_prompt_tokens, provider_prompt_tokens)
    history.metadata["compaction"] = state.to_dict()
    await self.update_history(history)

clear_history `abstractmethod` `async` ¶

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

Clear a conversation history.

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

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

list_sessions `abstractmethod` `async` ¶

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

List all session IDs for a user.

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

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

delete_history `abstractmethod` `async` ¶

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

Delete a conversation history entirely.

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

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

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, tool_invocations: List[ToolInvocation] = list(), error: Optional[str] = None, token_count: Optional[TokenCount] = None, state: TurnState = TurnState.RAW, schema_version: int = 1, norm_version: 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

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,
        "tool_invocations": [inv.to_dict() for inv in self.tool_invocations],
        "error": self.error,
        "token_count": self.token_count.to_dict() if self.token_count else None,
        "state": self.state.value,
        "schema_version": self.schema_version,
        "norm_version": self.norm_version,
    }

from_dict `classmethod` ¶

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

Deserialize turn from dictionary.

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

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "ConversationTurn":
    """Deserialize turn from dictionary."""
    token_count_data = data.get("token_count")
    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"),
        tool_invocations=[ToolInvocation.from_dict(d) for d in data.get("tool_invocations", []) or []],
        error=data.get("error"),
        token_count=TokenCount.from_dict(token_count_data) if token_count_data else None,
        state=TurnState(data.get("state", TurnState.RAW.value)),
        schema_version=data.get("schema_version", 1),
        norm_version=data.get("norm_version"),
    )

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, error: Optional[str] = None, tool_invocations: Optional[List[ToolInvocation]] = 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`
`error`	Round-level failure text (FEAT-525). Condensed by Stage 0 rule 5 when the turn is normalized; never omitted. TYPE: `Optional[str]` DEFAULT: `None`
`tool_invocations`	Captured invocations to persist INSTEAD OF deriving them from `response.tool_calls` (FEAT-538). When the task-memory observer watched this turn it already holds the canonical record of every dispatch — including the ones that never produced a `tool_call` because they were denied, cancelled or never executed. Passing them here replaces the derivation rather than adding to it, which is what stops an observed turn listing each call twice. `None` keeps the legacy conversion, so an unobserved turn is byte-identical to before. TYPE: `Optional[List[ToolInvocation]]` DEFAULT: `None`

RETURNS	DESCRIPTION
`ConversationTurn`	A fully populated :class:`ConversationTurn`.

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

@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,
    error: Optional[str] = None,
    tool_invocations: Optional[List[ToolInvocation]] = 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.
        error: Round-level failure text (FEAT-525). Condensed by Stage 0
            rule 5 when the turn is normalized; never omitted.
        tool_invocations: Captured invocations to persist INSTEAD OF
            deriving them from ``response.tool_calls`` (FEAT-538).
            When the task-memory observer watched this turn it already
            holds the canonical record of every dispatch — including
            the ones that never produced a ``tool_call`` because they
            were denied, cancelled or never executed. Passing them
            here *replaces* the derivation rather than adding to it,
            which is what stops an observed turn listing each call
            twice. ``None`` keeps the legacy conversion, so an
            unobserved turn is byte-identical to before.

    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)

    derived_invocations = [
        ToolInvocation(
            tool_name=tc.name,
            input=tc.arguments,
            output=_stringify(tc.result),
            status=ToolStatus.ERROR if tc.error else ToolStatus.COMPLETED,
            error=tc.error,
            elapsed_ms=(int(tc.execution_time * 1000) if tc.execution_time is not None else None),
            wm_key=_tee_key(tc.result),
        )
        for tc in tool_calls
    ]

    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,
        tool_invocations=(derived_invocations if tool_invocations is None else list(tool_invocations)),
        error=error,
    )

AnswerMemory ¶

AnswerMemory(agent_id: str)

Store and retrieve question/answer interactions by turn identifier.

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

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

store_interaction `async` ¶

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

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

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

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

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

get `async` ¶

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

Retrieve a stored interaction by turn identifier.

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

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

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

CompactionCommit `dataclass` ¶

CompactionCommit(prompt_estimate: int, boundary_turn_id: Optional[str], stage2_needed: bool, history_estimate: int = 0, dropped_turns: int = 0)

What the bot hands to save_conversation_turn after the round.

ATTRIBUTE	DESCRIPTION
`prompt_estimate`	`tokens(rendered history) + tokens(system_prompt) + tokens(prompt)`. TYPE: `int`
`boundary_turn_id`	The new monotonic boundary. TYPE: `Optional[str]`
`stage2_needed`	Whether Stage 2 is needed for this session. TYPE: `bool`
`history_estimate`	Telemetry only — the compacted history's total estimated size. Ignored by :func:`apply_commit`; populated by `AbstractBot.save_conversation_turn` to feed `Stage2CompactionNeededEvent` without a second channel. TYPE: `int`
`dropped_turns`	Telemetry only — count of turns dropped by the last compaction pass. Ignored by :func:`apply_commit`. TYPE: `int`

CompactionResult `dataclass` ¶

CompactionResult(views: Tuple[TurnView, ...], omissions: Tuple[Omission, ...], history_estimate: int, boundary_turn_id: Optional[str], stage2_needed: bool, dropped_turn_ids: Tuple[str, ...])

Output of :func:parrot.memory.compaction.compact.compact_history.

ContextBudget `dataclass` ¶

ContextBudget(window: int, reserve_output: int = 8192, reserve_fixed: int = 4096, high_watermark: float = 0.8, low_watermark: float = 0.6, max_turns: int = 30, verbatim_tokens: int = 15000, min_verbatim_turns: int = 2, oversize_tool_tokens: int = 2000, tool_activity_limit: Limit = Limit())

Token budget and retention configuration for one bot/agent.

ATTRIBUTE	DESCRIPTION
`window`	The provider's context window, from `MODEL_WINDOWS` when the model is known, else :data:`FALLBACK_WINDOW`. TYPE: `int`
`reserve_output`	Tokens reserved for the model's own output. TYPE: `int`
`reserve_fixed`	Tokens reserved for the system prompt, tool schemas and provider framing allowance. TYPE: `int`
`high_watermark`	Fraction of `available` the pruned tier may fill. TYPE: `float`
`low_watermark`	Reserved for Stage 2 (target after summarization); unused by the deterministic three-tier walk. TYPE: `float`
`max_turns`	Unified safety ceiling on the number of turns considered (was `AbstractBot` 50 / `Chatbot` 5 before this feature). TYPE: `int`
`verbatim_tokens`	Cumulative token budget for the verbatim (RAW) tier. TYPE: `int`
`min_verbatim_turns`	Minimum number of turns kept verbatim, regardless of size. TYPE: `int`
`oversize_tool_tokens`	Any tool output above this size is pruned from every turn but the newest, even inside the verbatim tier. TYPE: `int`
`tool_activity_limit`	Bound applied when rendering the RAW `<tool-activity>` block. TYPE: `Limit`

available `property` ¶

available: int

Tokens available for history after reserves (never negative).

Limit `dataclass` ¶

Limit(max_invocations: int = 12, max_input_chars: int = 200, max_output_chars: int = 400, max_block_tokens: int = 1500)

Bound on the RAW <tool-activity> block.

Keeps a chatty recent turn from blowing the whole budget by itself.

OmissionStore ¶

OmissionStore(*, ttl: Optional[int] = None)

Bases: ABC

Content-addressed store for content pruned or offloaded from turns.

Initialize the store.

PARAMETER	DESCRIPTION
`ttl`	Expiry in seconds for stored content. `None` (default) means no expiry, matching the history's own lack of a TTL. TYPE: `Optional[int]` DEFAULT: `None`

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

def __init__(self, *, ttl: Optional[int] = None) -> None:
    """Initialize the store.

    Args:
        ttl: Expiry in seconds for stored content. ``None`` (default)
            means no expiry, matching the history's own lack of a TTL.
    """
    self.ttl = ttl

put `abstractmethod` `async` ¶

put(session_key: str, content: str, *, turn_id: Optional[str] = None) -> str

Store content under session_key, indexed by turn_id.

Idempotent: storing the same content twice returns the same id and does not create a duplicate index entry.

PARAMETER	DESCRIPTION
`session_key`	Opaque scoping key (composed by the memory). TYPE: `str`
`content`	The full text to store. TYPE: `str`
`turn_id`	When given, indexes the resulting id under this turn. TYPE: `Optional[str]` DEFAULT: `None`

RETURNS	DESCRIPTION
`str`	The content id.

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

@abstractmethod
async def put(self, session_key: str, content: str, *, turn_id: Optional[str] = None) -> str:
    """Store ``content`` under ``session_key``, indexed by ``turn_id``.

    Idempotent: storing the same content twice returns the same id
    and does not create a duplicate index entry.

    Args:
        session_key: Opaque scoping key (composed by the memory).
        content: The full text to store.
        turn_id: When given, indexes the resulting id under this turn.

    Returns:
        The content id.
    """

put_many `async` ¶

put_many(session_key: str, omissions: Sequence[Omission]) -> None

Store every :class:Omission under its own turn_id (default: loop over :meth:put).

PARAMETER	DESCRIPTION
`session_key`	Opaque scoping key (composed by the memory). TYPE: `str`
`omissions`	The omissions to flush. TYPE: `Sequence[Omission]`

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

async def put_many(self, session_key: str, omissions: Sequence[Omission]) -> None:
    """Store every :class:`Omission` under its own ``turn_id`` (default: loop over :meth:`put`).

    Args:
        session_key: Opaque scoping key (composed by the memory).
        omissions: The omissions to flush.
    """
    for omission in omissions:
        await self.put(session_key, omission.content, turn_id=omission.turn_id)

get `abstractmethod` `async` ¶

get(session_key: str, content_id: str) -> Optional[str]

Return the stored content for content_id, or None if unknown.

PARAMETER	DESCRIPTION
`session_key`	Opaque scoping key (composed by the memory). TYPE: `str`
`content_id`	The id returned by a previous :meth:`put`. TYPE: `str`

RETURNS	DESCRIPTION
`Optional[str]`	The stored content, or `None` when the id is unknown or
`Optional[str]`	belongs to a different session.

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

@abstractmethod
async def get(self, session_key: str, content_id: str) -> Optional[str]:
    """Return the stored content for ``content_id``, or ``None`` if unknown.

    Args:
        session_key: Opaque scoping key (composed by the memory).
        content_id: The id returned by a previous :meth:`put`.

    Returns:
        The stored content, or ``None`` when the id is unknown or
        belongs to a different session.
    """

probe `async` ¶

probe(session_key: str, content_id: str) -> Optional[bool]

Report whether content_id is still stored, WITHOUT loading it.

Added by FEAT-538. :meth:get fetches the payload text, so using it to answer "does this still exist?" would drag potentially large omitted content across the wire purely to compute a boolean — and recall does exactly that check, for every omission reference it reports.

Deliberately not abstract: a store written before this method existed keeps working and inherits this default, which answers None (unknown). Unknown is never reported as available — claiming content is present when the store cannot say so would let recall promise recoverable text that may be long gone.

PARAMETER	DESCRIPTION
`session_key`	Opaque scoping key (composed by the memory). TYPE: `str`
`content_id`	The id to probe. TYPE: `str`

RETURNS	DESCRIPTION
`Optional[bool]`	`True` when the content is stored, `False` when it is
`Optional[bool]`	known to be absent or expired, and `None` when this backend
`Optional[bool]`	cannot answer cheaply.

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

async def probe(self, session_key: str, content_id: str) -> Optional[bool]:
    """Report whether ``content_id`` is still stored, WITHOUT loading it.

    Added by FEAT-538. :meth:`get` fetches the payload text, so using
    it to answer "does this still exist?" would drag potentially
    large omitted content across the wire purely to compute a
    boolean — and recall does exactly that check, for every omission
    reference it reports.

    Deliberately **not** abstract: a store written before this method
    existed keeps working and inherits this default, which answers
    ``None`` (*unknown*). Unknown is never reported as available —
    claiming content is present when the store cannot say so would
    let recall promise recoverable text that may be long gone.

    Args:
        session_key: Opaque scoping key (composed by the memory).
        content_id: The id to probe.

    Returns:
        ``True`` when the content is stored, ``False`` when it is
        known to be absent or expired, and ``None`` when this backend
        cannot answer cheaply.
    """
    return None

probe_many `async` ¶

probe_many(session_key: str, content_ids: Sequence[str]) -> Dict[str, Optional[bool]]

Probe several ids at once (default: loop over :meth:probe).

Backends with a batching primitive should override this; the default keeps the contract satisfiable by any store.

PARAMETER	DESCRIPTION
`session_key`	Opaque scoping key (composed by the memory). TYPE: `str`
`content_ids`	The ids to probe. TYPE: `Sequence[str]`

RETURNS	DESCRIPTION
`Dict[str, Optional[bool]]`	A mapping of id to availability, using :meth:`probe`'s
`Dict[str, Optional[bool]]`	three-valued answer.

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

async def probe_many(self, session_key: str, content_ids: Sequence[str]) -> Dict[str, Optional[bool]]:
    """Probe several ids at once (default: loop over :meth:`probe`).

    Backends with a batching primitive should override this; the
    default keeps the contract satisfiable by any store.

    Args:
        session_key: Opaque scoping key (composed by the memory).
        content_ids: The ids to probe.

    Returns:
        A mapping of id to availability, using :meth:`probe`'s
        three-valued answer.
    """
    return {cid: await self.probe(session_key, cid) for cid in content_ids}

list_by_turn `abstractmethod` `async` ¶

list_by_turn(session_key: str, turn_id: str) -> List[str]

Return the content ids stored for turn_id, in insertion order.

PARAMETER	DESCRIPTION
`session_key`	Opaque scoping key (composed by the memory). TYPE: `str`
`turn_id`	The turn to look up. TYPE: `str`

RETURNS	DESCRIPTION
`List[str]`	A list of content ids (empty when the turn has none).

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

@abstractmethod
async def list_by_turn(self, session_key: str, turn_id: str) -> List[str]:
    """Return the content ids stored for ``turn_id``, in insertion order.

    Args:
        session_key: Opaque scoping key (composed by the memory).
        turn_id: The turn to look up.

    Returns:
        A list of content ids (empty when the turn has none).
    """

clear `abstractmethod` `async` ¶

clear(session_key: str) -> None

Delete every stored content and index entry for session_key.

PARAMETER	DESCRIPTION
`session_key`	Opaque scoping key (composed by the memory). TYPE: `str`

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

@abstractmethod
async def clear(self, session_key: str) -> None:
    """Delete every stored content and index entry for ``session_key``.

    Args:
        session_key: Opaque scoping key (composed by the memory).
    """

TokenCount `dataclass` ¶

TokenCount(user: int, assistant: int, tools: int, total: int, tokenizer: str)

Per-turn token accounting.

context_used is deliberately NOT counted here (spec decision).

to_dict ¶

to_dict() -> Dict[str, int | str]

Serialize this token count to a plain dict.

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

def to_dict(self) -> Dict[str, int | str]:
    """Serialize this token count to a plain dict."""
    return {
        "user": self.user,
        "assistant": self.assistant,
        "tools": self.tools,
        "total": self.total,
        "tokenizer": self.tokenizer,
    }

from_dict `classmethod` ¶

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

Deserialize a token count from a dict.

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

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "TokenCount":
    """Deserialize a token count from a dict."""
    return cls(
        user=int(data.get("user", 0)),
        assistant=int(data.get("assistant", 0)),
        tools=int(data.get("tools", 0)),
        total=int(data.get("total", 0)),
        tokenizer=data.get("tokenizer", "heuristic"),
    )

TokenCounter ¶

Bases: Protocol

A named, synchronous text tokenizer.

count ¶

count(text: str) -> int

Return the token count for text.

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

def count(self, text: str) -> int:
    """Return the token count for ``text``."""
    ...

ToolInvocation `dataclass` ¶

ToolInvocation(tool_name: str, input: Dict[str, Any], output: Optional[str] = None, status: ToolStatus = ToolStatus.COMPLETED, error: Optional[str] = None, elapsed_ms: Optional[int] = None, output_chars: Optional[int] = None, omitted: Dict[str, str] = dict(), wm_key: Optional[str] = None)

One tool call captured from AIMessage.tool_calls.

See parrot/models/basic.py:23-30 for the source ToolCall shape.

ATTRIBUTE	DESCRIPTION
`tool_name`	Name of the invoked tool. TYPE: `str`
`input`	Canonical (`orjson` `OPT_SORT_KEYS`) arguments after Stage 0 normalization. TYPE: `Dict[str, Any]`
`output`	Full text, or the short preview left behind once the original output has been offloaded to the omission store. TYPE: `Optional[str]`
`status`	Whether the invocation completed or errored. TYPE: `ToolStatus`
`error`	Condensed error/traceback text. Never omitted (spec G7/C7). TYPE: `Optional[str]`
`elapsed_ms`	Wall-clock duration of the invocation, in milliseconds. TYPE: `Optional[int]`
`output_chars`	Length of the ORIGINAL output, captured before any offload so the preview notice can report the true size. TYPE: `Optional[int]`
`omitted`	Maps a field name (`"output"` is the only one used in v1) to the `content_id` ("om_...") it was offloaded to. TYPE: `Dict[str, str]`
`wm_key`	The FEAT-380 working-memory tee key copied from `result["_tee"]["key"]` when the tool result carries one. TYPE: `Optional[str]`

to_dict ¶

to_dict() -> Dict[str, Any]

Serialize this invocation to a plain dict.

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

def to_dict(self) -> Dict[str, Any]:
    """Serialize this invocation to a plain dict."""
    return {
        "tool_name": self.tool_name,
        "input": self.input,
        "output": self.output,
        "status": self.status.value,
        "error": self.error,
        "elapsed_ms": self.elapsed_ms,
        "output_chars": self.output_chars,
        "omitted": dict(self.omitted),
        "wm_key": self.wm_key,
    }

from_dict `classmethod` ¶

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

Deserialize an invocation from a dict, tolerant of missing keys.

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

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "ToolInvocation":
    """Deserialize an invocation from a dict, tolerant of missing keys."""
    status_raw = data.get("status", ToolStatus.COMPLETED.value)
    return cls(
        tool_name=data.get("tool_name", ""),
        input=data.get("input", {}) or {},
        output=data.get("output"),
        status=ToolStatus(status_raw) if status_raw else ToolStatus.COMPLETED,
        error=data.get("error"),
        elapsed_ms=data.get("elapsed_ms"),
        output_chars=data.get("output_chars"),
        omitted=dict(data.get("omitted") or {}),
        wm_key=data.get("wm_key"),
    )

ToolStatus ¶

Bases: str, Enum

Outcome of a single tool invocation.

TurnState ¶

Bases: str, Enum

Lifecycle state of a conversation turn.

Storage always writes RAW in v1. PRUNED is a view-only state produced by :func:parrot.memory.compaction.compact.compact_history. SUMMARIZED is reserved for Stage 2 (LLM summary turns) and is never produced by this feature.

TurnView `dataclass` ¶

TurnView(turn_id: str, chatbot_id: Optional[str], user_text: str, assistant_text: str, assistant_suffix: str, state: TurnState, estimated_tokens: int)

Materialized text for one turn.

render_history only concatenates user_text and assistant_text + assistant_suffix — it never re-derives them.

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

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

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

record_episode `async` ¶

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

Record a new episode with auto-enrichment.

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

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

RETURNS	DESCRIPTION
`EpisodicMemory`	The stored EpisodicMemory with all enrichments.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

record_tool_episode `async` ¶

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

Record an episode from a tool execution.

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

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

RETURNS	DESCRIPTION
`EpisodicMemory`	The stored episode.

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

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

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

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

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

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

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

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

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

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

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

record_crew_episode `async` ¶

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

Record episodes from a crew execution.

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

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

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

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

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

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

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

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

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

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

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

    return episodes

recall_similar `async` ¶

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

Recall episodes similar to a query.

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

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

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

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

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

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

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

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

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

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

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

    return results[:top_k]

get_failure_warnings `async` ¶

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

Generate injectable warning text from past failures.

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

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

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

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

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

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

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

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

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

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

    if not failures:
        return ""

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

    # Format warnings
    lines_mistakes = []
    lines_success = []

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

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

    return "\n".join(parts)

get_user_preferences `async` ¶

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

Get user preference episodes.

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

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

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

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

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

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

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

get_room_context `async` ¶

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

Get recent episodes for a room.

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

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

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

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

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

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

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

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

    return episodes[:limit]

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

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

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

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

compact_namespace `async` ¶

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

Compact a namespace by keeping only the most important episodes.

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

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

RETURNS	DESCRIPTION
`int`	Number of episodes deleted.

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

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

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

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

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

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

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

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

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

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

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

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

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

    return deleted

export_episodes `async` ¶

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

Export episodes as JSONL string.

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

RETURNS	DESCRIPTION
`str`	JSONL-formatted string of episodes.

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

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

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

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

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

create_pgvector `async` `classmethod` ¶

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

Create a store with PgVector backend.

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

RETURNS	DESCRIPTION
`'EpisodicMemoryStore'`	Configured EpisodicMemoryStore with PgVector backend.

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

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

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

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

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

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

create_redis_vector `async` `classmethod` ¶

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

Create a store with RedisVectorBackend.

Requires Redis Stack with RediSearch module enabled.

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

RETURNS	DESCRIPTION
`'EpisodicMemoryStore'`	Configured EpisodicMemoryStore with RedisVectorBackend.

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

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

    Requires Redis Stack with RediSearch module enabled.

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

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

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

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

create_faiss `async` `classmethod` ¶

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

Create a store with FAISS backend.

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

RETURNS	DESCRIPTION
`'EpisodicMemoryStore'`	Configured EpisodicMemoryStore with FAISS backend.

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

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

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

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

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

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

EpisodicMemoryToolkit ¶

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

Bases: AbstractToolkit

Toolkit exposing episodic memory as agent-callable tools.

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

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

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

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

search_episodic_memory `async` ¶

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

Search past agent experiences by semantic similarity.

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

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

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

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

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

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

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

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

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

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

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

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

record_lesson `async` ¶

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

Explicitly record a lesson learned for future reference.

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

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

RETURNS	DESCRIPTION
`str`	Confirmation that the lesson was recorded.

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

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

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

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

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

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

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

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

get_warnings `async` ¶

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

Get warnings about past mistakes relevant to the current task.

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

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

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

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

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

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

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

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

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

        return warnings

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

FileConversationMemory ¶

FileConversationMemory(base_path: str = './conversations', *, token_counter: Optional[TokenCounter] = None, omission_store: Optional[OmissionStore] = None, normalize: bool = True)

Bases: ConversationMemory

File-based implementation of conversation memory.

Initialize the store.

PARAMETER	DESCRIPTION
`base_path`	Root directory for conversation history files. TYPE: `str` DEFAULT: `'./conversations'`
`token_counter`	Stage 0.5 counter override (see :class:`~parrot.memory.abstract.ConversationMemory`). TYPE: `Optional[TokenCounter]` DEFAULT: `None`
`omission_store`	Omission-store override; defaults to a :class:`FileOmissionStore` sharing `base_path`. TYPE: `Optional[OmissionStore]` DEFAULT: `None`
`normalize`	Disables Stage 0 for this instance when `False`. TYPE: `bool` DEFAULT: `True`

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

def __init__(
    self,
    base_path: str = "./conversations",
    *,
    token_counter: Optional["TokenCounter"] = None,
    omission_store: Optional[OmissionStore] = None,
    normalize: bool = True,
) -> None:
    """Initialize the store.

    Args:
        base_path: Root directory for conversation history files.
        token_counter: Stage 0.5 counter override (see
            :class:`~parrot.memory.abstract.ConversationMemory`).
        omission_store: Omission-store override; defaults to a
            :class:`FileOmissionStore` sharing ``base_path``.
        normalize: Disables Stage 0 for this instance when ``False``.
    """
    self.base_path = Path(base_path)
    super().__init__(
        token_counter=token_counter,
        omission_store=omission_store or FileOmissionStore(self.base_path),
        normalize=normalize,
    )
    self.base_path.mkdir(exist_ok=True)
    self._lock = asyncio.Lock()

create_history `async` ¶

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

Create a new conversation history.

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

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

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

        return history

get_history `async` ¶

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

Get a conversation history, 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

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

async def update_history(self, history: ConversationHistory) -> None:
    """Update a conversation history."""
    async with self._lock:
        await self._write_history(history)

clear_history `async` ¶

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

Clear a conversation history.

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

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

list_sessions `async` ¶

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

List all session IDs for a user.

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

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

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

        return sessions

delete_history `async` ¶

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

Delete a conversation history entirely.

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

async def delete_history(self, user_id: str, session_id: str, chatbot_id: Optional[str] = None) -> bool:
    """Delete a conversation history entirely."""
    await self.omission_store.clear(self.omission_key(user_id, session_id, chatbot_id))
    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(*, token_counter: Optional[TokenCounter] = None, omission_store: Optional[OmissionStore] = None, normalize: bool = True)

Bases: ConversationMemory

In-memory implementation of conversation memory.

Initialize the store.

PARAMETER	DESCRIPTION
`token_counter`	Stage 0.5 counter override (see :class:`~parrot.memory.abstract.ConversationMemory`). TYPE: `Optional[TokenCounter]` DEFAULT: `None`
`omission_store`	Omission-store override; defaults to a fresh :class:`InMemoryOmissionStore`. TYPE: `Optional[OmissionStore]` DEFAULT: `None`
`normalize`	Disables Stage 0 for this instance when `False`. TYPE: `bool` DEFAULT: `True`

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

def __init__(
    self,
    *,
    token_counter: Optional["TokenCounter"] = None,
    omission_store: Optional[OmissionStore] = None,
    normalize: bool = True,
) -> None:
    """Initialize the store.

    Args:
        token_counter: Stage 0.5 counter override (see
            :class:`~parrot.memory.abstract.ConversationMemory`).
        omission_store: Omission-store override; defaults to a fresh
            :class:`InMemoryOmissionStore`.
        normalize: Disables Stage 0 for this instance when ``False``.
    """
    super().__init__(
        token_counter=token_counter,
        omission_store=omission_store or InMemoryOmissionStore(),
        normalize=normalize,
    )
    self._histories: Dict[str, Dict[str, Dict[str, ConversationHistory]]] = {}

create_history `async` ¶

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

Create a new conversation history.

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

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

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

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

get_history `async` ¶

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

Get a conversation history.

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

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

update_history `async` ¶

update_history(history: ConversationHistory) -> None

Update a conversation history.

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

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

clear_history `async` ¶

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

Clear a conversation history.

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

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

list_sessions `async` ¶

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

List all session IDs for a user.

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

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

delete_history `async` ¶

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

Delete a conversation history entirely.

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

async def delete_history(self, user_id: str, session_id: str, chatbot_id: Optional[str] = None) -> bool:
    """Delete a conversation history entirely."""
    await self.omission_store.clear(self.omission_key(user_id, session_id, chatbot_id))

    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, *, token_counter: Optional[TokenCounter] = None, omission_store: Optional[OmissionStore] = None, normalize: bool = True, omission_ttl: Optional[int] = None, history_ttl: Optional[int] = None)

Bases: ConversationMemory

Redis-based conversation memory with proper encoding handling.

Initialize the store.

PARAMETER	DESCRIPTION
`redis_url`	Redis connection URL; defaults to `REDIS_HISTORY_URL`. TYPE: `str` DEFAULT: `None`
`key_prefix`	Prefix for every history key this instance owns (e.g. `"chat"` for the `ChatStorage` cold tier). TYPE: `str` DEFAULT: `'conversation'`
`use_hash_storage`	Whether to use Redis hashes (recommended) or plain string keys. TYPE: `bool` DEFAULT: `True`
`token_counter`	Stage 0.5 counter override (see :class:`~parrot.memory.abstract.ConversationMemory`). TYPE: `Optional[TokenCounter]` DEFAULT: `None`
`omission_store`	Omission-store override; defaults to a :class:`~parrot.memory.compaction.omission.RedisOmissionStore` sharing this instance's Redis client. TYPE: `Optional[OmissionStore]` DEFAULT: `None`
`normalize`	Disables Stage 0 for this instance when `False`. TYPE: `bool` DEFAULT: `True`
`omission_ttl`	Expiry (seconds) for the default omission store; `None` (default) means no expiry, independent of the (TTL-less) history keys. TYPE: `Optional[int]` DEFAULT: `None`
`history_ttl`	Expiry (seconds) refreshed on every write to a history key, so an idle conversation expires. `None` (default) keeps history keys TTL-less. TYPE: `Optional[int]` DEFAULT: `None`

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

def __init__(
    self,
    redis_url: str = None,
    key_prefix: str = "conversation",
    use_hash_storage: bool = True,
    *,
    token_counter: Optional["TokenCounter"] = None,
    omission_store: Optional[OmissionStore] = None,
    normalize: bool = True,
    omission_ttl: Optional[int] = None,
    history_ttl: Optional[int] = None,
) -> None:
    """Initialize the store.

    Args:
        redis_url: Redis connection URL; defaults to
            ``REDIS_HISTORY_URL``.
        key_prefix: Prefix for every history key this instance owns
            (e.g. ``"chat"`` for the ``ChatStorage`` cold tier).
        use_hash_storage: Whether to use Redis hashes (recommended) or
            plain string keys.
        token_counter: Stage 0.5 counter override (see
            :class:`~parrot.memory.abstract.ConversationMemory`).
        omission_store: Omission-store override; defaults to a
            :class:`~parrot.memory.compaction.omission.RedisOmissionStore`
            sharing this instance's Redis client.
        normalize: Disables Stage 0 for this instance when ``False``.
        omission_ttl: Expiry (seconds) for the default omission store;
            ``None`` (default) means no expiry, independent of the
            (TTL-less) history keys.
        history_ttl: Expiry (seconds) refreshed on every write to a
            history key, so an idle conversation expires. ``None``
            (default) keeps history keys TTL-less.
    """
    self.history_ttl = history_ttl if history_ttl and history_ttl > 0 else None
    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,
    )
    super().__init__(
        token_counter=token_counter,
        omission_store=omission_store
        or RedisOmissionStore(self.redis, key_prefix=self.key_prefix, ttl=omission_ttl),
        normalize=normalize,
    )

create_history `async` ¶

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

Create a new conversation history.

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

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

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

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

    # Add to user sessions set
    await self.redis.sadd(self._get_user_sessions_key(user_id, chatbot_id), session_id)
    await self._refresh_ttl(user_id, session_id, chatbot_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

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

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)
    await self._refresh_ttl(history.user_id, history.session_id, history.chatbot_id)

read_metadata `async` ¶

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

Read a conversation's whole metadata blob (FEAT-538).

In hash mode this is a single targeted hget, avoiding the FEAT-524 lazy legacy re-key that a full :meth:get_history may perform as a side effect.

PARAMETER	DESCRIPTION
`user_id`	Owner of the conversation. TYPE: `str`
`session_id`	Conversation session. TYPE: `str`
`chatbot_id`	Agent key segment. TYPE: `Optional[str]` DEFAULT: `None`

RETURNS	DESCRIPTION
`Optional[Dict[str, Any]]`	The metadata dict, or `None` when the conversation has none.

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

async def read_metadata(
    self, user_id: str, session_id: str, chatbot_id: Optional[str] = None
) -> Optional[Dict[str, Any]]:
    """Read a conversation's whole ``metadata`` blob (FEAT-538).

    In hash mode this is a single targeted ``hget``, avoiding the
    FEAT-524 lazy legacy re-key that a full :meth:`get_history` may
    perform as a side effect.

    Args:
        user_id: Owner of the conversation.
        session_id: Conversation session.
        chatbot_id: Agent key segment.

    Returns:
        The metadata dict, or ``None`` when the conversation has none.
    """
    if self.use_hash_storage:
        raw = await self.redis.hget(self._get_key(user_id, session_id, chatbot_id), "metadata")
        if not raw:
            return None
        meta = self._deserialize_data(raw)
        return meta if isinstance(meta, dict) else None
    history = await self.get_history(user_id, session_id, chatbot_id)
    return dict(history.metadata) if history is not None else None

merge_metadata `async` ¶

merge_metadata(user_id: str, session_id: str, chatbot_id: Optional[str] = None, *, mutate: Callable[[Dict[str, Any]], Dict[str, Any]], ttl: Optional[int] = None, max_attempts: int = _MERGE_MAX_ATTEMPTS) -> Dict[str, Any]

Atomically compare-and-merge a conversation's metadata (FEAT-538).

metadata is shared by writers that know nothing about each other: the compaction layer owns metadata["compaction"] and task memory owns metadata["task_memory"]. The previous read-modify-write lost whichever update landed second.

A task-scoped lock cannot fix that, because the other writer would not take it. So this uses Redis' own WATCH/MULTI/EXEC: the transaction aborts if anyone touches the key in between, and mutate is re-run against the freshly read value. Correctness therefore does not depend on every writer cooperating — only on them going through Redis.

mutate must be a pure function of the metadata it is given: it may be called several times, and it must merge into the dict it receives rather than into one captured earlier.

PARAMETER	DESCRIPTION
`user_id`	Owner of the conversation. TYPE: `str`
`session_id`	Conversation session. TYPE: `str`
`chatbot_id`	Agent key segment. TYPE: `Optional[str]` DEFAULT: `None`
`mutate`	Pure function from current metadata to new metadata. TYPE: `Callable[[Dict[str, Any]], Dict[str, Any]]`
`ttl`	Optional expiry in seconds applied to the conversation key. Task-memory mode requires an explicit finite hot-key retention when the underlying history has no TTL. TYPE: `Optional[int]` DEFAULT: `None`
`max_attempts`	Bound on compare-and-swap retries. TYPE: `int` DEFAULT: `_MERGE_MAX_ATTEMPTS`

RETURNS	DESCRIPTION
`Dict[str, Any]`	The metadata that was written.

RAISES	DESCRIPTION
`RuntimeError`	If the merge still conflicts after `max_attempts`. Callers that must not fail a user action should treat this as a degraded cache write, not as a lost record.

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

async def merge_metadata(
    self,
    user_id: str,
    session_id: str,
    chatbot_id: Optional[str] = None,
    *,
    mutate: Callable[[Dict[str, Any]], Dict[str, Any]],
    ttl: Optional[int] = None,
    max_attempts: int = _MERGE_MAX_ATTEMPTS,
) -> Dict[str, Any]:
    """Atomically compare-and-merge a conversation's ``metadata`` (FEAT-538).

    ``metadata`` is shared by writers that know nothing about each
    other: the compaction layer owns ``metadata["compaction"]`` and
    task memory owns ``metadata["task_memory"]``. The previous
    read-modify-write lost whichever update landed second.

    A task-scoped lock cannot fix that, because **the other writer
    would not take it**. So this uses Redis' own
    ``WATCH``/``MULTI``/``EXEC``: the transaction aborts if *anyone*
    touches the key in between, and ``mutate`` is re-run against the
    freshly read value. Correctness therefore does not depend on every
    writer cooperating — only on them going through Redis.

    ``mutate`` must be a **pure** function of the metadata it is
    given: it may be called several times, and it must merge into the
    dict it receives rather than into one captured earlier.

    Args:
        user_id: Owner of the conversation.
        session_id: Conversation session.
        chatbot_id: Agent key segment.
        mutate: Pure function from current metadata to new metadata.
        ttl: Optional expiry in seconds applied to the conversation
            key. Task-memory mode requires an explicit finite hot-key
            retention when the underlying history has no TTL.
        max_attempts: Bound on compare-and-swap retries.

    Returns:
        The metadata that was written.

    Raises:
        RuntimeError: If the merge still conflicts after
            ``max_attempts``. Callers that must not fail a user
            action should treat this as a degraded cache write, not
            as a lost record.
    """
    key = self._get_key(user_id, session_id, chatbot_id)

    if not self._supports_cas():
        return await self._merge_metadata_unguarded(key, mutate, ttl)

    for attempt in range(max_attempts):
        async with self.redis.pipeline() as pipe:
            try:
                await pipe.watch(key)
                if self.use_hash_storage:
                    raw = await pipe.hget(key, "metadata")
                    current = self._deserialize_data(raw) if raw else {}
                    if not isinstance(current, dict):
                        current = {}
                    merged = mutate(dict(current))
                    pipe.multi()
                    pipe.hset(key, "metadata", self._serialize_data(merged))
                    if ttl:
                        pipe.expire(key, ttl)
                    await pipe.execute()
                else:
                    raw = await pipe.get(key)
                    payload = self._deserialize_data(raw) if raw else None
                    if not isinstance(payload, dict):
                        payload = {}
                    current = payload.get("metadata")
                    if not isinstance(current, dict):
                        current = {}
                    merged = mutate(dict(current))
                    payload["metadata"] = merged
                    pipe.multi()
                    pipe.set(key, self._serialize_data(payload))
                    if ttl:
                        pipe.expire(key, ttl)
                    await pipe.execute()
                return merged
            except WatchError:
                # Somebody else wrote the key between our read and our
                # write. Re-read and re-merge; do NOT reuse the stale
                # value, which is exactly how the update was lost.
                pass
        await _merge_backoff(attempt)

    raise RuntimeError(f"metadata merge for {key} still conflicting after {max_attempts} attempts")

clear_history `async` ¶

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

Clear a conversation history.

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

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

list_sessions `async` ¶

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

List all session IDs for a user.

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

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

delete_history `async` ¶

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

Delete a conversation history entirely.

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

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

close `async` ¶

close()

Close the Redis connection.

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

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

ping `async` ¶

ping() -> bool

Test Redis connection.

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

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

get_raw_data `async` ¶

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

Get raw data from Redis for debugging.

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

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

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

debug_conversation `async` ¶

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

Debug method to inspect conversation data.

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

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

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

list_sessions_by_chatbot `async` ¶

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

List all sessions for a specific chatbot.

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

RETURNS	DESCRIPTION
`List[str]`	List of session IDs

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

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

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

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

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

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

        if cursor == 0:
            break

    return sessions

get_chatbot_stats `async` ¶

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

Get statistics for a specific chatbot.

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

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

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

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

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

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

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

        if cursor == 0:
            break

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

delete_all_chatbot_conversations `async` ¶

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

Delete all conversations for a chatbot.

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

RETURNS	DESCRIPTION
`int`	Number of conversations deleted

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

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

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

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

    deleted_count = 0
    cursor = 0

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

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

        if cursor == 0:
            break

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

    return deleted_count

get_chatbot_users `async` ¶

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

Get all users who have interacted with a chatbot.

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

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

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

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

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

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

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

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

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

MemoryConfig ¶

Bases: BaseModel

Configuration for UnifiedMemoryManager.

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

MemoryContext ¶

Bases: BaseModel

Assembled context from all memory subsystems.

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

to_prompt_string ¶

to_prompt_string() -> str

Format as injectable system prompt sections.

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

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

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

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

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

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

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

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

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

    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

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

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

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

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

cleanup `async` ¶

cleanup() -> None

Release resources held by all non-None subsystems.

Calls cleanup() on each subsystem that exposes it.

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

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

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

get_context_for_query `async` ¶

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

Retrieve and assemble context from all memory subsystems.

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

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

RETURNS	DESCRIPTION
`MemoryContext`	`MemoryContext` assembled within the configured token budget.

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

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

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

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

    Returns:
        ``MemoryContext`` assembled within the configured token budget.
    """
    episodic_task = self._get_episodic_warnings(query)
    skills_task = self._get_relevant_skills(query)
    conversation_task = self._get_conversation(user_id, session_id)
    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

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)

compact_history ¶

compact_history(history: ConversationHistory, budget: ContextBudget, *, policies: Optional[Dict[str, PrunePolicy]] = None, boundary_turn_id: Optional[str] = None, counter: Optional[TokenCounter] = None, calibration: float = 1.0, current_chatbot_id: Optional[str] = None, include_other_agents: bool = True) -> CompactionResult

Walk a history newest → oldest and classify each turn RAW / PRUNED / DROPPED.

Pure and synchronous (spec G1/G9): never mutates history, its turns, or their invocations; performs no I/O; the same inputs always produce an equal :class:CompactionResult.

Three tiers (spec G10), walked newest → oldest with a running calibrated token sum:

Verbatim (RAW) while n_raw < budget.min_verbatim_turns or the cumulative size stays within both budget.verbatim_tokens and the high-watermark budget.
Pruned while the cumulative size stays within the high watermark (or the turn is the newest kept turn — the newest is never dropped).
Dropped beyond that — and every older turn is dropped too (tiers are contiguous); sets stage2_needed.

A persisted boundary_turn_id forces every turn at or before it to render PRUNED regardless of budget (monotonic: turns that once rendered PRUNED never render RAW again for this history). Any invocation whose output exceeds budget.oversize_tool_tokens is pruned in every turn but the newest, even inside the verbatim tier.

PARAMETER	DESCRIPTION
`history`	The history to compact. TYPE: `ConversationHistory`
`budget`	The retention configuration. TYPE: `ContextBudget`
`policies`	Per-tool prune-policy overrides, consulted before the module registry. TYPE: `Optional[Dict[str, PrunePolicy]]` DEFAULT: `None`
`boundary_turn_id`	The previously persisted boundary, if any. TYPE: `Optional[str]` DEFAULT: `None`
`counter`	The token counter to size turns and rendered blocks with. Defaults to :func:`~parrot.memory.compaction.tokens.get_default_counter`. TYPE: `Optional[TokenCounter]` DEFAULT: `None`
`calibration`	Multiplier applied to every calibrated size (the memory's EWMA of provider/estimate ratio). TYPE: `float` DEFAULT: `1.0`
`current_chatbot_id`	The agent doing the asking, for the foreign-turn rule (mirrors :func:`~parrot.memory.render.render_history`). TYPE: `Optional[str]` DEFAULT: `None`
`include_other_agents`	When `False`, foreign turns are excluded from consideration entirely (not even counted as dropped). TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`A`	class:`CompactionResult` with `views` ordered oldest → newest, TYPE: `CompactionResult`
`CompactionResult`	ready for :func:`~parrot.memory.render.render_history`.

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

def compact_history(
    history: ConversationHistory,
    budget: ContextBudget,
    *,
    policies: Optional[Dict[str, PrunePolicy]] = None,
    boundary_turn_id: Optional[str] = None,
    counter: Optional[TokenCounter] = None,
    calibration: float = 1.0,
    current_chatbot_id: Optional[str] = None,
    include_other_agents: bool = True,
) -> CompactionResult:
    """Walk a history newest → oldest and classify each turn RAW / PRUNED / DROPPED.

    Pure and synchronous (spec G1/G9): never mutates ``history``, its
    turns, or their invocations; performs no I/O; the same inputs always
    produce an equal :class:`CompactionResult`.

    Three tiers (spec G10), walked newest → oldest with a running
    calibrated token sum:

    1. **Verbatim (RAW)** while ``n_raw < budget.min_verbatim_turns`` or
       the cumulative size stays within both ``budget.verbatim_tokens``
       and the high-watermark budget.
    2. **Pruned** while the cumulative size stays within the high
       watermark (or the turn is the newest kept turn — the newest is
       never dropped).
    3. **Dropped** beyond that — and every older turn is dropped too
       (tiers are contiguous); sets ``stage2_needed``.

    A persisted ``boundary_turn_id`` forces every turn at or before it to
    render PRUNED regardless of budget (monotonic: turns that once
    rendered PRUNED never render RAW again for this history). Any
    invocation whose output exceeds ``budget.oversize_tool_tokens`` is
    pruned in every turn but the newest, even inside the verbatim tier.

    Args:
        history: The history to compact.
        budget: The retention configuration.
        policies: Per-tool prune-policy overrides, consulted before the
            module registry.
        boundary_turn_id: The previously persisted boundary, if any.
        counter: The token counter to size turns and rendered blocks
            with. Defaults to :func:`~parrot.memory.compaction.tokens.get_default_counter`.
        calibration: Multiplier applied to every calibrated size (the
            memory's EWMA of provider/estimate ratio).
        current_chatbot_id: The agent doing the asking, for the foreign-turn
            rule (mirrors :func:`~parrot.memory.render.render_history`).
        include_other_agents: When ``False``, foreign turns are excluded
            from consideration entirely (not even counted as dropped).

    Returns:
        A :class:`CompactionResult` with ``views`` ordered oldest → newest,
        ready for :func:`~parrot.memory.render.render_history`.
    """
    counter = counter or get_default_counter()
    available = budget.available
    watermark = int(budget.high_watermark * available)

    candidates = [
        t
        for t in history.turns
        if (t.assistant_response or "").strip() and (include_other_agents or not _is_foreign(t, current_chatbot_id))
    ]

    if len(candidates) > budget.max_turns:
        ceiling_dropped: Tuple[str, ...] = tuple(t.turn_id for t in candidates[: -budget.max_turns])
        kept = candidates[-budget.max_turns :]
    else:
        ceiling_dropped = ()
        kept = candidates

    boundary_index = next((i for i, t in enumerate(kept) if t.turn_id == boundary_turn_id), None)

    views_rev: List[TurnView] = []
    omissions: List[Omission] = []
    seen_content_ids: Set[str] = set()
    cum = 0
    n_raw = 0
    pruned_seen = False
    watermark_overflow = False
    extra_dropped_rev: List[str] = []

    last_index = len(kept) - 1
    for idx in range(last_index, -1, -1):
        turn = kept[idx]
        newest = idx == last_index

        tc = turn.token_count if not needs_recount(turn, counter) else count_turn(turn, counter)

        # An invocation counts as "oversize" for the render-time rule when
        # its current output still exceeds the threshold, OR when it was
        # already offloaded at write time (`inv.output` is then just the
        # preview — small, but the content is gone regardless of size).
        oversize = (
            ()
            if newest
            else tuple(
                i
                for i, inv in enumerate(turn.tool_invocations)
                if "output" in inv.omitted or counter.count(inv.output or "") > budget.oversize_tool_tokens
            )
        )

        forced = boundary_index is not None and idx <= boundary_index

        raw_suffix, raw_omissions = render_raw_view(
            turn, budget.tool_activity_limit, oversize=oversize, policies=policies, counter=counter
        )
        raw_size = math.ceil(calibration * (tc.user + tc.assistant + counter.count(raw_suffix)))

        pr_suffix, pr_omissions = prune_turn(turn, limit=budget.tool_activity_limit, policies=policies)
        pr_size = math.ceil(calibration * (tc.user + tc.assistant + counter.count(pr_suffix)))

        # A turn with an oversize invocation can never be classified RAW —
        # its "verbatim" rendering already collapses that invocation to the
        # same notice PRUNED would use (spec: oversize applies "even in the
        # verbatim tier"). Without this, `min_verbatim_turns` would force
        # such turns RAW purely on their (now-tiny, notice-only) rendered
        # size, defeating the pruned tier for every oversize-heavy history.
        has_oversize = len(oversize) > 0

        if (
            not forced
            and not pruned_seen
            and not has_oversize
            and (
                n_raw < budget.min_verbatim_turns
                or (cum + raw_size <= budget.verbatim_tokens and cum + raw_size <= watermark)
            )
        ):
            state: Optional[TurnState] = TurnState.RAW
        elif cum + pr_size <= watermark or n_raw == 0:
            state = TurnState.PRUNED
        else:
            state = None

        if state is None:
            watermark_overflow = True
            for j in range(idx, -1, -1):
                extra_dropped_rev.append(kept[j].turn_id)
            break

        if state is TurnState.RAW:
            views_rev.append(
                TurnView(
                    turn_id=turn.turn_id,
                    chatbot_id=turn.chatbot_id,
                    user_text=turn.user_message or "",
                    assistant_text=turn.assistant_response or "",
                    assistant_suffix=raw_suffix,
                    state=TurnState.RAW,
                    estimated_tokens=raw_size,
                )
            )
            for om in raw_omissions:
                if om.content_id not in seen_content_ids:
                    seen_content_ids.add(om.content_id)
                    omissions.append(om)
            cum += raw_size
            n_raw += 1
        else:
            views_rev.append(
                TurnView(
                    turn_id=turn.turn_id,
                    chatbot_id=turn.chatbot_id,
                    user_text=turn.user_message or "",
                    assistant_text=turn.assistant_response or "",
                    assistant_suffix=pr_suffix,
                    state=TurnState.PRUNED,
                    estimated_tokens=pr_size,
                )
            )
            for om in pr_omissions:
                if om.content_id not in seen_content_ids:
                    seen_content_ids.add(om.content_id)
                    omissions.append(om)
            cum += pr_size
            pruned_seen = True

    views = tuple(reversed(views_rev))
    dropped_turn_ids = ceiling_dropped + tuple(reversed(extra_dropped_rev))

    stage2_needed = watermark_overflow or cum > available

    pruned_views = [v for v in views if v.state is TurnState.PRUNED]
    new_boundary = pruned_views[-1].turn_id if pruned_views else boundary_turn_id

    return CompactionResult(
        views=views,
        omissions=tuple(omissions),
        history_estimate=cum,
        boundary_turn_id=new_boundary,
        stage2_needed=stage2_needed,
        dropped_turn_ids=dropped_turn_ids,
    )

render_history ¶

render_history(history: Union[Optional[ConversationHistory], Sequence['TurnView']], *, 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 (or turn views) into alternating messages.

PARAMETER	DESCRIPTION
`history`	A :class:`ConversationHistory` to render, or a `Sequence[TurnView]` (e.g. from :func:`parrot.memory.compaction.compact.compact_history`). `None` or an empty history/sequence renders to `[]`. For a view sequence, each view's `assistant_suffix` (already rendered tool activity / omission notices) is appended to its assistant message before the merge/alternation logic below; plain-history output is unaffected and byte-identical. TYPE: `Union[Optional[ConversationHistory], Sequence['TurnView']]`
`max_turns`	Keep only the most recent `N` turns. `None` keeps all. Values `<= 0` also render to `[]`. Ignored for a view sequence — compaction already applied its own ceiling. TYPE: `Optional[int]` DEFAULT: `None`
`current_chatbot_id`	The agent doing the asking. Turns/views whose `chatbot_id` differs are considered foreign; one 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/views 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. The label precedes both the assistant text and its suffix. 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]`	a turn/view whose assistant text is empty or whitespace-only contributes nothing at all — never an empty assistant message, and never rescued by a non-empty `assistant_suffix`;
`List[HistoryMessage]`	the input `history`/views are not modified.

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

def render_history(
    history: Union[Optional[ConversationHistory], Sequence["TurnView"]],
    *,
    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 (or turn views) into alternating messages.

    Args:
        history: A :class:`ConversationHistory` to render, or a
            ``Sequence[TurnView]`` (e.g. from
            :func:`parrot.memory.compaction.compact.compact_history`).
            ``None`` or an empty history/sequence renders to ``[]``. For a
            view sequence, each view's ``assistant_suffix`` (already
            rendered tool activity / omission notices) is appended to its
            assistant message before the merge/alternation logic below;
            plain-history output is unaffected and byte-identical.
        max_turns: Keep only the most recent ``N`` turns. ``None`` keeps all.
            Values ``<= 0`` also render to ``[]``. Ignored for a view
            sequence — compaction already applied its own ceiling.
        current_chatbot_id: The agent doing the asking. Turns/views whose
            ``chatbot_id`` differs are considered *foreign*; one 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/views 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. The
            label precedes both the assistant text and its suffix.
        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;
        * a turn/view whose assistant text is empty or whitespace-only
          contributes nothing at all — never an empty assistant message,
          and never rescued by a non-empty ``assistant_suffix``;
        * the input ``history``/views are not modified.
    """
    rendered: List[HistoryMessage] = []

    for turn_id, chatbot_id, user_text, assistant_text in _iter_rows(history, max_turns):
        if not assistant_text.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 row so alternation stays intact.
            continue

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

        _append(
            rendered,
            HistoryMessage(role="user", content=user_text, chatbot_id=chatbot_id, turn_id=turn_id),
        )
        _append(
            rendered,
            HistoryMessage(role="assistant", content=assistant_text, chatbot_id=chatbot_id, turn_id=turn_id),
        )

    return rendered

Memory¶

ConversationHistory dataclass ¶

add_turn ¶

get_recent_turns ¶

clear_turns ¶

to_dict ¶

from_dict classmethod ¶

ConversationMemory ¶

token_counter property ¶

omission_store property ¶

omission_key ¶

create_history abstractmethod async ¶

get_history abstractmethod async ¶

update_history abstractmethod async ¶

add_turn async ¶

report_usage async ¶

clear_history abstractmethod async ¶

list_sessions abstractmethod async ¶

delete_history abstractmethod async ¶

ConversationTurn dataclass ¶

to_dict ¶

from_dict classmethod ¶

from_ai_message classmethod ¶

AnswerMemory ¶

store_interaction async ¶

get async ¶

CompactionCommit dataclass ¶

CompactionResult dataclass ¶

ContextBudget dataclass ¶

available property ¶

Limit dataclass ¶

OmissionStore ¶

put abstractmethod async ¶

put_many async ¶

get abstractmethod async ¶

probe async ¶

probe_many async ¶

list_by_turn abstractmethod async ¶

clear abstractmethod async ¶

TokenCount dataclass ¶

to_dict ¶

from_dict classmethod ¶

TokenCounter ¶

count ¶

ToolInvocation dataclass ¶

to_dict ¶

from_dict classmethod ¶

ToolStatus ¶

TurnState ¶

TurnView dataclass ¶

EpisodicMemoryMixin ¶

flush_episodic_memory async ¶

EpisodicMemoryStore ¶

record_episode async ¶

record_tool_episode async ¶

record_crew_episode async ¶

recall_similar async ¶

get_failure_warnings async ¶

get_user_preferences async ¶

get_room_context async ¶

mark_consolidated async ¶

cleanup_expired async ¶

compact_namespace async ¶

export_episodes async ¶

create_pgvector async classmethod ¶

create_redis_vector async classmethod ¶

create_faiss async classmethod ¶

EpisodicMemoryToolkit ¶

search_episodic_memory async ¶

record_lesson async ¶

get_warnings async ¶

FileConversationMemory ¶

create_history async ¶

get_history async ¶

update_history async ¶

clear_history async ¶

list_sessions async ¶

delete_history async ¶

InMemoryConversation ¶

create_history async ¶

ConversationHistory `dataclass` ¶

from_dict `classmethod` ¶

token_counter `property` ¶

omission_store `property` ¶

create_history `abstractmethod` `async` ¶

get_history `abstractmethod` `async` ¶

update_history `abstractmethod` `async` ¶

add_turn `async` ¶

report_usage `async` ¶

clear_history `abstractmethod` `async` ¶

list_sessions `abstractmethod` `async` ¶

delete_history `abstractmethod` `async` ¶

ConversationTurn `dataclass` ¶

from_dict `classmethod` ¶

from_ai_message `classmethod` ¶

store_interaction `async` ¶

get `async` ¶

CompactionCommit `dataclass` ¶

CompactionResult `dataclass` ¶

ContextBudget `dataclass` ¶

available `property` ¶

Limit `dataclass` ¶

put `abstractmethod` `async` ¶

put_many `async` ¶

get `abstractmethod` `async` ¶

probe `async` ¶

probe_many `async` ¶

list_by_turn `abstractmethod` `async` ¶

clear `abstractmethod` `async` ¶

TokenCount `dataclass` ¶

from_dict `classmethod` ¶

ToolInvocation `dataclass` ¶

from_dict `classmethod` ¶

TurnView `dataclass` ¶

flush_episodic_memory `async` ¶

record_episode `async` ¶

record_tool_episode `async` ¶

record_crew_episode `async` ¶

recall_similar `async` ¶

get_failure_warnings `async` ¶

get_user_preferences `async` ¶

get_room_context `async` ¶

mark_consolidated `async` ¶

cleanup_expired `async` ¶

compact_namespace `async` ¶

export_episodes `async` ¶

create_pgvector `async` `classmethod` ¶

create_redis_vector `async` `classmethod` ¶

create_faiss `async` `classmethod` ¶

search_episodic_memory `async` ¶

record_lesson `async` ¶

get_warnings `async` ¶

create_history `async` ¶

get_history `async` ¶

update_history `async` ¶

clear_history `async` ¶

list_sessions `async` ¶

delete_history `async` ¶

create_history `async` ¶

get_history `async` ¶

update_history `async` ¶

clear_history `async` ¶

list_sessions `async` ¶

delete_history `async` ¶

create_history `async` ¶

get_history `async` ¶

update_history `async` ¶

read_metadata `async` ¶

merge_metadata `async` ¶

clear_history `async` ¶

list_sessions `async` ¶

delete_history `async` ¶

close `async` ¶

ping `async` ¶

get_raw_data `async` ¶

debug_conversation `async` ¶

list_sessions_by_chatbot `async` ¶

get_chatbot_stats `async` ¶

delete_all_chatbot_conversations `async` ¶

get_chatbot_users `async` ¶