Core API Reference

This page provides detailed documentation for the core GUM module.

Main Class

gum.gum

Attributes

Classes

gum(user_name: str, model: str, *observers: Observer, propose_prompt: str | None = None, similar_prompt: str | None = None, revise_prompt: str | None = None, audit_prompt: str | None = None, data_directory: str = '~/.cache/gum', db_name: str = 'gum.db', max_concurrent_updates: int = 4, verbosity: int = logging.INFO, audit_enabled: bool = False, api_base: str | None = None, api_key: str | None = None)

A class for managing general user models.

This class provides functionality for observing user behavior, generating and managing propositions about user behavior, and maintaining relationships between observations and propositions.

Parameters:

Name	Type	Description	Default
user_name	str	The name of the user being modeled.	required
*observers	Observer	Variable number of observer instances to track user behavior.	()
propose_prompt	str	Custom prompt for proposition generation.	None
similar_prompt	str	Custom prompt for similarity analysis.	None
revise_prompt	str	Custom prompt for proposition revision.	None
audit_prompt	str	Custom prompt for auditing.	None
data_directory	str	Directory for storing data. Defaults to "~/.cache/gum".	'~/.cache/gum'
db_name	str	Name of the database file. Defaults to "gum.db".	'gum.db'
max_concurrent_updates	int	Maximum number of concurrent updates. Defaults to 4.	4
verbosity	int	Logging verbosity level. Defaults to logging.INFO.	INFO
audit_enabled	bool	Whether to enable auditing. Defaults to False.	False

Source code in gum/gum.py

def __init__(
    self,
    user_name: str,
    model: str,
    *observers: Observer,
    propose_prompt: str | None = None,
    similar_prompt: str | None = None,
    revise_prompt: str | None = None,
    audit_prompt: str | None = None,
    data_directory: str = "~/.cache/gum",
    db_name: str = "gum.db",
    max_concurrent_updates: int = 4,
    verbosity: int = logging.INFO,
    audit_enabled: bool = False,
    api_base: str | None = None,
    api_key: str | None = None,
):
    # basic paths
    data_directory = os.path.expanduser(data_directory)
    os.makedirs(data_directory, exist_ok=True)

    # runtime
    self.user_name = user_name
    self.observers: list[Observer] = list(observers)
    self.model = model
    self.audit_enabled = audit_enabled

    # logging
    self.logger = logging.getLogger("gum")
    self.logger.setLevel(verbosity)
    if not self.logger.handlers:
        h = logging.StreamHandler()
        h.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s"))
        self.logger.addHandler(h)

    # prompts
    self.propose_prompt = propose_prompt or PROPOSE_PROMPT
    self.similar_prompt = similar_prompt or SIMILAR_PROMPT
    self.revise_prompt = revise_prompt or REVISE_PROMPT
    self.audit_prompt = audit_prompt or AUDIT_PROMPT

    self.client = AsyncOpenAI(
        base_url=api_base or os.getenv("GUM_LM_API_BASE"), 
        api_key=api_key or os.getenv("GUM_LM_API_KEY") or os.getenv("OPENAI_API_KEY") or "None"
    )

    self.engine = None
    self.Session = None
    self._db_name        = db_name
    self._data_directory = data_directory

    self._update_sem = asyncio.Semaphore(max_concurrent_updates)
    self._tasks: set[asyncio.Task] = set()
    self._loop_task: asyncio.Task | None = None
    self.update_handlers: list[Callable[[Observer, Update], None]] = []

Attributes

Session = None instance-attribute

audit_enabled = audit_enabled instance-attribute

audit_prompt = audit_prompt or AUDIT_PROMPT instance-attribute

client = AsyncOpenAI(base_url=api_base or os.getenv('GUM_LM_API_BASE'), api_key=api_key or os.getenv('GUM_LM_API_KEY') or os.getenv('OPENAI_API_KEY') or 'None') instance-attribute

engine = None instance-attribute

logger = logging.getLogger('gum') instance-attribute

model = model instance-attribute

observers: list[Observer] = list(observers) instance-attribute

propose_prompt = propose_prompt or PROPOSE_PROMPT instance-attribute

revise_prompt = revise_prompt or REVISE_PROMPT instance-attribute

similar_prompt = similar_prompt or SIMILAR_PROMPT instance-attribute

update_handlers: list[Callable[[Observer, Update], None]] = [] instance-attribute

user_name = user_name instance-attribute

Functions

__aenter__() async

Async context manager entry point.

Returns:

Name	Type	Description
gum		The instance of the gum class.

Source code in gum/gum.py

async def __aenter__(self):
    """Async context manager entry point.

    Returns:
        gum: The instance of the gum class.
    """
    await self.connect_db()
    self.start_update_loop()
    return self

__aexit__(exc_type, exc, tb) async

Async context manager exit point.

Parameters:

Name	Type	Description	Default
exc_type		The type of exception if any.	required
exc		The exception instance if any.	required
tb		The traceback if any.	required

Source code in gum/gum.py

async def __aexit__(self, exc_type, exc, tb):
    """Async context manager exit point.

    Args:
        exc_type: The type of exception if any.
        exc: The exception instance if any.
        tb: The traceback if any.
    """
    await self.stop_update_loop()

    # wait for any in-flight handlers
    if self._tasks:
        await asyncio.gather(*self._tasks, return_exceptions=True)

    # stop observers
    for obs in self.observers:
        await obs.stop()

add_observer(observer: Observer)

Add an observer to track user behavior.

Parameters:

Name	Type	Description	Default
observer	Observer	The observer to add.	required

Source code in gum/gum.py

def add_observer(self, observer: Observer):
    """Add an observer to track user behavior.

    Args:
        observer (Observer): The observer to add.
    """
    self.observers.append(observer)

connect_db() async

Initialize the database connection if not already connected.

Source code in gum/gum.py

async def connect_db(self):
    """Initialize the database connection if not already connected."""
    if self.engine is None:
        self.engine, self.Session = await init_db(
            self._db_name, self._data_directory
        )

query(user_query: str, *, limit: int = 3, mode: str = 'OR', start_time: datetime | None = None, end_time: datetime | None = None) -> list[tuple[Proposition, float]] async

Query the database for propositions matching the user query.

Parameters:

Name	Type	Description	Default
user_query	str	The query string to search for.	required
limit	int	Maximum number of results to return. Defaults to 3.	3
mode	str	Search mode ("OR" or "AND"). Defaults to "OR".	'OR'
start_time	datetime	Start time for filtering results. Defaults to None.	None
end_time	datetime	End time for filtering results. Defaults to None.	None

Returns:

Type	Description
list[tuple[Proposition, float]]	list[tuple[Proposition, float]]: List of tuples containing propositions and their relevance scores.

Source code in gum/gum.py

async def query(
    self,
    user_query: str,
    *,
    limit: int = 3,
    mode: str = "OR",
    start_time: datetime | None = None,
    end_time: datetime | None = None,
) -> list[tuple[Proposition, float]]:
    """Query the database for propositions matching the user query.

    Args:
        user_query (str): The query string to search for.
        limit (int, optional): Maximum number of results to return. Defaults to 3.
        mode (str, optional): Search mode ("OR" or "AND"). Defaults to "OR".
        start_time (datetime, optional): Start time for filtering results. Defaults to None.
        end_time (datetime, optional): End time for filtering results. Defaults to None.

    Returns:
        list[tuple[Proposition, float]]: List of tuples containing propositions and their relevance scores.
    """
    async with self._session() as session:
        return await search_propositions_bm25(
            session,
            user_query,
            limit=limit,
            mode=mode,
            start_time=start_time,
            end_time=end_time,
        )

register_update_handler(fn: Callable[[Observer, Update], None])

Register a custom update handler function.

Parameters:

Name	Type	Description	Default
fn	Callable[[Observer, Update], None]	The handler function to register.	required

Source code in gum/gum.py

def register_update_handler(self, fn: Callable[[Observer, Update], None]):
    """Register a custom update handler function.

    Args:
        fn (Callable[[Observer, Update], None]): The handler function to register.
    """
    self.update_handlers.append(fn)

remove_observer(observer: Observer)

Remove an observer from tracking.

Parameters:

Name	Type	Description	Default
observer	Observer	The observer to remove.	required

Source code in gum/gum.py

def remove_observer(self, observer: Observer):
    """Remove an observer from tracking.

    Args:
        observer (Observer): The observer to remove.
    """
    if observer in self.observers:
        self.observers.remove(observer)

start_update_loop()

Start the asynchronous update loop for processing observer updates.

Source code in gum/gum.py

def start_update_loop(self):
    """Start the asynchronous update loop for processing observer updates."""
    if self._loop_task is None:
        self._loop_task = asyncio.create_task(self._update_loop())

stop_update_loop() async

Stop the asynchronous update loop and clean up resources.

Source code in gum/gum.py

async def stop_update_loop(self):
    """Stop the asynchronous update loop and clean up resources."""
    if self._loop_task:
        self._loop_task.cancel()
        try:
            await self._loop_task
        except asyncio.CancelledError:
            pass
        self._loop_task = None

Functions