semlib

Session

Bases: Sort, Filter, Extrema, Find, Apply, Reduce

A session provides a context for performing Semlib operations.

Sessions provide defaults (e.g., default model), manage caching, implement concurrency control, and track costs.

All of a Session's methods have analogous standalone functions (e.g., the Session.map method has an analogous map function). It is recommended to use a Session for any non-trivial use of the semlib library.

Source code in src/semlib/session.py

class Session(Sort, Filter, Extrema, Find, Apply, Reduce):
    """A session provides a context for performing Semlib operations.

    Sessions provide defaults (e.g., default model), manage caching, implement concurrency control, and track costs.

    All of a Session's methods have analogous standalone functions (e.g., the [Session.map][semlib.map.Map.map] method
    has an analogous [map][semlib.map.map] function). It is recommended to use a Session for any non-trivial use of the
    semlib library.
    """

model property

model: str

Get the current model being used for completions.

Returns:

Type	Description
str	The model name as a string.

__init__

__init__(
    *,
    model: str | None = None,
    max_concurrency: int | None = None,
    cache: QueryCache | None = None,
)

Initialize.

Parameters:

Name	Type	Description	Default
model	str | None	The language model to use for completions. If not specified, uses the value from the SEMLIB_DEFAULT_MODEL environment variable, or falls back to the default model (currently "openai/gpt-4o"). This is used as the model argument for litellm unless overridden in individual method calls.	None
max_concurrency	int | None	Maximum number of concurrent API requests. If not specified, uses the value from the SEMLIB_MAX_CONCURRENCY environment variable, or defaults to 10 for most models, or 1 for Ollama models.	None
cache	QueryCache | None	If provided, this is used to cache LLM responses to avoid redundant API calls.	None

Raises:

Type	Description
ValueError	If max_concurrency is provided but is not a positive integer.

Source code in src/semlib/_internal/base.py

def __init__(
    self, *, model: str | None = None, max_concurrency: int | None = None, cache: QueryCache | None = None
):
    """Initialize.

    Args:
        model: The language model to use for completions. If not specified, uses the value from the
            `SEMLIB_DEFAULT_MODEL` environment variable, or falls back to the default model (currently
            `"openai/gpt-4o"`). This is used as the `model` argument for
            [litellm](https://docs.litellm.ai/docs/providers) unless overridden in individual method calls.
        max_concurrency: Maximum number of concurrent API requests. If not specified, uses the value from the
            `SEMLIB_MAX_CONCURRENCY` environment variable, or defaults to 10 for most models, or 1 for Ollama
            models.
        cache: If provided, this is used to cache LLM responses to avoid redundant API calls.

    Raises:
        ValueError: If `max_concurrency` is provided but is not a positive integer.
    """
    self._model = model or os.getenv("SEMLIB_DEFAULT_MODEL") or DEFAULT_MODEL
    if max_concurrency is None and (env_max_concurrency := os.getenv("SEMLIB_MAX_CONCURRENCY")) is not None:
        try:
            max_concurrency = int(env_max_concurrency)
        except ValueError:
            msg = "SEMLIB_MAX_CONCURRENCY must be an integer"
            raise ValueError(msg) from None
    self._max_concurrency = parse_max_concurrency(max_concurrency, self._model)
    self._sem = asyncio.Semaphore(self._max_concurrency)
    self._pending_requests: set[bytes] = set()
    self._cond = asyncio.Condition()  # for pending requests deduplication
    self._cache = cache

    self._total_cost: float = 0.0

apply async

apply[T, U: BaseModel](
    item: T,
    /,
    template: str | Callable[[T], str],
    *,
    return_type: type[U],
    model: str | None = None,
) -> U

apply[T, U](
    item: T,
    /,
    template: str | Callable[[T], str],
    *,
    return_type: Bare[U],
    model: str | None = None,
) -> U

apply[T](
    item: T,
    /,
    template: str | Callable[[T], str],
    *,
    return_type: None = None,
    model: str | None = None,
) -> str

Apply a language model prompt to a single item.

This method formats a prompt template with the given item, sends it to the language model, and returns the response. The response can be returned as a raw string, parsed into a Pydantic model, or extracted as a bare value using the Bare marker.

This method is a simple wrapper around prompt.

Parameters:

Name	Type	Description	Default
item	T	The item to apply the template to.	required
template	str | Callable[[T], str]	A template to format with the item. This can be either a string template with a single positional placeholder, or a callable that takes the item and returns a formatted string.	required
return_type	type[U] | Bare[V] | None	If not specified, the response is returned as a raw string. If a Pydantic model class is provided, the response is parsed into an instance of that model. If a Bare instance is provided, a single value of the specified type is extracted from the response.	None
model	str | None	If specified, overrides the default model for this call.	None

Returns:

Type	Description
U | V | str	The language model's response in the format specified by return_type.

Raises:

Type	Description
ValidationError	If return_type is a Pydantic model or a Bare type and the response cannot be parsed into the specified type.

Examples:

Basic usage:

>>> await session.apply(
...     [1, 2, 3, 4, 5],
...     template="What is the sum of these numbers: {}?",
...     return_type=Bare(int),
... )
15

Source code in src/semlib/apply.py

async def apply[T, U: BaseModel, V](
    self,
    item: T,
    /,
    template: str | Callable[[T], str],
    *,
    return_type: type[U] | Bare[V] | None = None,
    model: str | None = None,
) -> U | V | str:
    """Apply a language model prompt to a single item.

    This method formats a prompt template with the given item, sends it to the language model, and returns the
    response. The response can be returned as a raw string, parsed into a [Pydantic](https://pydantic.dev/) model,
    or extracted as a bare value using the [Bare][semlib.bare.Bare] marker.

    This method is a simple wrapper around [prompt][semlib._internal.base.Base.prompt].

    Args:
        item: The item to apply the `template` to.
        template: A template to format with the item. This can be either a string template with a single positional
            placeholder, or a callable that takes the item and returns a formatted string.
        return_type: If not specified, the response is returned as a raw string. If a Pydantic model class is
            provided, the response is parsed into an instance of that model. If a [Bare][semlib.bare.Bare] instance
            is provided, a single value of the specified type is extracted from the response.
        model: If specified, overrides the default model for this call.

    Returns:
        The language model's response in the format specified by return_type.

    Raises:
        ValidationError: If `return_type` is a Pydantic model or a Bare type and the response cannot be
            parsed into the specified type.

    Examples:
        Basic usage:
        >>> await session.apply(
        ...     [1, 2, 3, 4, 5],
        ...     template="What is the sum of these numbers: {}?",
        ...     return_type=Bare(int),
        ... )
        15
    """
    formatter = template.format if isinstance(template, str) else template

    model = model if model is not None else self._model

    return await self.prompt(formatter(item), return_type=return_type, model=model)

clear_cache

clear_cache() -> None

Clear the internal cache of LLM responses, if caching is enabled.

Source code in src/semlib/_internal/base.py

def clear_cache(self) -> None:
    """Clear the internal cache of LLM responses, if caching is enabled."""
    if self._cache is not None:
        self._cache.clear()

compare async

compare[T](
    a: T,
    b: T,
    /,
    *,
    by: str | None = None,
    to_str: Callable[[T], str] | None = None,
    template: str | Callable[[T, T], str] | None = None,
    task: Task | str | None = None,
    model: str | None = None,
) -> Order

Compare two items.

This method uses a language model to compare two items and determine the relative ordering of the two items. The comparison can be customized by specifying either a criteria to compare by, or a custom prompt template. The comparison task can be framed in a number of ways (choosing the greater item, lesser item, or the ordering).

Parameters:

Name	Type	Description	Default
a	T	The first item to compare.	required
b	T	The second item to compare.	required
by	str | None	A criteria specifying what aspect to compare by. If this is provided, template cannot be provided.	None
to_str	Callable[[T], str] | None	If specified, used to convert items to string representation. Otherewise, uses str() on each item. If this is provided, a callable template cannot be provided.	None
template	str | Callable[[T, T], str] | None	A custom prompt template for the comparison. Must be either a string template with two positional placeholders, or a callable that takes two items and returns a formatted string. If this is provided, by cannot be provided.	None
task	Task | str | None	The type of comparison task that is being performed in template. This allows for writing the template in the most convenient way possible (e.g., in some scenarios, it's easier to specify a criteria for which item is lesser, and in others, it's easier to specify a criteria for which item is greater). If this is provided, a custom template must also be provided. Defaults to Task.CHOOSE_GREATER if not specified.	None
model	str | None	If specified, overrides the default model for this call.	None

Returns:

Type	Description
Order	The ordering of the two items.

Raises:

Type	Description
ValidationError	If parsing the LLM response fails.

Examples:

Basic comparison:

>>> await session.compare("twelve", "seventy two")
<Order.LESS: 'less'>

Custom criteria:

>>> await session.compare("California condor", "Bald eagle", by="wingspan")
<Order.GREATER: 'greater'>

Custom template and task:

>>> await session.compare(
...     "proton",
...     "electron",
...     template="Which is smaller, (A) {} or (B) {}?",
...     task=Task.CHOOSE_LESSER,
... )
<Order.GREATER: 'greater'>

Source code in src/semlib/compare.py

async def compare[T](
    self,
    a: T,
    b: T,
    /,
    *,
    by: str | None = None,
    to_str: Callable[[T], str] | None = None,
    template: str | Callable[[T, T], str] | None = None,
    task: Task | str | None = None,
    model: str | None = None,
) -> Order:
    """Compare two items.

    This method uses a language model to compare two items and determine the relative ordering of the two items.
    The comparison can be customized by specifying either a criteria to compare by, or a custom prompt template. The
    comparison task can be framed in a number of ways (choosing the greater item, lesser item, or the ordering).

    Args:
        a: The first item to compare.
        b: The second item to compare.
        by: A criteria specifying what aspect to compare by. If this is provided, `template` cannot be
            provided.
        to_str: If specified, used to convert items to string representation. Otherewise, uses `str()` on each item.
            If this is provided, a callable template cannot be provided.
        template: A custom prompt template for the comparison. Must be either a string template with two positional
            placeholders, or a callable that takes two items and returns a formatted string. If this is provided,
            `by` cannot be provided.
        task: The type of comparison task that is being performed in `template`. This allows for writing the
            template in the most convenient way possible (e.g., in some scenarios, it's easier to specify a criteria
            for which item is lesser, and in others, it's easier to specify a criteria for which item is greater).
            If this is provided, a custom `template` must also be provided.  Defaults to
            [Task.CHOOSE_GREATER][semlib.compare.Task.CHOOSE_GREATER] if not specified.
        model: If specified, overrides the default model for this call.

    Returns:
        The ordering of the two items.

    Raises:
        ValidationError: If parsing the LLM response fails.

    Examples:
        Basic comparison:
        >>> await session.compare("twelve", "seventy two")
        <Order.LESS: 'less'>

        Custom criteria:
        >>> await session.compare("California condor", "Bald eagle", by="wingspan")
        <Order.GREATER: 'greater'>

        Custom template and task:
        >>> await session.compare(
        ...     "proton",
        ...     "electron",
        ...     template="Which is smaller, (A) {} or (B) {}?",
        ...     task=Task.CHOOSE_LESSER,
        ... )
        <Order.GREATER: 'greater'>
    """
    if task is not None and task not in {Task.CHOOSE_GREATER, Task.CHOOSE_GREATER_OR_ABSTAIN} and template is None:
        msg = "if 'task' is not CHOOSE_GREATER or CHOOSE_GREATER_OR_ABSTAIN, 'template' must also be provided"
        raise ValueError(msg)
    if template is not None:
        if callable(template) and to_str is not None:
            msg = "cannot provide 'to_str' when a template function is provided"
            raise ValueError(msg)
        if by is not None:
            msg = "cannot provide 'by' when a custom template is provided"
            raise ValueError(msg)

    to_str = to_str if to_str is not None else str
    if task is None:
        task = Task.CHOOSE_GREATER
    elif isinstance(task, str):
        task = Task(task)
    model = model if model is not None else self._model

    if isinstance(template, str):
        prompt = template.format(to_str(a), to_str(b))
    elif template is not None:
        # callable
        prompt = template(a, b)
    elif by is None:
        prompt = _DEFAULT_TEMPLATE.format(a=to_str(a), b=to_str(b))
    else:
        prompt = _DEFAULT_TEMPLATE_BY.format(criteria=by, a=to_str(a), b=to_str(b))

    response = await self.prompt(
        prompt,
        model=model,
        return_type=_RETURN_TYPE_BY_TASK[task],
    )

    match task:
        case Task.COMPARE:
            strict_compare_result = cast(_StrictCompareResult, response)
            return strict_compare_result.order.to_order()
        case Task.COMPARE_OR_ABSTAIN:
            compare_result = cast(_CompareResult, response)
            return compare_result.order
        case Task.CHOOSE_GREATER:
            strict_choose_result = cast(_StrictChooseResult, response)
            match strict_choose_result.choice:
                case _StrictChoice.A:
                    return Order.GREATER
                case _StrictChoice.B:
                    return Order.LESS
        case Task.CHOOSE_GREATER_OR_ABSTAIN:
            choose_result = cast(_ChooseResult, response)
            match choose_result.choice:
                case _Choice.A:
                    return Order.GREATER
                case _Choice.B:
                    return Order.LESS
                case _Choice.NEITHER:
                    return Order.NEITHER
        case Task.CHOOSE_LESSER:
            strict_choose_result = cast(_StrictChooseResult, response)
            match strict_choose_result.choice:
                case _StrictChoice.A:
                    return Order.LESS
                case _StrictChoice.B:
                    return Order.GREATER
        case Task.CHOOSE_LESSER_OR_ABSTAIN:
            choose_result = cast(_ChooseResult, response)
            match choose_result.choice:
                case _Choice.A:
                    return Order.LESS
                case _Choice.B:
                    return Order.GREATER
                case _Choice.NEITHER:
                    return Order.NEITHER

filter async

filter[T](
    iterable: Iterable[T],
    /,
    *,
    by: str | None = None,
    to_str: Callable[[T], str] | None = None,
    template: str | Callable[[T], str] | None = None,
    negate: bool = False,
    model: str | None = None,
) -> list[T]

Filter an iterable based on a criteria.

This method is analogous to Python's built-in filter function.

Parameters:

Name	Type	Description	Default
iterable	Iterable[T]	The collection of items to filter.	required
by	str | None	A criteria specifying a predicate to filter by. If this is provided, template cannot be provided.	None
to_str	Callable[[T], str] | None	If specified, used to convert items to string representation. Otherewise, uses str() on each item. If this is provided, a callable template cannot be provided.	None
template	str | Callable[[T], str] | None	A custom prompt template for predicates. Must be either a string template with a single positional placeholder, or a callable that takes an item and returns a formatted string. If this is provided, by cannot be provided.	None
negate	bool	If True, keep items that do not match the criteria. If False, keep items that match the criteria.	False
model	str | None	If specified, overrides the default model for this call.	None

Returns:

Type	Description
list[T]	A new list containing items from the iterable if they match the criteria.

Raises:

Type	Description
ValidationError	If parsing any LLM response fails.

Examples:

Basic filter:

>>> await session.filter(["Tom Hanks", "Tom Cruise", "Tom Brady"], by="actor?")
['Tom Hanks', 'Tom Cruise']

Custom template:

>>> await session.filter(
...     [(123, 321), (384, 483), (134, 431)],
...     template=lambda pair: f"Is {pair[0]} backwards {pair[1]}?",
...     negate=True,
... )
[(384, 483)]

Source code in src/semlib/filter.py

async def filter[T](
    self,
    iterable: Iterable[T],
    /,
    *,
    by: str | None = None,
    to_str: Callable[[T], str] | None = None,
    template: str | Callable[[T], str] | None = None,
    negate: bool = False,
    model: str | None = None,
) -> list[T]:
    """Filter an iterable based on a criteria.

    This method is analogous to Python's built-in
    [`filter`](https://docs.python.org/3/library/functions.html#filter) function.

    Args:
        iterable: The collection of items to filter.
        by: A criteria specifying a predicate to filter by. If this is provided, `template` cannot be provided.
        to_str: If specified, used to convert items to string representation. Otherewise, uses `str()` on each item.
            If this is provided, a callable template cannot be provided.
        template: A custom prompt template for predicates. Must be either a string template with a single positional
            placeholder, or a callable that takes an item and returns a formatted string. If this is provided, `by`
            cannot be provided.
        negate: If `True`, keep items that do **not** match the criteria. If `False`, keep items that match the
            criteria.
        model: If specified, overrides the default model for this call.

    Returns:
        A new list containing items from the iterable if they match the criteria.

    Raises:
        ValidationError: If parsing any LLM response fails.

    Examples:
        Basic filter:
        >>> await session.filter(["Tom Hanks", "Tom Cruise", "Tom Brady"], by="actor?")
        ['Tom Hanks', 'Tom Cruise']

        Custom template:
        >>> await session.filter(
        ...     [(123, 321), (384, 483), (134, 431)],
        ...     template=lambda pair: f"Is {pair[0]} backwards {pair[1]}?",
        ...     negate=True,
        ... )
        [(384, 483)]
    """
    if template is None:
        if by is None:
            msg = "must specify either 'by' or 'template'"
            raise ValueError(msg)
    else:
        if callable(template) and to_str is not None:
            msg = "cannot provide 'to_str' when a template function is provided"
            raise ValueError(msg)
        if by is not None:
            msg = "cannot provide 'by' when a custom template is provided"
            raise ValueError(msg)

    to_str = to_str if to_str is not None else str

    if template is None:

        def map_template(item: T, /) -> str:
            return _DEFAULT_TEMPLATE.format(by=by or "", item=to_str(item))
    elif isinstance(template, str):

        def map_template(item: T, /) -> str:
            return template.format(to_str(item))
    else:
        # callable
        map_template = template

    decisions = await self.map(iterable, map_template, return_type=_Decision, model=model)
    return [
        item
        for item, decision in zip(iterable, decisions, strict=False)
        if ((not decision.decision) if negate else decision.decision)
    ]

find async

find[T](
    iterable: Iterable[T],
    /,
    *,
    by: str | None = None,
    to_str: Callable[[T], str] | None = None,
    template: str | Callable[[T], str] | None = None,
    negate: bool = False,
    model: str | None = None,
) -> T | None

Find an item in an iterable based on a criteria.

This method searches through the provided iterable and returns some item (not necessarily the first) that matches the specified criteria.

Parameters:

Name	Type	Description	Default
iterable	Iterable[T]	The collection of items to search.	required
by	str | None	A criteria specifying a predicate to search by. If this is provided, template cannot be provided.	None
to_str	Callable[[T], str] | None	If specified, used to convert items to string representation. Otherewise, uses str() on each item. If this is provided, a callable template cannot be provided.	None
template	str | Callable[[T], str] | None	A custom prompt template for predicates. Must be either a string template with a single positional placeholder, or a callable that takes an item and returns a formatted string. If this is provided, by cannot be provided.	None
negate	bool	If True, find an item that does not match the criteria. If False, find an item that does match the criteria.	False
model	str | None	If specified, overrides the default model for this call.	None

Returns:

Type	Description
T | None	An item from the iterable if it matches the criteria, or None if no such item is found.

Raises:

Type	Description
ValidationError	If parsing any LLM response fails.

Examples:

Basic find:

>>> await session.find(["Tom Hanks", "Tom Cruise", "Tom Brady"], by="actor?")
'Tom Cruise'  # nondeterministic, could also return "Tom Hanks"

Custom template:

>>> await session.find(
...     [(123, 321), (384, 483), (134, 431)],
...     template=lambda pair: f"Is {pair[0]} backwards {pair[1]}?",
...     negate=True,
... )
(384, 483)

Source code in src/semlib/find.py

async def find[T](
    self,
    iterable: Iterable[T],
    /,
    *,
    by: str | None = None,
    to_str: Callable[[T], str] | None = None,
    template: str | Callable[[T], str] | None = None,
    negate: bool = False,
    model: str | None = None,
) -> T | None:
    """Find an item in an iterable based on a criteria.

    This method searches through the provided iterable and returns some item (not necessarily the first) that
    matches the specified criteria.

    Args:
        iterable: The collection of items to search.
        by: A criteria specifying a predicate to search by. If this is provided, `template` cannot be provided.
        to_str: If specified, used to convert items to string representation. Otherewise, uses `str()` on each item.
            If this is provided, a callable template cannot be provided.
        template: A custom prompt template for predicates. Must be either a string template with a single positional
            placeholder, or a callable that takes an item and returns a formatted string. If this is provided, `by`
            cannot be provided.
        negate: If `True`, find an item that does **not** match the criteria. If `False`, find an item that does
            match the criteria.
        model: If specified, overrides the default model for this call.

    Returns:
        An item from the iterable if it matches the criteria, or `None` if no such item is found.

    Raises:
        ValidationError: If parsing any LLM response fails.

    Examples:
        Basic find:
        >>> await session.find(["Tom Hanks", "Tom Cruise", "Tom Brady"], by="actor?")
        'Tom Cruise'  # nondeterministic, could also return "Tom Hanks"

        Custom template:
        >>> await session.find(
        ...     [(123, 321), (384, 483), (134, 431)],
        ...     template=lambda pair: f"Is {pair[0]} backwards {pair[1]}?",
        ...     negate=True,
        ... )
        (384, 483)
    """
    if template is None:
        if by is None:
            msg = "must specify either 'by' or 'template'"
            raise ValueError(msg)
    else:
        if callable(template) and to_str is not None:
            msg = "cannot provide 'to_str' when a template function is provided"
            raise ValueError(msg)
        if by is not None:
            msg = "cannot provide 'by' when a custom template is provided"
            raise ValueError(msg)

    to_str = to_str if to_str is not None else str

    if template is None:

        def map_template(item: T, /) -> str:
            return _DEFAULT_TEMPLATE.format(by=by or "", item=to_str(item))
    elif isinstance(template, str):

        def map_template(item: T, /) -> str:
            return template.format(to_str(item))
    else:
        # callable
        map_template = template

    model = model if model is not None else self._model

    async def fn(item: T) -> tuple[T, bool]:
        decision = await self.prompt(
            map_template(item),
            return_type=_Decision,
            model=model,
        )
        if negate:
            decision.decision = not decision.decision
        return item, decision.decision

    tasks: list[asyncio.Task[tuple[T, bool]]] = [asyncio.create_task(fn(item)) for item in iterable]
    try:
        for next_finished in asyncio.as_completed(tasks):
            item, decision = await next_finished
            if decision:
                return item
        return None
    finally:
        for task in tasks:
            if not task.done():
                task.cancel()
        await asyncio.wait(tasks)

map async

map[T, U: BaseModel](
    iterable: Iterable[T],
    /,
    template: str | Callable[[T], str],
    *,
    return_type: type[U],
    model: str | None = None,
) -> list[U]

map[T, U](
    iterable: Iterable[T],
    /,
    template: str | Callable[[T], str],
    *,
    return_type: Bare[U],
    model: str | None = None,
) -> list[U]

map[T](
    iterable: Iterable[T],
    /,
    template: str | Callable[[T], str],
    *,
    return_type: None = None,
    model: str | None = None,
) -> list[str]

Map a prompt template over an iterable and get responses from the language model.

This method applies a prompt template to each item in the provided iterable, sends the resulting prompts to the language model, and collects the responses. The responses can be returned as raw strings, parsed into Pydantic models, or extracted as bare values using the Bare marker.

This method is analogous to Python's built-in map function.

Parameters:

Name	Type	Description	Default
iterable	Iterable[T]	The collection of items to map over.	required
template	str | Callable[[T], str]	A prompt template to apply to each item. This can be either a string template with a single positional placeholder, or a callable that takes an item and returns a formatted string.	required
return_type	type[U] | Bare[V] | None	If not specified, the responses are returned as raw strings. If a Pydantic model class is provided, the responses are parsed into instances of that model. If a Bare instance is provided, single values of the specified type are extracted from the responses.	None
model	str | None	If specified, overrides the default model for this call.	None

Returns:

Type	Description
list[U] | list[V] | list[str]	A list of responses from the language model in the format specified by return_type.

Raises:

Type	Description
ValidationError	If return_type is a Pydantic model or a Bare type and the response cannot be parsed into the specified type.

Examples:

Basic map:

>>> await session.map(
...     ["apple", "banana", "kiwi"],
...     template="What color is {}? Reply in a single word.",
... )
['Red.', 'Yellow.', 'Green.']

Map with structured return type:

>>> class Person(pydantic.BaseModel):
...     name: str
...     age: int
>>> await session.map(
...     ["Barack Obama", "Angela Merkel"],
...     template="Who is {}?",
...     return_type=Person,
... )
[Person(name='Barack Obama', age=62), Person(name='Angela Merkel', age=69)]

Map with bare return type:

>>> await session.map(
...     [42, 1337, 2025],
...     template="What are the unique prime factors of {}?",
...     return_type=Bare(list[int]),
... )
[[2, 3, 7], [7, 191], [3, 5]]

Source code in src/semlib/map.py

async def map[T, U: BaseModel, V](
    self,
    iterable: Iterable[T],
    /,
    template: str | Callable[[T], str],
    *,
    return_type: type[U] | Bare[V] | None = None,
    model: str | None = None,
) -> list[U] | list[V] | list[str]:
    """Map a prompt template over an iterable and get responses from the language model.

    This method applies a prompt template to each item in the provided iterable, sends the resulting prompts to the
    language model, and collects the responses. The responses can be returned as raw strings, parsed into
    [Pydantic](https://pydantic.dev/) models, or extracted as bare values using the [Bare][semlib.bare.Bare] marker.

    This method is analogous to Python's built-in [`map`](https://docs.python.org/3/library/functions.html#map)
    function.

    Args:
        iterable: The collection of items to map over.
        template: A prompt template to apply to each item. This can be either a string template with a single
            positional placeholder, or a callable that takes an item and returns a formatted string.
        return_type: If not specified, the responses are returned as raw strings. If a Pydantic model class is provided,
            the responses are parsed into instances of that model. If a [Bare][semlib.bare.Bare] instance is
            provided, single values of the specified type are extracted from the responses.
        model: If specified, overrides the default model for this call.

    Returns:
        A list of responses from the language model in the format specified by return_type.

    Raises:
        ValidationError: If `return_type` is a Pydantic model or a Bare type and the response cannot be
            parsed into the specified type.

    Examples:
        Basic map:
        >>> await session.map(
        ...     ["apple", "banana", "kiwi"],
        ...     template="What color is {}? Reply in a single word.",
        ... )
        ['Red.', 'Yellow.', 'Green.']

        Map with structured return type:
        >>> class Person(pydantic.BaseModel):
        ...     name: str
        ...     age: int
        >>> await session.map(
        ...     ["Barack Obama", "Angela Merkel"],
        ...     template="Who is {}?",
        ...     return_type=Person,
        ... )
        [Person(name='Barack Obama', age=62), Person(name='Angela Merkel', age=69)]

        Map with bare return type:
        >>> await session.map(
        ...     [42, 1337, 2025],
        ...     template="What are the unique prime factors of {}?",
        ...     return_type=Bare(list[int]),
        ... )
        [[2, 3, 7], [7, 191], [3, 5]]
    """

    formatter = template.format if isinstance(template, str) else template
    model = model if model is not None else self._model
    # case analysis for type checker
    if return_type is None:
        return await util.gather(*[self.prompt(formatter(item), model=model) for item in iterable])
    if isinstance(return_type, Bare):
        return await util.gather(
            *[self.prompt(formatter(item), return_type=return_type, model=model) for item in iterable]
        )
    return await util.gather(
        *[self.prompt(formatter(item), return_type=return_type, model=model) for item in iterable]
    )

max async

max[T](
    iterable: Iterable[T],
    /,
    *,
    by: str | None = None,
    to_str: Callable[[T], str] | None = None,
    template: str | Callable[[T, T], str] | None = None,
    task: Task | str | None = None,
    model: str | None = None,
) -> T

Get the largest item in an iterable.

This method finds the largest item in a collection by using a language model to perform pairwise comparisons.

This method is analogous to Python's built-in max function.

Parameters:

Name	Type	Description	Default
iterable	Iterable[T]	The collection of items to search.	required
by	str | None	A criteria specifying what aspect to compare by. If this is provided, template cannot be provided.	None
to_str	Callable[[T], str] | None	If specified, used to convert items to string representation. Otherewise, uses str() on each item. If this is provided, a callable template cannot be provided.	None
template	str | Callable[[T, T], str] | None	A custom prompt template for comparisons. Must be either a string template with two positional placeholders, or a callable that takes two items and returns a formatted string. If this is provided, by cannot be provided.	None
task	Task | str | None	The type of comparison task that is being performed in template. This allows for writing the template in the most convenient way possible (e.g., in some scenarios, it's easier to specify a criteria for which item is lesser, and in others, it's easier to specify a criteria for which item is greater). If this is provided, a custom template must also be provided. Defaults to Task.CHOOSE_GREATER if not specified.	None
model	str | None	If specified, overrides the default model for this call.	None

Returns:

Type	Description
T	The largest item.

Raises:

Type	Description
ValidationError	If parsing any LLM response fails.

Examples:

Basic usage:

>>> await session.max(
...     ["LeBron James", "Kobe Bryant", "Magic Johnson"], by="assists"
... )
'Magic Johnson'

Source code in src/semlib/extrema.py

async def max[T](
    self,
    iterable: Iterable[T],
    /,
    *,
    by: str | None = None,
    to_str: Callable[[T], str] | None = None,
    template: str | Callable[[T, T], str] | None = None,
    task: Task | str | None = None,
    model: str | None = None,
) -> T:
    """Get the largest item in an iterable.

    This method finds the largest item in a collection by using a language model to perform pairwise comparisons.

    This method is analogous to Python's built-in [`max`](https://docs.python.org/3/library/functions.html#max)
    function.

    Args:
        iterable: The collection of items to search.
        by: A criteria specifying what aspect to compare by. If this is provided, `template` cannot be
            provided.
        to_str: If specified, used to convert items to string representation. Otherewise, uses `str()` on each item.
            If this is provided, a callable template cannot be provided.
        template: A custom prompt template for comparisons. Must be either a string template with two positional
            placeholders, or a callable that takes two items and returns a formatted string. If this is provided,
            `by` cannot be provided.
        task: The type of comparison task that is being performed in `template`. This allows for writing the
            template in the most convenient way possible (e.g., in some scenarios, it's easier to specify a criteria
            for which item is lesser, and in others, it's easier to specify a criteria for which item is greater).
            If this is provided, a custom `template` must also be provided.  Defaults to
            [Task.CHOOSE_GREATER][semlib.compare.Task.CHOOSE_GREATER] if not specified.
        model: If specified, overrides the default model for this call.

    Returns:
        The largest item.

    Raises:
        ValidationError: If parsing any LLM response fails.

    Examples:
        Basic usage:
        >>> await session.max(
        ...     ["LeBron James", "Kobe Bryant", "Magic Johnson"], by="assists"
        ... )
        'Magic Johnson'
    """
    return await self._get_extreme(
        list(iterable),
        find_min=False,
        by=by,
        to_str=to_str,
        template=template,
        task=task,
        model=model,
    )

min async

min[T](
    iterable: Iterable[T],
    /,
    *,
    by: str | None = None,
    to_str: Callable[[T], str] | None = None,
    template: str | Callable[[T, T], str] | None = None,
    task: Task | str | None = None,
    model: str | None = None,
) -> T

Get the smallest item in an iterable.

This method finds the smallest item in a collection by using a language model to perform pairwise comparisons.

This method is analogous to Python's built-in min function.

Parameters:

Name	Type	Description	Default
iterable	Iterable[T]	The collection of items to search.	required
by	str | None	A criteria specifying what aspect to compare by. If this is provided, template cannot be provided.	None
to_str	Callable[[T], str] | None	If specified, used to convert items to string representation. Otherewise, uses str() on each item. If this is provided, a callable template cannot be provided.	None
template	str | Callable[[T, T], str] | None	A custom prompt template for comparisons. Must be either a string template with two positional placeholders, or a callable that takes two items and returns a formatted string. If this is provided, by cannot be provided.	None
task	Task | str | None	The type of comparison task that is being performed in template. This allows for writing the template in the most convenient way possible (e.g., in some scenarios, it's easier to specify a criteria for which item is lesser, and in others, it's easier to specify a criteria for which item is greater). If this is provided, a custom template must also be provided. Defaults to Task.CHOOSE_GREATER if not specified.	None
model	str | None	If specified, overrides the default model for this call.	None

Returns:

Type	Description
T	The smallest item.

Raises:

Type	Description
ValidationError	If parsing any LLM response fails.

Examples:

Basic usage:

>>> await session.min(["blue", "red", "green"], by="wavelength")
'blue'

Source code in src/semlib/extrema.py

async def min[T](
    self,
    iterable: Iterable[T],
    /,
    *,
    by: str | None = None,
    to_str: Callable[[T], str] | None = None,
    template: str | Callable[[T, T], str] | None = None,
    task: Task | str | None = None,
    model: str | None = None,
) -> T:
    """Get the smallest item in an iterable.

    This method finds the smallest item in a collection by using a language model to perform pairwise comparisons.

    This method is analogous to Python's built-in [`min`](https://docs.python.org/3/library/functions.html#min)
    function.

    Args:
        iterable: The collection of items to search.
        by: A criteria specifying what aspect to compare by. If this is provided, `template` cannot be
            provided.
        to_str: If specified, used to convert items to string representation. Otherewise, uses `str()` on each item.
            If this is provided, a callable template cannot be provided.
        template: A custom prompt template for comparisons. Must be either a string template with two positional
            placeholders, or a callable that takes two items and returns a formatted string. If this is provided,
            `by` cannot be provided.
        task: The type of comparison task that is being performed in `template`. This allows for writing the
            template in the most convenient way possible (e.g., in some scenarios, it's easier to specify a criteria
            for which item is lesser, and in others, it's easier to specify a criteria for which item is greater).
            If this is provided, a custom `template` must also be provided.  Defaults to
            [Task.CHOOSE_GREATER][semlib.compare.Task.CHOOSE_GREATER] if not specified.
        model: If specified, overrides the default model for this call.

    Returns:
        The smallest item.

    Raises:
        ValidationError: If parsing any LLM response fails.

    Examples:
        Basic usage:
        >>> await session.min(["blue", "red", "green"], by="wavelength")
        'blue'
    """
    return await self._get_extreme(
        list(iterable),
        find_min=True,
        by=by,
        to_str=to_str,
        template=template,
        task=task,
        model=model,
    )

prompt async

prompt[T: BaseModel](
    prompt: str,
    /,
    *,
    return_type: type[T],
    model: str | None = None,
) -> T

prompt[T](
    prompt: str,
    /,
    *,
    return_type: Bare[T],
    model: str | None = None,
) -> T

prompt(
    prompt: str,
    /,
    *,
    return_type: None = None,
    model: str | None = None,
) -> str

Send a prompt to the language model and get a response.

This method sends a single user message to the language model and returns the response. The response can be returned as a raw string, parsed into a Pydantic model, or extracted as a bare value using the Bare marker.

Parameters:

Name	Type	Description	Default
prompt	str	The text prompt to send to the language model.	required
return_type	type[T] | Bare[U] | None	If not specified, the response is returned as a raw string. If a Pydantic model class is provided, the response is parsed into an instance of that model. If a Bare instance is provided, a single value of the specified type is extracted from the response.	None
model	str | None	If specified, overrides the default model for this call.	None

Returns:

Type	Description
str | T | U	The language model's response in the format specified by return_type.

Raises:

Type	Description
ValidationError	If return_type is a Pydantic model or a Bare type and the response cannot be parsed into the specified type.

Examples:

Get raw string response:

>>> await session.prompt("What is 2+2?")
'2 + 2 equals 4.'

Get structured value:

>>> class Person(BaseModel):
...     name: str
...     age: int
>>> await session.prompt("Who is Barack Obama?", return_type=Person)
Person(name='Barack Obama', age=62)

Get bare value:

>>> await session.prompt("What is 2+2?", return_type=Bare(int))
4

Source code in src/semlib/_internal/base.py

async def prompt[T: BaseModel, U](
    self, prompt: str, /, *, return_type: type[T] | Bare[U] | None = None, model: str | None = None
) -> str | T | U:
    """Send a prompt to the language model and get a response.

    This method sends a single user message to the language model and returns the response. The response can be
    returned as a raw string, parsed into a [Pydantic](https://pydantic.dev/) model, or extracted as a bare value
    using the [Bare][semlib.bare.Bare] marker.

    Args:
        prompt: The text prompt to send to the language model.
        return_type: If not specified, the response is returned as a raw string. If a Pydantic model class is
            provided, the response is parsed into an instance of that model. If a [Bare][semlib.bare.Bare] instance
            is provided, a single value of the specified type is extracted from the response.
        model: If specified, overrides the default model for this call.

    Returns:
        The language model's response in the format specified by return_type.

    Raises:
        ValidationError: If `return_type` is a Pydantic model or a Bare type and the response cannot be
            parsed into the specified type.

    Examples:
        Get raw string response:
        >>> await session.prompt("What is 2+2?")
        '2 + 2 equals 4.'

        Get structured value:
        >>> class Person(BaseModel):
        ...     name: str
        ...     age: int
        >>> await session.prompt("Who is Barack Obama?", return_type=Person)
        Person(name='Barack Obama', age=62)

        Get bare value:
        >>> await session.prompt("What is 2+2?", return_type=Bare(int))
        4
    """
    return await self._acompletion(
        messages=[Message(role="user", content=prompt)], return_type=return_type, model=model
    )

reduce async

reduce(
    iterable: Iterable[str],
    /,
    template: str | Callable[[str, str], str],
    *,
    associative: bool = False,
    model: str | None = None,
) -> str

reduce[T](
    iterable: Iterable[str | T],
    /,
    template: str | Callable[[str | T, str | T], str],
    *,
    associative: bool = False,
    model: str | None = None,
) -> str | T

reduce[T: BaseModel](
    iterable: Iterable[T],
    /,
    template: str | Callable[[T, T], str],
    *,
    return_type: type[T],
    associative: bool = False,
    model: str | None = None,
) -> T

reduce[T](
    iterable: Iterable[T],
    /,
    template: str | Callable[[T, T], str],
    *,
    return_type: Bare[T],
    associative: bool = False,
    model: str | None = None,
) -> T

reduce[T, U: BaseModel](
    iterable: Iterable[T],
    /,
    template: str | Callable[[U, T], str],
    initial: U,
    *,
    return_type: type[U],
    model: str | None = None,
) -> U

reduce[T, U](
    iterable: Iterable[T],
    /,
    template: str | Callable[[U, T], str],
    initial: U,
    *,
    return_type: Bare[U],
    model: str | None = None,
) -> U

Reduce an iterable to a single value using a language model.

This method is analogous to Python's functools.reduce function.

Parameters:

Name	Type	Description	Default
iterable	Iterable[Any]	The collection of items to reduce.	required
template	str | Callable[[Any, Any], str]	A prompt template to apply to each item. This can be either a string template with two positional placeholders (with the first placeholder being the accumulator and the second placeholder being an item), or a callable that takes an accumulator and an item and returns a formatted string.	required
initial	Any	If provided, this value is placed before the items of the iterable in the calculation, and serves as a default when the iterable is empty.	None
return_type	Any	The return type is also the type of the accumulator. If not specified, the responses are returned as raw strings. If a Pydantic model class is provided, the responses are parsed into instances of that model. If a Bare instance is provided, single values of the specified type are extracted from the responses.	None
associative	bool	If True, the reduction is performed in a balanced tree manner, which unlocks concurrency and can provide significant speedups for large iterables. This requires the reduction operation to be associative.	False
model	str | None	If specified, overrides the default model for this call.	None

Returns:

Type	Description
Any	The final accumulated value.

Raises:

Type	Description
ValidationError	If return_type is a Pydantic model or a Bare type and the response cannot be parsed into the specified type.

Examples:

Basic reduce:

>>> await session.reduce(
...     ["one", "three", "seven", "twelve"], "{} + {} = ?", return_type=Bare(int)
... )
23

Reduce with initial value:

>>> await session.reduce(
...     range(20),
...     template=lambda acc, n: f"If {n} is prime, append it to this list: {acc}.",
...     initial=[],
...     return_type=Bare(list[int]),
...     model="openai/o4-mini",
... )
[2, 3, 5, 7, 11, 13, 17, 19]

Associative reduce:

>>> await session.reduce(
...     [[i] for i in range(20)],
...     template=lambda acc,
...     n: f"Compute the union of these two sets, and then remove any non-prime numbers: {acc} and {n}. Return the result as a list.",
...     return_type=Bare(list[int]),
...     associative=True,
...     model="openai/o4-mini",
... )
[2, 3, 5, 7, 11, 13, 17, 19]

Distinguishing between leaf nodes and internal nodes in an associative reduce with Box:

>>> reviews: list[str] = [
...     "The instructions are a bit confusing. It took me a while to figure out how to use it.",
...     "It's so loud!",
...     "I regret buying this microwave. It's the worst appliance I've ever owned.",
...     "This microwave is great! It heats up food quickly and evenly.",
...     "This microwave is a waste of money. It doesn't work at all.",
...     "I hate the design of this microwave. It looks cheap and ugly.",
...     "The turntable is a bit small, so I can't fit larger plates in it.",
...     "The microwave is a bit noisy when it's running.",
...     "The microwave is a bit expensive compared to other models with similar features.",
...     "The turntable is useless, so I can't fit any plates in it.",
...     "I love the sleek design of this microwave. It looks great in my kitchen.",
... ]
>>> def template(a: str | Box[str], b: str | Box[str]) -> str:
...     # leaf nodes (raw reviews)
...     if isinstance(a, Box) and isinstance(b, Box):
...         return f'''
... Consider the following two product reviews, and return a bulleted list
... summarizing any actionable product improvements that could be made based on
... the reviews. If there are no actionable product improvements, return an empty
... string.
...
... - Review 1: {a.value}
... - Review 2: {b.value}'''
...     # summaries of reviews
...     if not isinstance(a, Box) and not isinstance(b, Box):
...         return f'''
... Consider the following two lists of ideas for product improvements, and
... combine them while de-duplicating similar ideas. If there are no ideas, return
... an empty string.
...
... # List 1:
... {a}
...
... # List 2:
... {b}'''
...     # one is a summary, the other is a raw review
...     if isinstance(a, Box) and not isinstance(b, Box):
...         ideas = b
...         review = a.value
...     if not isinstance(a, Box) and isinstance(b, Box):
...         ideas = a
...         review = b.value
...     return f'''
... Consider the following list of ideas for product improvements, and a product
... review. Update the list of ideas based on the review, de-duplicating similar
... ideas. If there are no ideas, return an empty string.
...
... # List of ideas:
... {ideas}
...
... # Review:
... {review}'''
>>> result = await session.reduce(
...     map(Box, reviews), template=template, associative=True
... )
>>> print(result)
- Clarify and simplify the product instructions to make them easier to understand.
- Consider reducing the noise level of the product to make it quieter during operation.
- Improve product reliability to ensure the microwave functions correctly for all users.
- Increase the size or adjust the design of the turntable to accommodate larger plates.
- Improve the design to enhance the aesthetic appeal and make it look more premium.

Source code in src/semlib/reduce.py

async def reduce(
    self,
    iterable: Iterable[Any],
    /,
    template: str | Callable[[Any, Any], str],
    initial: Any = None,
    *,
    return_type: Any = None,
    associative: bool = False,
    model: str | None = None,
) -> Any:
    """Reduce an iterable to a single value using a language model.

    This method is analogous to Python's
    [`functools.reduce`](https://docs.python.org/3/library/functools.html#functools.reduce) function.

    Args:
        iterable: The collection of items to reduce.
        template: A prompt template to apply to each item. This can be either a string template with two positional
            placeholders (with the first placeholder being the accumulator and the second placeholder being an
            item), or a callable that takes an accumulator and an item and returns a formatted string.
        initial: If provided, this value is placed before the items of the iterable in the calculation, and serves
            as a default when the iterable is empty.
        return_type: The return type is also the type of the accumulator. If not specified, the responses are
            returned as raw strings. If a Pydantic model class is provided, the responses are parsed into instances
            of that model. If a [Bare][semlib.bare.Bare] instance is provided, single values of the specified type
            are extracted from the responses.
        associative: If `True`, the reduction is performed in a balanced tree manner, which unlocks concurrency and
            can provide significant speedups for large iterables. This requires the reduction operation to be
            associative.
        model: If specified, overrides the default model for this call.

    Returns:
        The final accumulated value.

    Raises:
        ValidationError: If `return_type` is a Pydantic model or a Bare type and the response cannot be
            parsed into the specified type.

    Examples:
        Basic reduce:
        >>> await session.reduce(
        ...     ["one", "three", "seven", "twelve"], "{} + {} = ?", return_type=Bare(int)
        ... )
        23

        Reduce with initial value:
        >>> await session.reduce(
        ...     range(20),
        ...     template=lambda acc, n: f"If {n} is prime, append it to this list: {acc}.",
        ...     initial=[],
        ...     return_type=Bare(list[int]),
        ...     model="openai/o4-mini",
        ... )
        [2, 3, 5, 7, 11, 13, 17, 19]

        Associative reduce:
        >>> await session.reduce(
        ...     [[i] for i in range(20)],
        ...     template=lambda acc,
        ...     n: f"Compute the union of these two sets, and then remove any non-prime numbers: {acc} and {n}. Return the result as a list.",
        ...     return_type=Bare(list[int]),
        ...     associative=True,
        ...     model="openai/o4-mini",
        ... )
        [2, 3, 5, 7, 11, 13, 17, 19]

        Distinguishing between leaf nodes and internal nodes in an associative reduce with [Box][semlib.box.Box]:

        >>> reviews: list[str] = [
        ...     "The instructions are a bit confusing. It took me a while to figure out how to use it.",
        ...     "It's so loud!",
        ...     "I regret buying this microwave. It's the worst appliance I've ever owned.",
        ...     "This microwave is great! It heats up food quickly and evenly.",
        ...     "This microwave is a waste of money. It doesn't work at all.",
        ...     "I hate the design of this microwave. It looks cheap and ugly.",
        ...     "The turntable is a bit small, so I can't fit larger plates in it.",
        ...     "The microwave is a bit noisy when it's running.",
        ...     "The microwave is a bit expensive compared to other models with similar features.",
        ...     "The turntable is useless, so I can't fit any plates in it.",
        ...     "I love the sleek design of this microwave. It looks great in my kitchen.",
        ... ]
        >>> def template(a: str | Box[str], b: str | Box[str]) -> str:
        ...     # leaf nodes (raw reviews)
        ...     if isinstance(a, Box) and isinstance(b, Box):
        ...         return f'''
        ... Consider the following two product reviews, and return a bulleted list
        ... summarizing any actionable product improvements that could be made based on
        ... the reviews. If there are no actionable product improvements, return an empty
        ... string.
        ...
        ... - Review 1: {a.value}
        ... - Review 2: {b.value}'''
        ...     # summaries of reviews
        ...     if not isinstance(a, Box) and not isinstance(b, Box):
        ...         return f'''
        ... Consider the following two lists of ideas for product improvements, and
        ... combine them while de-duplicating similar ideas. If there are no ideas, return
        ... an empty string.
        ...
        ... # List 1:
        ... {a}
        ...
        ... # List 2:
        ... {b}'''
        ...     # one is a summary, the other is a raw review
        ...     if isinstance(a, Box) and not isinstance(b, Box):
        ...         ideas = b
        ...         review = a.value
        ...     if not isinstance(a, Box) and isinstance(b, Box):
        ...         ideas = a
        ...         review = b.value
        ...     return f'''
        ... Consider the following list of ideas for product improvements, and a product
        ... review. Update the list of ideas based on the review, de-duplicating similar
        ... ideas. If there are no ideas, return an empty string.
        ...
        ... # List of ideas:
        ... {ideas}
        ...
        ... # Review:
        ... {review}'''
        >>> result = await session.reduce(
        ...     map(Box, reviews), template=template, associative=True
        ... )
        >>> print(result)
        - Clarify and simplify the product instructions to make them easier to understand.
        - Consider reducing the noise level of the product to make it quieter during operation.
        - Improve product reliability to ensure the microwave functions correctly for all users.
        - Increase the size or adjust the design of the turntable to accommodate larger plates.
        - Improve the design to enhance the aesthetic appeal and make it look more premium.
    """
    if initial is not None:
        return await self._reduce2(
            iterable,
            template,
            initial,
            return_type=return_type,
            model=model,
        )
    return await self._reduce1(
        iterable,
        template,
        return_type=return_type,
        associative=associative,
        model=model,
    )

sort async

sort[T](
    iterable: Iterable[T],
    /,
    *,
    by: str | None = None,
    to_str: Callable[[T], str] | None = None,
    template: str | Callable[[T, T], str] | None = None,
    task: Task | str | None = None,
    algorithm: Algorithm | None = None,
    reverse: bool = False,
    model: str | None = None,
) -> list[T]

Sort an iterable.

This method sorts a collection of items by using a language model to perform pairwise comparisons. The sorting algorithm determines which comparisons to make and how to aggregate the results into a final ranking.

This method is analogous to Python's built-in sorted function.

Parameters:

Name	Type	Description	Default
iterable	Iterable[T]	The collection of items to sort.	required
by	str | None	A criteria specifying what aspect to compare by. If this is provided, template cannot be provided.	None
to_str	Callable[[T], str] | None	If specified, used to convert items to string representation. Otherewise, uses str() on each item. If this is provided, a callable template cannot be provided.	None
template	str | Callable[[T, T], str] | None	A custom prompt template for comparisons. Must be either a string template with two positional placeholders, or a callable that takes two items and returns a formatted string. If this is provided, by cannot be provided.	None
task	Task | str | None	The type of comparison task that is being performed in template. This allows for writing the template in the most convenient way possible (e.g., in some scenarios, it's easier to specify a criteria for which item is lesser, and in others, it's easier to specify a criteria for which item is greater). If this is provided, a custom template must also be provided. Defaults to Task.CHOOSE_GREATER if not specified.	None
algorithm	Algorithm | None	The sorting algorithm to use. If not specified, defaults to BordaCount. Different algorithms make different tradeoffs between accuracy, latency, and cost. See the documentation for each algorithm for details.	None
reverse	bool	If True, sort in descending order. If False, sort in ascending order.	False
model	str | None	If specified, overrides the default model for this call.	None

Returns:

Type	Description
list[T]	A new list containing all items from the iterable in sort order.

Raises:

Type	Description
ValidationError	If parsing any LLM response fails.

Examples:

Basic sort:

>>> await session.sort(["blue", "red", "green"], by="wavelength", reverse=True)
['red', 'green', 'blue']

Custom template and task:

>>> from dataclasses import dataclass
>>> @dataclass
... class Person:
...     name: str
...     job: str
>>> people = [
...     Person(name="Barack Obama", job="President of the United States"),
...     Person(name="Dalai Lama", job="Spiritual Leader of Tibet"),
...     Person(name="Sundar Pichai", job="CEO of Google"),
... ]
>>> await session.sort(
...     people,
...     template=lambda a, b: f"Which job earns more, (a) {a.job} or (b) {b.job}?",
... )
[
    Person(name='Dalai Lama', job='Spiritual Leader of Tibet'),
    Person(name='Barack Obama', job='President of the United States'),
    Person(name='Sundar Pichai', job='CEO of Google'),
]

Source code in src/semlib/sort/sort.py

async def sort[T](
    self,
    iterable: Iterable[T],
    /,
    *,
    by: str | None = None,
    to_str: Callable[[T], str] | None = None,
    template: str | Callable[[T, T], str] | None = None,
    task: Task | str | None = None,
    algorithm: Algorithm | None = None,
    reverse: bool = False,
    model: str | None = None,
) -> list[T]:
    """Sort an iterable.

    This method sorts a collection of items by using a language model to perform pairwise comparisons. The sorting
    algorithm determines which comparisons to make and how to aggregate the results into a final ranking.

    This method is analogous to Python's built-in
    [`sorted`](https://docs.python.org/3/library/functions.html#sorted) function.

    Args:
        iterable: The collection of items to sort.
        by: A criteria specifying what aspect to compare by. If this is provided, `template` cannot be
            provided.
        to_str: If specified, used to convert items to string representation. Otherewise, uses `str()` on each item.
            If this is provided, a callable template cannot be provided.
        template: A custom prompt template for comparisons. Must be either a string template with two positional
            placeholders, or a callable that takes two items and returns a formatted string. If this is provided,
            `by` cannot be provided.
        task: The type of comparison task that is being performed in `template`. This allows for writing the
            template in the most convenient way possible (e.g., in some scenarios, it's easier to specify a criteria
            for which item is lesser, and in others, it's easier to specify a criteria for which item is greater).
            If this is provided, a custom `template` must also be provided.  Defaults to
            [Task.CHOOSE_GREATER][semlib.compare.Task.CHOOSE_GREATER] if not specified.
        algorithm: The sorting algorithm to use. If not specified, defaults to
            [BordaCount][semlib.sort.algorithm.BordaCount]. Different algorithms make different tradeoffs between
            accuracy, latency, and cost. See the documentation for each algorithm for details.
        reverse: If `True`, sort in descending order. If `False`, sort in ascending order.
        model: If specified, overrides the default model for this call.

    Returns:
        A new list containing all items from the iterable in sort order.

    Raises:
        ValidationError: If parsing any LLM response fails.

    Examples:
        Basic sort:
        >>> await session.sort(["blue", "red", "green"], by="wavelength", reverse=True)
        ['red', 'green', 'blue']

        Custom template and task:
        >>> from dataclasses import dataclass
        >>> @dataclass
        ... class Person:
        ...     name: str
        ...     job: str
        >>> people = [
        ...     Person(name="Barack Obama", job="President of the United States"),
        ...     Person(name="Dalai Lama", job="Spiritual Leader of Tibet"),
        ...     Person(name="Sundar Pichai", job="CEO of Google"),
        ... ]
        >>> await session.sort(
        ...     people,
        ...     template=lambda a, b: f"Which job earns more, (a) {a.job} or (b) {b.job}?",
        ... )
        [
            Person(name='Dalai Lama', job='Spiritual Leader of Tibet'),
            Person(name='Barack Obama', job='President of the United States'),
            Person(name='Sundar Pichai', job='CEO of Google'),
        ]
    """
    algorithm = algorithm if algorithm is not None else BordaCount()

    async def comparator(a: T, b: T) -> Order:
        return await self.compare(a, b, by=by, to_str=to_str, template=template, task=task, model=model)

    return await algorithm._sort(  # noqa: SLF001
        iterable, reverse=reverse, comparator=comparator, max_concurrency=self._max_concurrency
    )

total_cost

total_cost() -> float

Get the total cost incurred so far for API calls made through this instance.

Returns:

Type	Description
float	The total cost in USD.

Source code in src/semlib/_internal/base.py

def total_cost(self) -> float:
    """Get the total cost incurred so far for API calls made through this instance.

    Returns:
        The total cost in USD.
    """
    return self._total_cost

QueryCache

Bases: ABC

Abstract base class for a cache of LLM query results.

Caches can be used with Session to avoid repeating identical queries to the LLM.

Source code in src/semlib/cache.py

class QueryCache(ABC):
    """Abstract base class for a cache of LLM query results.

    Caches can be used with [Session][semlib.session.Session] to avoid repeating identical queries to the LLM.
    """

    @abstractmethod
    def _set[T: BaseModel](self, key: CacheKey[T], value: str) -> None: ...

    @abstractmethod
    def _get[T: BaseModel](self, key: CacheKey[T]) -> str | None: ...

    @abstractmethod
    def clear(self) -> None: ...

    @abstractmethod
    def __len__(self) -> int: ...

    def _hash_key[T: BaseModel](self, key: CacheKey[T]) -> bytes:
        messages, pydantic_model, llm_model = key
        key_components: list[str] = [llm_model]
        key_components.extend(message.to_json() for message in messages)
        if pydantic_model is not None:
            key_components.append(json.dumps(pydantic_model.model_json_schema()))
        h = sha256()
        for part in key_components:
            h.update(part.encode("utf-8"))
        return h.digest()

OnDiskCache

Bases: QueryCache

A persistent on-disk cache of LLM query results, backed by SQLite.

Source code in src/semlib/cache.py

class OnDiskCache(QueryCache):
    """A persistent on-disk cache of LLM query results, backed by SQLite."""

    @override
    def __init__(self, path: str) -> None:
        """Initialize an on-disk cache.

        Args:
            path: Path to the SQLite database file. If the file does not exist, it will be created. By convention, the
                filename should have a ".db" or ".sqlite" extension.
        """
        self._conn = sqlite3.connect(path, autocommit=True)
        self._conn.execute(
            """
            CREATE TABLE IF NOT EXISTS metadata (
                key TEXT PRIMARY KEY,
                value TEXT NOT NULL
            )
            """
        )
        cur = self._conn.execute("SELECT value FROM metadata WHERE key = ?", (_VERSION_KEY,))
        row = cur.fetchone()
        if row is None:
            self._conn.execute("INSERT INTO metadata (key, value) VALUES (?, ?)", (_VERSION_KEY, _VERSION))
        elif row[0] != _VERSION:
            msg = f"cache version mismatch: expected {_VERSION}, got {row[0]}"
            raise ValueError(msg)
        self._conn.execute(
            """
            CREATE TABLE IF NOT EXISTS data (
                key BLOB PRIMARY KEY,
                value TEXT NOT NULL
            )
            """
        )

    @override
    def _set[T: BaseModel](self, key: CacheKey[T], value: str) -> None:
        self._conn.execute(
            "INSERT OR REPLACE INTO data (key, value) VALUES (?, ?)",
            (self._hash_key(key), value),
        )

    @override
    def _get[T: BaseModel](self, key: CacheKey[T]) -> str | None:
        cur = self._conn.execute("SELECT value FROM data WHERE key = ?", (self._hash_key(key),))
        row = cur.fetchone()
        if row is None:
            return None
        return cast(str, row[0])

    @override
    def clear(self) -> None:
        self._conn.execute("DELETE FROM data")

    @override
    def __len__(self) -> int:
        cur = self._conn.execute("SELECT COUNT(*) FROM data")
        row = cur.fetchone()
        return cast(int, row[0])

__init__

__init__(path: str) -> None

Initialize an on-disk cache.

Parameters:

Name	Type	Description	Default
path	str	Path to the SQLite database file. If the file does not exist, it will be created. By convention, the filename should have a ".db" or ".sqlite" extension.	required

Source code in src/semlib/cache.py

@override
def __init__(self, path: str) -> None:
    """Initialize an on-disk cache.

    Args:
        path: Path to the SQLite database file. If the file does not exist, it will be created. By convention, the
            filename should have a ".db" or ".sqlite" extension.
    """
    self._conn = sqlite3.connect(path, autocommit=True)
    self._conn.execute(
        """
        CREATE TABLE IF NOT EXISTS metadata (
            key TEXT PRIMARY KEY,
            value TEXT NOT NULL
        )
        """
    )
    cur = self._conn.execute("SELECT value FROM metadata WHERE key = ?", (_VERSION_KEY,))
    row = cur.fetchone()
    if row is None:
        self._conn.execute("INSERT INTO metadata (key, value) VALUES (?, ?)", (_VERSION_KEY, _VERSION))
    elif row[0] != _VERSION:
        msg = f"cache version mismatch: expected {_VERSION}, got {row[0]}"
        raise ValueError(msg)
    self._conn.execute(
        """
        CREATE TABLE IF NOT EXISTS data (
            key BLOB PRIMARY KEY,
            value TEXT NOT NULL
        )
        """
    )

InMemoryCache

Bases: QueryCache

An in-memory cache of LLM query results.

Source code in src/semlib/cache.py

class InMemoryCache(QueryCache):
    """An in-memory cache of LLM query results."""

    @override
    def __init__(self) -> None:
        """Initialize an in-memory cache."""
        self._data: dict[bytes, str] = {}

    @override
    def _set[T: BaseModel](self, key: CacheKey[T], value: str) -> None:
        self._data[self._hash_key(key)] = value

    @override
    def _get[T: BaseModel](self, key: CacheKey[T]) -> str | None:
        return self._data.get(self._hash_key(key))

    @override
    def clear(self) -> None:
        self._data.clear()

    @override
    def __len__(self) -> int:
        return len(self._data)

__init__

__init__() -> None

Initialize an in-memory cache.

Source code in src/semlib/cache.py

@override
def __init__(self) -> None:
    """Initialize an in-memory cache."""
    self._data: dict[bytes, str] = {}

Bare

A marker to indicate that a function should return a bare value of type T.

This can be passed to the return_type parameter of functions like prompt. For situations where you want to extract a single value of a given base type (like int or list[float]), this is more convenient than the alternative of defining a Pydantic model with a single field for the purpose of extracting that value.

Examples:

Extract a bare value using prompt:

>>> await session.prompt("What is 2+2?", return_type=Bare(int))
4

Influence model output using class_name and field_name:

>>> await session.prompt(
...     "Give me a list",
...     return_type=Bare(
...         list[int], class_name="list_of_three_values", field_name="primes"
...     ),
... )
[3, 7, 11]

Source code in src/semlib/bare.py

class Bare[T]:
    """A marker to indicate that a function should return a bare value of type `T`.

    This can be passed to the `return_type` parameter of functions like [prompt][semlib._internal.base.Base.prompt]. For
    situations where you want to extract a single value of a given base type (like `int` or `list[float]`), this is more
    convenient than the alternative of defining a Pydantic model with a single field for the purpose of extracting that
    value.

    Examples:
        Extract a bare value using prompt:
        >>> await session.prompt("What is 2+2?", return_type=Bare(int))
        4

        Influence model output using `class_name` and `field_name`:
        >>> await session.prompt(
        ...     "Give me a list",
        ...     return_type=Bare(
        ...         list[int], class_name="list_of_three_values", field_name="primes"
        ...     ),
        ... )
        [3, 7, 11]
    """

    def __init__(self, typ: type[T], /, class_name: str | None = None, field_name: str | None = None):
        """Initialize a Bare instance.

        Args:
            typ: The type of the bare value to extract.
            class_name: Name for a dynamically created Pydantic model class. If not provided, defaults to
                the name of `typ`. This name is visible to the LLM and may affect model output.
            field_name: Name for the field in the dynamically created Pydantic model that holds the bare value.
                If not provided, defaults to "value". This name is visible to the LLM and may affect model output.
        """
        self._typ = typ
        self._class_name = class_name if class_name is not None else typ.__name__
        self._field_name = field_name if field_name is not None else "value"
        field_definitions: Any = {self._field_name: (self._typ, ...)}
        self._model: type[pydantic.BaseModel] = pydantic.create_model(self._class_name, **field_definitions)

    def _extract(self, obj: Any) -> T:
        if isinstance(obj, self._model):
            return cast(T, getattr(obj, self._field_name))
        msg = f"expected instance of {self._model.__name__}, got {type(obj).__name__}"
        raise TypeError(msg)

__init__

__init__(
    typ: type[T],
    /,
    class_name: str | None = None,
    field_name: str | None = None,
)

Initialize a Bare instance.

Parameters:

Name	Type	Description	Default
typ	type[T]	The type of the bare value to extract.	required
class_name	str | None	Name for a dynamically created Pydantic model class. If not provided, defaults to the name of typ. This name is visible to the LLM and may affect model output.	None
field_name	str | None	Name for the field in the dynamically created Pydantic model that holds the bare value. If not provided, defaults to "value". This name is visible to the LLM and may affect model output.	None

Source code in src/semlib/bare.py

def __init__(self, typ: type[T], /, class_name: str | None = None, field_name: str | None = None):
    """Initialize a Bare instance.

    Args:
        typ: The type of the bare value to extract.
        class_name: Name for a dynamically created Pydantic model class. If not provided, defaults to
            the name of `typ`. This name is visible to the LLM and may affect model output.
        field_name: Name for the field in the dynamically created Pydantic model that holds the bare value.
            If not provided, defaults to "value". This name is visible to the LLM and may affect model output.
    """
    self._typ = typ
    self._class_name = class_name if class_name is not None else typ.__name__
    self._field_name = field_name if field_name is not None else "value"
    field_definitions: Any = {self._field_name: (self._typ, ...)}
    self._model: type[pydantic.BaseModel] = pydantic.create_model(self._class_name, **field_definitions)

Box

A container that holds a value of type T.

This can be used to tag values, so that they can be distinguished from other values of the same underlying type. Such a tag can be useful in the context of methods like reduce, where you can use this marker to distinguish leaf nodes from internal nodes in an associative reduce.

Source code in src/semlib/box.py

class Box[T]:
    """A container that holds a value of type `T`.

    This can be used to tag values, so that they can be distinguished from other values of the same underlying type.
    Such a tag can be useful in the context of methods like [reduce][semlib.reduce.Reduce.reduce], where you can use
    this marker to distinguish leaf nodes from internal nodes in an associative reduce.
    """

    def __init__(self, value: T) -> None:
        """Initialize a Box instance.

        Args:
            value: The value to be contained in the Box.
        """
        self._value = value

    @property
    def value(self) -> T:
        """Get the value contained in the Box.

        Returns:
            The value contained in the Box.
        """
        return self._value

value property

value: T

Get the value contained in the Box.

Returns:

Type	Description
T	The value contained in the Box.

__init__

__init__(value: T) -> None

Initialize a Box instance.

Parameters:

Name	Type	Description	Default
value	T	The value to be contained in the Box.	required

Source code in src/semlib/box.py

def __init__(self, value: T) -> None:
    """Initialize a Box instance.

    Args:
        value: The value to be contained in the Box.
    """
    self._value = value

:::