Skip to main content

Table of Contents

Module spoon_ai.chat

ShortTermMemoryConfig Objects​

class ShortTermMemoryConfig(BaseModel)

Configuration for short-term memory management.

enabled​

Enable automatic short-term memory management.

max_tokens​

Maximum token count before triggering trimming/summarization.

strategy​

Strategy to use when exceeding max_tokens: 'summarize' or 'trim'.

messages_to_keep​

Number of recent messages to keep when summarizing.

trim_strategy​

Trimming strategy when using 'trim' mode.

keep_system_messages​

Always keep system messages during trimming.

auto_checkpoint​

Automatically save checkpoints before trimming/summarization.

checkpoint_thread_id​

Thread ID for checkpoint management.

summary_model​

Model to use for summarization (defaults to ChatBot's model).

ChatBot Objects​

class ChatBot()

__init__​

def __init__(use_llm_manager: bool = True,
             model_name: str = None,
             llm_provider: str = None,
             api_key: str = None,
             base_url: str = None,
             enable_short_term_memory: bool = True,
             short_term_memory_config: Optional[Union[Dict[
                 str, Any], ShortTermMemoryConfig]] = None,
             token_counter: Optional[MessageTokenCounter] = None,
             enable_long_term_memory: bool = False,
             mem0_config: Optional[Dict[str, Any]] = None,
             callbacks: Optional[List[BaseCallbackHandler]] = None,
             **kwargs)

Initialize ChatBot with hierarchical configuration priority system.

Configuration Priority System:

  1. Full manual override (highest priority) - all params provided
  2. Partial override with config fallback - llm_provider provided, credentials pulled from environment (or config files if explicitly enabled)
  3. Full environment-based loading - only use_llm_manager=True, reads from environment variables

Arguments:

  • use_llm_manager - Enable LLM manager architecture (default: True)
  • model_name - Model name override
  • llm_provider - Provider name override
  • api_key - API key override
  • base_url - Base URL override
  • enable_short_term_memory - Enable short-term memory management (default: True)
  • short_term_memory_config - Configuration dict or ShortTermMemoryConfig instance
  • token_counter - Optional custom token counter instance
  • enable_long_term_memory - Enable Mem0-backed long-term memory retrieval/storage
  • mem0_config - Configuration dict for Mem0 (api_key, user_id/agent_id, collection, etc.)
  • callbacks - Optional list of callback handlers for monitoring
  • **kwargs - Additional parameters

update_mem0_config​

def update_mem0_config(config: Optional[Dict[str, Any]] = None,
                       enable: Optional[bool] = None) -> None

Update Mem0 configuration and re-initialize the client if needed.

ask​

async def ask(messages: List[Union[dict, Message]],
              system_msg: Optional[str] = None,
              output_queue: Optional[asyncio.Queue] = None) -> str

Ask method using the LLM manager architecture.

Automatically applies short-term memory strategy if enabled.

ask_tool​

async def ask_tool(messages: List[Union[dict, Message]],
                   system_msg: Optional[str] = None,
                   tools: Optional[List[dict]] = None,
                   tool_choice: Optional[str] = None,
                   output_queue: Optional[asyncio.Queue] = None,
                   **kwargs) -> LLMResponse

Ask tool method using the LLM manager architecture.

Automatically applies short-term memory strategy if enabled.

trim_messages​

async def trim_messages(messages: List[Message],
                        max_tokens: int,
                        strategy: TrimStrategy = TrimStrategy.FROM_END,
                        keep_system: bool = True,
                        model: Optional[str] = None) -> List[Message]

Trim messages to stay within the token budget.

Arguments:

  • messages - List of messages to trim
  • max_tokens - Maximum token count to retain
  • strategy - Trimming strategy (from_start or from_end)
  • keep_system - Whether to always keep the leading system message
  • model - Model name for token counting

Returns:

  • List[Message] - Trimmed messages list

remove_message​

def remove_message(message_id: str, **kwargs: Any) -> "RemoveMessage"

Construct a removal instruction for the message with the given ID.

remove_all_messages​

def remove_all_messages() -> "RemoveMessage"

Construct a removal instruction that clears the entire history.

summarize_messages​

async def summarize_messages(
    messages: List[Message],
    max_tokens_before_summary: int,
    messages_to_keep: int = 5,
    summary_model: Optional[str] = None,
    existing_summary: str = ""
) -> Tuple[List[Message], List[RemoveMessage], Optional[str]]

Summarize earlier messages and emit removal directives.

Returns a tuple (messages_for_llm, removals, summary_text) where messages_for_llm are the messages that should be sent to the language model for the next turn, removals contains RemoveMessage directives that should be applied to the stored history, and summary_text is the newly generated summary (if any).

Arguments:

  • messages - List of messages to process
  • max_tokens_before_summary - Token threshold for triggering summary
  • messages_to_keep - Number of recent messages to keep uncompressed
  • summary_model - Model to use for summarization
  • existing_summary - Previously stored summary text

latest_summary​

@property
def latest_summary() -> Optional[str]

Return the most recent summary generated by short-term memory.

latest_removals​

@property
def latest_removals() -> List[RemoveMessage]

Return the most recent removal directives emitted by summarization.

save_checkpoint​

def save_checkpoint(thread_id: str,
                    messages: List[Message],
                    metadata: Optional[dict] = None) -> str

Save current message state to checkpoint.

Arguments:

  • thread_id - Thread identifier
  • messages - Messages to save
  • metadata - Optional metadata to store

Returns:

  • str - Checkpoint ID

restore_checkpoint​

def restore_checkpoint(
        thread_id: str,
        checkpoint_id: Optional[str] = None) -> Optional[List[Message]]

Restore messages from checkpoint.

Arguments:

  • thread_id - Thread identifier
  • checkpoint_id - Optional specific checkpoint ID

Returns:

  • Optional[List[Message]] - Restored messages, or None if checkpoint not found

list_checkpoints​

def list_checkpoints(thread_id: str) -> List[dict]

List all checkpoints for a thread.

Arguments:

  • thread_id - Thread identifier

Returns:

  • List[dict] - List of checkpoint metadata

clear_checkpoints​

def clear_checkpoints(thread_id: str) -> None

Clear all checkpoints for a thread.

Arguments:

  • thread_id - Thread identifier

astream​

async def astream(messages: List[Union[dict, Message]],
                  system_msg: Optional[str] = None,
                  callbacks: Optional[List[BaseCallbackHandler]] = None,
                  **kwargs: Any) -> AsyncIterator[LLMResponseChunk]

Stream LLM responses chunk by chunk.

astream_events​

async def astream_events(messages: List[Union[dict, Message]],
                         system_msg: Optional[str] = None,
                         callbacks: Optional[List[BaseCallbackHandler]] = None,
                         **kwargs) -> AsyncIterator[dict]

Stream structured events during LLM execution.

This method yields detailed events tracking the execution flow, useful for monitoring and debugging.

Arguments:

  • messages - List of messages or dicts
  • system_msg - Optional system message
  • callbacks - Optional callback handlers
  • **kwargs - Additional provider parameters

Yields:

Event dictionaries with structure: {

  • "event" - event_type,
  • "run_id" - str,
  • "timestamp" - ISO datetime string,
  • "data" - {event-specific data} }

astream_log​

async def astream_log(messages: List[Union[dict, Message]],
                      system_msg: Optional[str] = None,
                      callbacks: Optional[List[BaseCallbackHandler]] = None,
                      *,
                      diff: bool = True,
                      **kwargs: Any) -> AsyncIterator[RunLogPatch]

Stream run log patches describing ChatBot execution.