Model: openai/gpt-4.1-mini
-- none --
050384c5942ada355d94cdc149b13fe42b50b112a1108bb2863d5ca847edb585
# from docs.plugins.llm-markov.llm_markov import register_models
def register_models(register):
# from docs.plugins.llm-markov.llm_markov import build_markov_table
def build_markov_table(text):
# from docs.plugins.llm-markov.llm_markov import generate
def generate(transitions, length, start_word):
# from docs.plugins.llm-markov.llm_markov import Markov
class Markov:
def execute(self, prompt, stream, response, conversation):
# from llm.__init__ import get_plugins
def get_plugins(all):
# from llm.__init__ import get_models_with_aliases
def get_models_with_aliases() -> List[?]:
# from llm.__init__ import _get_loaders
def _get_loaders(hook_method) -> Dict[(str, Callable)]:
# from llm.__init__ import get_template_loaders
def get_template_loaders() -> Dict[(str, Callable[(?, Template)])]:
"Get template loaders registered by plugins."
# from llm.__init__ import get_fragment_loaders
def get_fragment_loaders() -> Dict[(str, Callable[(?, Union[(Fragment, List[Fragment])])])]:
"Get fragment loaders registered by plugins."
# from llm.__init__ import get_embedding_models_with_aliases
def get_embedding_models_with_aliases() -> List[?]:
# from llm.__init__ import get_embedding_models
def get_embedding_models():
# from llm.__init__ import get_embedding_model
def get_embedding_model(name):
# from llm.__init__ import get_embedding_model_aliases
def get_embedding_model_aliases() -> Dict[(str, EmbeddingModel)]:
# from llm.__init__ import get_async_model_aliases
def get_async_model_aliases() -> Dict[(str, AsyncModel)]:
# from llm.__init__ import get_model_aliases
def get_model_aliases() -> Dict[(str, Model)]:
# from llm.__init__ import UnknownModelError
class UnknownModelError(KeyError):
# from llm.__init__ import get_models
def get_models() -> List[Model]:
"Get all registered models"
# from llm.__init__ import get_async_models
def get_async_models() -> List[AsyncModel]:
"Get all registered async models"
# from llm.__init__ import get_async_model
def get_async_model(name: Optional[str]) -> AsyncModel:
"Get an async model by name or alias"
# from llm.__init__ import get_model
def get_model(name: Optional[str], _skip_async: bool) -> Model:
"Get a model by name or alias"
# from llm.__init__ import get_key
def get_key(explicit_key: Optional[str], key_alias: str, env_var: Optional[str]) -> Optional[str]:
"""Return an API key based on a hierarchy of potential sources.
:param provided_key: A key provided by the user. This may be the key, or an alias of a key in keys.json.
:param key_alias: The alias used to retrieve the key from the keys.json file.
:param env_var: Name of the environment variable to check for the key."""
# from llm.__init__ import load_keys
def load_keys():
# from llm.__init__ import user_dir
def user_dir():
# from llm.__init__ import set_alias
def set_alias(alias, model_id_or_alias):
"Set an alias to point to the specified model."
# from llm.__init__ import remove_alias
def remove_alias(alias):
"Remove an alias."
# from llm.__init__ import encode
def encode(values):
# from llm.__init__ import decode
def decode(binary):
# from llm.__init__ import cosine_similarity
def cosine_similarity(a, b):
# from llm.__init__ import get_default_model
def get_default_model(filename="default_model.txt", default=DEFAULT_MODEL):
# from llm.__init__ import set_default_model
def set_default_model(model, filename="default_model.txt"):
# from llm.__init__ import get_default_embedding_model
def get_default_embedding_model():
# from llm.__init__ import set_default_embedding_model
def set_default_embedding_model(model):
# from llm.cli import FragmentNotFound
class FragmentNotFound(Exception):
# from llm.cli import validate_fragment_alias
def validate_fragment_alias(ctx, param, value):
# from llm.cli import resolve_fragments
def resolve_fragments(db: ?, fragments: Iterable[str]) -> List[Fragment]:
"Resolve fragments into a list of (content, source) tuples"
# from llm.cli import AttachmentError
class AttachmentError(Exception):
"Exception raised for errors in attachment resolution."
# from llm.cli import resolve_attachment
def resolve_attachment(value):
"""Resolve an attachment from a string value which could be:
- "-" for stdin
- A URL
- A file path
Returns an Attachment object.
Raises AttachmentError if the attachment cannot be resolved."""
# from llm.cli import AttachmentType
class AttachmentType:
def convert(self, value, param, ctx):
# from llm.cli import resolve_attachment_with_type
def resolve_attachment_with_type(value: str, mimetype: str) -> Attachment:
# from llm.cli import attachment_types_callback
def attachment_types_callback(ctx, param, values) -> List[Attachment]:
# from llm.cli import json_validator
def json_validator(object_name):
# from llm.cli import schema_option
def schema_option(fn):
# from llm.cli import cli
def cli():
"""Access Large Language Models from the command-line
Documentation: https://llm.datasette.io/
LLM can run models from many different providers. Consult the
plugin directory for a list of available models:
https://llm.datasette.io/en/stable/plugins/directory.html
To get started with OpenAI, obtain an API key from them and:
�
$ llm keys set openai
Enter key: ...
Then execute a prompt like this:
llm 'Five outrageous names for a pet pelican'
For a full list of prompting options run:
llm prompt --help"""
# from llm.cli import prompt
def prompt(prompt, system, model_id, database, queries, attachments, attachment_types, options, schema_input, schema_multi, fragments, system_fragments, template, param, no_stream, no_log, log, _continue, conversation_id, key, save, async_, usage, extract, extract_last):
"""Execute a prompt
Documentation: https://llm.datasette.io/en/stable/usage.html
Examples:
�
llm 'Capital of France?'
llm 'Capital of France?' -m gpt-4o
llm 'Capital of France?' -s 'answer in Spanish'
Multi-modal models can be called with attachments like this:
�
llm 'Extract text from this image' -a image.jpg
llm 'Describe' -a https://static.simonwillison.net/static/2024/pelicans.jpg
cat image | llm 'describe image' -a -
# With an explicit mimetype:
cat image | llm 'describe image' --at - image/jpeg
The -x/--extract option returns just the content of the first ``` fenced code
block, if one is present. If none are present it returns the full response.
�
llm 'JavaScript function for reversing a string' -x"""
# from llm.cli import chat
def chat(system, model_id, _continue, conversation_id, template, param, options, no_stream, key):
"Hold an ongoing chat with a model."
# from llm.cli import load_conversation
def load_conversation(conversation_id: Optional[str], async_) -> Optional[_BaseConversation]:
# from llm.cli import keys
def keys():
"Manage stored API keys for different models"
# from llm.cli import keys_list
def keys_list():
"List names of all stored keys"
# from llm.cli import keys_path_command
def keys_path_command():
"Output the path to the keys.json file"
# from llm.cli import keys_get
def keys_get(name):
"""Return the value of a stored key
Example usage:
�
export OPENAI_API_KEY=$(llm keys get openai)"""
# from llm.cli import keys_set
def keys_set(name, value):
"""Save a key in the keys.json file
Example usage:
�
$ llm keys set openai
Enter key: ..."""
# from llm.cli import logs
def logs():
"Tools for exploring logged prompts and responses"
# from llm.cli import logs_path
def logs_path():
"Output the path to the logs.db file"
# from llm.cli import logs_status
def logs_status():
"Show current status of database logging"
# from llm.cli import backup
def backup(path):
"Backup your logs database to this file"
# from llm.cli import logs_turn_on
def logs_turn_on():
"Turn on logging for all prompts"
# from llm.cli import logs_turn_off
def logs_turn_off():
"Turn off logging for all prompts"
# from llm.cli import logs_list
def logs_list(count, path, database, model, query, fragments, schema_input, schema_multi, data, data_array, data_key, data_ids, truncate, short, usage, response, extract, extract_last, current_conversation, conversation_id, id_gt, id_gte, json_output, expand):
"Show logged prompts and their responses"
# from llm.cli import models
def models():
"Manage available models"
# from llm.cli import models_list
def models_list(options, async_, schemas, query, model_ids):
"List available models"
# from llm.cli import models_default
def models_default(model):
"Show or set the default model"
# from llm.cli import templates
def templates():
"Manage stored prompt templates"
# from llm.cli import templates_list
def templates_list():
"List available prompt templates"
# from llm.cli import templates_show
def templates_show(name):
"Show the specified prompt template"
# from llm.cli import templates_edit
def templates_edit(name):
"Edit the specified prompt template using the default $EDITOR"
# from llm.cli import templates_path
def templates_path():
"Output the path to the templates directory"
# from llm.cli import templates_loaders
def templates_loaders():
"Show template loaders registered by plugins"
# from llm.cli import schemas
def schemas():
"Manage stored schemas"
# from llm.cli import schemas_list
def schemas_list(path, database, queries, full):
"List stored schemas"
# from llm.cli import schemas_show
def schemas_show(schema_id, path, database):
"Show a stored schema"
# from llm.cli import schemas_dsl_debug
def schemas_dsl_debug(input, multi):
"""Convert LLM's schema DSL to a JSON schema
�
llm schema dsl 'name, age int, bio: their bio'"""
# from llm.cli import aliases
def aliases():
"Manage model aliases"
# from llm.cli import aliases_list
def aliases_list(json_):
"List current aliases"
# from llm.cli import aliases_set
def aliases_set(alias, model_id, query):
"""Set an alias for a model
Example usage:
�
llm aliases set mini gpt-4o-mini
Alternatively you can omit the model ID and specify one or more -q options.
The first model matching all of those query strings will be used.
�
llm aliases set mini -q 4o -q mini"""
# from llm.cli import aliases_remove
def aliases_remove(alias):
"""Remove an alias
Example usage:
�
$ llm aliases remove turbo"""
# from llm.cli import aliases_path
def aliases_path():
"Output the path to the aliases.json file"
# from llm.cli import fragments
def fragments():
"""Manage fragments that are stored in the database
Fragments are reusable snippets of text that are shared across multiple prompts."""
# from llm.cli import fragments_list
def fragments_list(queries, aliases, json_):
"List current fragments"
# from llm.cli import fragments_set
def fragments_set(alias, fragment):
"""Set an alias for a fragment
Accepts an alias and a file path, URL, hash or '-' for stdin
Example usage:
�
llm fragments set mydocs ./docs.md"""
# from llm.cli import fragments_show
def fragments_show(alias_or_hash):
"""Display the fragment stored under an alias or hash
�
llm fragments show mydocs"""
# from llm.cli import fragments_remove
def fragments_remove(alias):
"""Remove a fragment alias
Example usage:
�
llm fragments remove docs"""
# from llm.cli import fragments_loaders
def fragments_loaders():
"Show fragment loaders registered by plugins"
# from llm.cli import plugins_list
def plugins_list(all):
"List installed plugins"
# from llm.cli import display_truncated
def display_truncated(text):
# from llm.cli import install
def install(packages, upgrade, editable, force_reinstall, no_cache_dir):
"Install packages from PyPI into the same environment as LLM"
# from llm.cli import uninstall
def uninstall(packages, yes):
"Uninstall Python packages from the LLM environment"
# from llm.cli import embed
def embed(collection, id, input, model, store, database, content, binary, metadata, format_):
"Embed text and store or return the result"
# from llm.cli import embed_multi
def embed_multi(collection, input_path, format, files, encodings, binary, sql, attach, batch_size, prefix, model, prepend, store, database):
"""Store embeddings for multiple strings at once in the specified collection.
Input data can come from one of three sources:
�
1. A CSV, TSV, JSON or JSONL file:
- CSV/TSV: First column is ID, remaining columns concatenated as content
- JSON: Array of objects with "id" field and content fields
- JSONL: Newline-delimited JSON objects
�
Examples:
llm embed-multi docs input.csv
cat data.json | llm embed-multi docs -
llm embed-multi docs input.json --format json
�
2. A SQL query against a SQLite database:
- First column returned is used as ID
- Other columns concatenated to form content
�
Examples:
llm embed-multi docs --sql "SELECT id, title, body FROM posts"
llm embed-multi docs --attach blog blog.db --sql "SELECT id, content FROM blog.posts"
�
3. Files in directories matching glob patterns:
- Each file becomes one embedding
- Relative file paths become IDs
�
Examples:
llm embed-multi docs --files docs '**/*.md'
llm embed-multi images --files photos '*.jpg' --binary
llm embed-multi texts --files texts '*.txt' --encoding utf-8 --encoding latin-1"""
# from llm.cli import similar
def similar(collection, id, input, content, binary, number, plain, database):
"""Return top N similar IDs from a collection using cosine similarity.
Example usage:
�
llm similar my-collection -c "I like cats"
Or to find content similar to a specific stored ID:
�
llm similar my-collection 1234"""
# from llm.cli import embed_models
def embed_models():
"Manage available embedding models"
# from llm.cli import embed_models_list
def embed_models_list(query):
"List available embedding models"
# from llm.cli import embed_models_default
def embed_models_default(model, remove_default):
"Show or set the default embedding model"
# from llm.cli import collections
def collections():
"View and manage collections of embeddings"
# from llm.cli import collections_path
def collections_path():
"Output the path to the embeddings database"
# from llm.cli import embed_db_collections
def embed_db_collections(database, json_):
"View a list of collections"
# from llm.cli import collections_delete
def collections_delete(collection, database):
"""Delete the specified collection
Example usage:
�
llm collections delete my-collection"""
# from llm.cli import options
def options():
"Manage default options for models"
# from llm.cli import options_list
def options_list():
"""List default options for all models
Example usage:
�
llm models options list"""
# from llm.cli import options_show
def options_show(model):
"""List default options set for a specific model
Example usage:
�
llm models options show gpt-4o"""
# from llm.cli import options_set
def options_set(model, key, value):
"""Set a default option for a model
Example usage:
�
llm models options set gpt-4o temperature 0.5"""
# from llm.cli import options_clear
def options_clear(model, key):
"""Clear default option(s) for a model
Example usage:
�
llm models options clear gpt-4o
# Or for a single option
llm models options clear gpt-4o temperature"""
# from llm.cli import template_dir
def template_dir():
# from llm.cli import logs_db_path
def logs_db_path():
# from llm.cli import get_history
def get_history(chat_id):
# from llm.cli import render_errors
def render_errors(errors):
# from llm.cli import _human_readable_size
def _human_readable_size(size_bytes):
# from llm.cli import logs_on
def logs_on():
# from llm.cli import get_all_model_options
def get_all_model_options() -> dict:
"Get all default options for all models"
# from llm.cli import get_model_options
def get_model_options(model_id: str) -> dict:
"""Get default options for a specific model
Args:
model_id: Return options for model with this ID
Returns:
A dictionary of model options"""
# from llm.cli import set_model_option
def set_model_option(model_id: str, key: str, value: Any) -> ?:
"""Set a default option for a model.
Args:
model_id: The model ID
key: The option key
value: The option value"""
# from llm.cli import clear_model_option
def clear_model_option(model_id: str, key: str) -> ?:
"""Clear a model option
Args:
model_id: The model ID
key: Key to clear"""
# from llm.cli import LoadTemplateError
class LoadTemplateError(ValueError):
# from llm.cli import _parse_yaml_template
def _parse_yaml_template(name, content):
# from llm.cli import load_template
def load_template(name: str) -> Template:
"Load template, or raise LoadTemplateError(msg)"
# from llm.default_plugins.chunkers import lines
def lines(text):
"Chunk text into lines"
# from llm.default_plugins.chunkers import register_chunker_functions
def register_chunker_functions(register):
# from llm.default_plugins.openai_models import register_models
def register_models(register):
# from llm.default_plugins.openai_models import register_embedding_models
def register_embedding_models(register):
# from llm.default_plugins.openai_models import OpenAIEmbeddingModel
class OpenAIEmbeddingModel(EmbeddingModel):
def __init__(self, model_id, openai_model_id, dimensions):
def embed_batch(self, items: Iterable[Union[(str, bytes)]]) -> Iterator[List[float]]:
# from llm.default_plugins.openai_models import register_commands
def register_commands(cli):
# from llm.default_plugins.openai_models import SharedOptions
class SharedOptions:
def validate_logit_bias(cls, logit_bias):
# from llm.default_plugins.openai_models import ReasoningEffortEnum
class ReasoningEffortEnum(str, Enum):
# from llm.default_plugins.openai_models import OptionsForReasoning
class OptionsForReasoning(SharedOptions):
# from llm.default_plugins.openai_models import _attachment
def _attachment(attachment):
# from llm.default_plugins.openai_models import _Shared
class _Shared:
def __init__(self, model_id, key, model_name, api_base, api_type, api_version, api_engine, headers, can_stream=True, vision, audio, reasoning, supports_schema, allows_system_prompt=True):
def __str__(self):
def build_messages(self, prompt, conversation):
def set_usage(self, response, usage):
def get_client(self, key, *, async_):
def build_kwargs(self, prompt, stream):
# from llm.default_plugins.openai_models import Chat
class Chat(_Shared, KeyModel):
def execute(self, prompt, stream, response, conversation, key):
# from llm.default_plugins.openai_models import AsyncChat
class AsyncChat(_Shared, AsyncKeyModel):
async def execute(self, prompt, stream, response, conversation, key) -> AsyncGenerator[(str, ?)]:
# from llm.default_plugins.openai_models import Completion
class Completion(Chat):
def __init__(self, *, default_max_tokens, *args, **kwargs):
def __str__(self):
def execute(self, prompt, stream, response, conversation, key):
# from llm.default_plugins.openai_models import not_nulls
def not_nulls(data) -> dict:
# from llm.default_plugins.openai_models import combine_chunks
def combine_chunks(chunks: List) -> dict:
# from llm.default_plugins.openai_models import redact_data
def redact_data(input_dict):
"""Recursively search through the input dictionary for any 'image_url' keys
and modify the 'url' value to be just 'data:...'.
Also redact input_audio.data keys"""
# from llm.embeddings import Entry
class Entry:
# from llm.embeddings import Collection
class Collection:
def __init__(self, name: str, db: Optional[Database], *, model: Optional[EmbeddingModel], model_id: Optional[str], create: bool) -> ?:
"""A collection of embeddings
Returns the collection with the given name, creating it if it does not exist.
If you set create=False a Collection.DoesNotExist exception will be raised if the
collection does not already exist.
Args:
db (sqlite_utils.Database): Database to store the collection in
name (str): Name of the collection
model (llm.models.EmbeddingModel, optional): Embedding model to use
model_id (str, optional): Alternatively, ID of the embedding model to use
create (bool, optional): Whether to create the collection if it does not exist"""
def model(self) -> EmbeddingModel:
"Return the embedding model used by this collection"
def count(self) -> int:
"""Count the number of items in the collection.
Returns:
int: Number of items in the collection"""
def embed(self, id: str, value: Union[(str, bytes)], metadata: Optional[Dict[(str, Any)]], store: bool) -> ?:
"""Embed value and store it in the collection with a given ID.
Args:
id (str): ID for the value
value (str or bytes): value to be embedded
metadata (dict, optional): Metadata to be stored
store (bool, optional): Whether to store the value in the content or content_blob column"""
def embed_multi(self, entries: Iterable[Tuple[(str, Union[(str, bytes)])]], store: bool, batch_size: int=100) -> ?:
"""Embed multiple texts and store them in the collection with given IDs.
Args:
entries (iterable): Iterable of (id: str, text: str) tuples
store (bool, optional): Whether to store the text in the content column
batch_size (int, optional): custom maximum batch size to use"""
def embed_multi_with_metadata(self, entries: Iterable[Tuple[(str, Union[(str, bytes)], Optional[Dict[(str, Any)]])]], store: bool, batch_size: int=100) -> ?:
"""Embed multiple values along with metadata and store them in the collection with given IDs.
Args:
entries (iterable): Iterable of (id: str, value: str or bytes, metadata: None or dict)
store (bool, optional): Whether to store the value in the content or content_blob column
batch_size (int, optional): custom maximum batch size to use"""
def similar_by_vector(self, vector: List[float], number: int=10, skip_id: Optional[str]) -> List[Entry]:
"""Find similar items in the collection by a given vector.
Args:
vector (list): Vector to search by
number (int, optional): Number of similar items to return
Returns:
list: List of Entry objects"""
def similar_by_id(self, id: str, number: int=10) -> List[Entry]:
"""Find similar items in the collection by a given ID.
Args:
id (str): ID to search by
number (int, optional): Number of similar items to return
Returns:
list: List of Entry objects"""
def similar(self, value: Union[(str, bytes)], number: int=10) -> List[Entry]:
"""Find similar items in the collection by a given value.
Args:
value (str or bytes): value to search by
number (int, optional): Number of similar items to return
Returns:
list: List of Entry objects"""
def exists(cls, db: Database, name: str) -> bool:
"""Does this collection exist in the database?
Args:
name (str): Name of the collection"""
def delete(self):
"Delete the collection and its embeddings from the database"
def content_hash(input: Union[(str, bytes)]) -> bytes:
"Hash content for deduplication. Override to change hashing behavior."
# from llm.embeddings_migrations import m001_create_tables
def m001_create_tables(db):
# from llm.embeddings_migrations import m002_foreign_key
def m002_foreign_key(db):
# from llm.embeddings_migrations import m003_add_updated
def m003_add_updated(db):
# from llm.embeddings_migrations import m004_store_content_hash
def m004_store_content_hash(db):
# from llm.embeddings_migrations import m005_add_content_blob
def m005_add_content_blob(db):
# from llm.errors import ModelError
class ModelError(Exception):
"Models can raise this error, which will be displayed to the user"
# from llm.errors import NeedsKeyException
class NeedsKeyException(ModelError):
"Model needs an API key which has not been provided"
# from llm.hookspecs import register_commands
def register_commands(cli):
"Register additional CLI commands, e.g. 'llm mycommand ...'"
# from llm.hookspecs import register_models
def register_models(register):
"Register additional model instances representing LLM models that can be called"
# from llm.hookspecs import register_embedding_models
def register_embedding_models(register):
"Register additional model instances that can be used for embedding"
# from llm.hookspecs import register_template_loaders
def register_template_loaders(register):
"Register additional template loaders with prefixes"
# from llm.hookspecs import register_fragment_loaders
def register_fragment_loaders(register):
"Register additional fragment loaders with prefixes"
# from llm.migrations import migrate
def migrate(db):
# from llm.migrations import ensure_migrations_table
def ensure_migrations_table(db):
# from llm.migrations import m001_initial
def m001_initial(db):
# from llm.migrations import m002_id_primary_key
def m002_id_primary_key(db):
# from llm.migrations import m003_chat_id_foreign_key
def m003_chat_id_foreign_key(db):
# from llm.migrations import m004_column_order
def m004_column_order(db):
# from llm.migrations import m004_drop_provider
def m004_drop_provider(db):
# from llm.migrations import m005_debug
def m005_debug(db):
# from llm.migrations import m006_new_logs_table
def m006_new_logs_table(db):
# from llm.migrations import m007_finish_logs_table
def m007_finish_logs_table(db):
# from llm.migrations import m008_reply_to_id_foreign_key
def m008_reply_to_id_foreign_key(db):
# from llm.migrations import m008_fix_column_order_in_logs
def m008_fix_column_order_in_logs(db):
# from llm.migrations import m009_delete_logs_table_if_empty
def m009_delete_logs_table_if_empty(db):
# from llm.migrations import m010_create_new_log_tables
def m010_create_new_log_tables(db):
# from llm.migrations import m011_fts_for_responses
def m011_fts_for_responses(db):
# from llm.migrations import m012_attachments_tables
def m012_attachments_tables(db):
# from llm.migrations import m013_usage
def m013_usage(db):
# from llm.migrations import m014_schemas
def m014_schemas(db):
# from llm.migrations import m015_fragments_tables
def m015_fragments_tables(db):
# from llm.migrations import m016_fragments_table_pks
def m016_fragments_table_pks(db):
# from llm.migrations import m017_first_token_ms
def m017_first_token_ms(db):
# from llm.models import SpeedEstimate
class SpeedEstimate:
def __str__(self):
# from llm.models import Usage
class Usage:
# from llm.models import Attachment
class Attachment:
def id(self):
def resolve_type(self):
def content_bytes(self):
def base64_content(self):
def from_row(cls, row):
# from llm.models import Prompt
class Prompt:
def __init__(self, prompt, model, *, fragments, attachments, system, system_fragments, prompt_json, options, schema):
def prompt(self):
def system(self):
# from llm.models import _BaseConversation
class _BaseConversation:
def from_row(cls, row: Any) -> ?:
# from llm.models import Conversation
class Conversation(_BaseConversation):
def prompt(self, prompt: Optional[str], *, fragments: Optional[List[str]], attachments: Optional[List[Attachment]], system: Optional[str], schema: Optional[Union[(dict, type[BaseModel])]], system_fragments: Optional[List[str]], stream: bool, key: Optional[str], **options) -> ?:
def from_row(cls, row):
def __repr__(self):
# from llm.models import AsyncConversation
class AsyncConversation(_BaseConversation):
def prompt(self, prompt: Optional[str], *, fragments: Optional[List[str]], attachments: Optional[List[Attachment]], system: Optional[str], schema: Optional[Union[(dict, type[BaseModel])]], system_fragments: Optional[List[str]], stream: bool, key: Optional[str], **options) -> ?:
def from_row(cls, row):
def __repr__(self):
# from llm.models import _BaseResponse
class _BaseResponse:
"Base response class shared between sync and async responses"
def __init__(self, prompt: Prompt, model: ?, stream: bool, conversation: Optional[_BaseConversation], key: Optional[str]):
def set_usage(self, *, input: Optional[int], output: Optional[int], details: Optional[dict]):
def from_row(cls, db, row, _async):
def token_usage(self) -> str:
def log_to_db(self, db):
# from llm.models import Response
class Response(_BaseResponse):
def on_done(self, callback):
def _on_done(self):
def __str__(self) -> str:
def _force(self):
def text(self) -> str:
def text_or_raise(self) -> str:
def json(self) -> Optional[Dict[(str, Any)]]:
def duration_ms(self) -> int:
def first_token_ms(self) -> Optional[int]:
def speed_estimate(self) -> SpeedEstimate:
def datetime_utc(self) -> str:
def usage(self) -> Usage:
def __iter__(self) -> Iterator[str]:
def __repr__(self):
# from llm.models import AsyncResponse
class AsyncResponse(_BaseResponse):
def from_row(cls, db, row, _async):
async def on_done(self, callback):
async def _on_done(self):
def __aiter__(self):
async def __anext__(self) -> str:
async def _force(self):
def text_or_raise(self) -> str:
async def text(self) -> str:
async def json(self) -> Optional[Dict[(str, Any)]]:
async def duration_ms(self) -> int:
async def datetime_utc(self) -> str:
async def usage(self) -> Usage:
def __await__(self):
async def to_sync_response(self) -> Response:
def fake(cls, model: ?, prompt: str, *, system: str, response: str, *attachments):
"Utility method to help with writing tests"
def __repr__(self):
# from llm.models import Options
class Options(BaseModel):
# from llm.models import _get_key_mixin
class _get_key_mixin:
def get_key(self, explicit_key: Optional[str]) -> Optional[str]:
# from llm.models import _BaseModel
class _BaseModel(ABC, _get_key_mixin):
def _validate_attachments(self, attachments: Optional[List[Attachment]]) -> ?:
def __str__(self) -> str:
def __repr__(self) -> str:
# from llm.models import _Model
class _Model(_BaseModel):
def conversation(self) -> Conversation:
def prompt(self, prompt: Optional[str], *, fragments: Optional[List[str]], attachments: Optional[List[Attachment]], system: Optional[str], system_fragments: Optional[List[str]], stream: bool, schema: Optional[Union[(dict, type[BaseModel])]], **options) -> Response:
# from llm.models import Model
class Model(_Model):
def execute(self, prompt: Prompt, stream: bool, response: Response, conversation: Optional[Conversation]) -> Iterator[str]:
# from llm.models import KeyModel
class KeyModel(_Model):
def execute(self, prompt: Prompt, stream: bool, response: Response, conversation: Optional[Conversation], key: Optional[str]) -> Iterator[str]:
# from llm.models import _AsyncModel
class _AsyncModel(_BaseModel):
def conversation(self) -> AsyncConversation:
def prompt(self, prompt: Optional[str], *, fragments: Optional[List[str]], attachments: Optional[List[Attachment]], system: Optional[str], schema: Optional[Union[(dict, type[BaseModel])]], system_fragments: Optional[List[str]], stream: bool, **options) -> AsyncResponse:
# from llm.models import AsyncModel
class AsyncModel(_AsyncModel):
async def execute(self, prompt: Prompt, stream: bool, response: AsyncResponse, conversation: Optional[AsyncConversation]) -> AsyncGenerator[(str, ?)]:
# from llm.models import AsyncKeyModel
class AsyncKeyModel(_AsyncModel):
async def execute(self, prompt: Prompt, stream: bool, response: AsyncResponse, conversation: Optional[AsyncConversation], key: Optional[str]) -> AsyncGenerator[(str, ?)]:
# from llm.models import EmbeddingModel
class EmbeddingModel(ABC, _get_key_mixin):
def _check(self, item: Union[(str, bytes)]):
def embed(self, item: Union[(str, bytes)]) -> List[float]:
"Embed a single text string or binary blob, return a list of floats"
def embed_multi(self, items: Iterable[Union[(str, bytes)]], batch_size: Optional[int]) -> Iterator[List[float]]:
"Embed multiple items in batches according to the model batch_size"
def embed_batch(self, items: Iterable[Union[(str, bytes)]]) -> Iterator[List[float]]:
"Embed a batch of strings or blobs, return a list of lists of floats"
def __str__(self) -> str:
def __repr__(self) -> str:
# from llm.models import ModelWithAliases
class ModelWithAliases:
def matches(self, query: str) -> bool:
# from llm.models import EmbeddingModelWithAliases
class EmbeddingModelWithAliases:
def matches(self, query: str) -> bool:
# from llm.models import _conversation_name
def _conversation_name(text):
# from llm.plugins import load_plugins
def load_plugins():
# from llm.serve import error
async def error(send, status_code: int, message: str):
# from llm.serve import read_request_body
async def read_request_body(receive):
"Reads and concatenates all HTTP request body chunks into a single bytes object."
# from llm.serve import handle_completions_request
async def handle_completions_request(scope, receive, send):
"Handle POST /v1/completions with possible streaming (SSE) or non-streamed JSON output."
# from llm.serve import app
async def app(scope, receive, send):
"A simple ASGI application that routes /v1/completions to our OpenAI-compatible handler."
# from llm.templates import AttachmentType
class AttachmentType(BaseModel):
# from llm.templates import Template
class Template(BaseModel):
def evaluate(self, input: str, params: Optional[Dict[(str, Any)]]) -> Tuple[(Optional[str], Optional[str])]:
def vars(self) -> set:
def interpolate(cls, text: Optional[str], params: Dict[(str, Any)]) -> Optional[str]:
def extract_vars(string_template: ?) -> List[str]:
# from llm.utils import Fragment
class Fragment(str):
def __new__(cls, content, *args, **kwargs):
def __init__(self, content, source=""):
def id(self):
# from llm.utils import mimetype_from_string
def mimetype_from_string(content) -> Optional[str]:
# from llm.utils import mimetype_from_path
def mimetype_from_path(path) -> Optional[str]:
# from llm.utils import dicts_to_table_string
def dicts_to_table_string(headings: List[str], dicts: List[Dict[(str, str)]]) -> List[str]:
# from llm.utils import remove_dict_none_values
def remove_dict_none_values(d):
"Recursively remove keys with value of None or value of a dict that is all values of None"
# from llm.utils import _LogResponse
class _LogResponse:
def iter_bytes(self, *args, **kwargs):
# from llm.utils import _LogTransport
class _LogTransport:
def __init__(self, transport: ?):
def handle_request(self, request: ?) -> ?:
# from llm.utils import _no_accept_encoding
def _no_accept_encoding(request: ?):
# from llm.utils import _log_response
def _log_response(response: ?):
# from llm.utils import logging_client
def logging_client() -> ?:
# from llm.utils import simplify_usage_dict
def simplify_usage_dict(d):
# from llm.utils import token_usage_string
def token_usage_string(input_tokens, output_tokens, token_details) -> str:
# from llm.utils import extract_fenced_code_block
def extract_fenced_code_block(text: str, last: bool) -> Optional[str]:
"""Extracts and returns Markdown fenced code block found in the given text.
The function handles fenced code blocks that:
- Use at least three backticks (`).
- May include a language tag immediately after the opening backticks.
- Use more than three backticks as long as the closing fence has the same number.
If no fenced code block is found, the function returns None.
Args:
text (str): The input text to search for a fenced code block.
last (bool): Extract the last code block if True, otherwise the first.
Returns:
Optional[str]: The content of the fenced code block, or None if not found."""
# from llm.utils import make_schema_id
def make_schema_id(schema: dict) -> Tuple[(str, str)]:
# from llm.utils import output_rows_as_json
def output_rows_as_json(rows, nl):
"""Output rows as JSON - either newline-delimited or an array
Parameters:
- rows: List of dictionaries to output
- nl: Boolean, if True, use newline-delimited JSON
Returns:
- String with formatted JSON output"""
# from llm.utils import resolve_schema_input
def resolve_schema_input(db, schema_input, load_template):
# from llm.utils import schema_summary
def schema_summary(schema: dict) -> str:
"""Extract property names from a JSON schema and format them in a
concise way that highlights the array/object structure.
Args:
schema (dict): A JSON schema dictionary
Returns:
str: A human-friendly summary of the schema structure"""
# from llm.utils import schema_dsl
def schema_dsl(schema_dsl: str, multi: bool) -> Dict[(str, Any)]:
"""Build a JSON schema from a concise schema string.
Args:
schema_dsl: A string representing a schema in the concise format.
Can be comma-separated or newline-separated.
multi: Boolean, return a schema for an "items" array of these
Returns:
A dictionary representing the JSON schema."""
# from llm.utils import multi_schema
def multi_schema(schema: dict) -> dict:
"Wrap JSON schema in an 'items': [] array"
# from llm.utils import find_unused_key
def find_unused_key(item: dict, key: str) -> str:
"Return unused key, e.g. for {\"id\": \"1\"} and key \"id\" returns \"id_\""
# from llm.utils import truncate_string
def truncate_string(text: str, max_length: int=100, normalize_whitespace: bool, keep_end: bool) -> str:
"""Truncate a string to a maximum length, with options to normalize whitespace and keep both start and end.
Args:
text: The string to truncate
max_length: Maximum length of the result string
normalize_whitespace: If True, replace all whitespace with a single space
keep_end: If True, keep both beginning and end of string
Returns:
Truncated string"""
# from llm.utils import ensure_fragment
def ensure_fragment(db, content):
# from llm.utils import maybe_fenced_code
def maybe_fenced_code(content: str) -> str:
"Return the content as a fenced code block if it looks like code"
# from llm.utils import has_plugin_prefix
def has_plugin_prefix(value: str) -> bool:
"Check if value starts with alphanumeric prefix followed by a colon"
# from tests.conftest import pytest_configure
def pytest_configure(config):
# from tests.conftest import user_path
def user_path(tmpdir):
# from tests.conftest import logs_db
def logs_db(user_path):
# from tests.conftest import user_path_with_embeddings
def user_path_with_embeddings(user_path):
# from tests.conftest import templates_path
def templates_path(user_path):
# from tests.conftest import env_setup
def env_setup(monkeypatch, user_path):
# from tests.conftest import MockModel
class MockModel:
def __init__(self):
def enqueue(self, messages):
def execute(self, prompt, stream, response, conversation):
# from tests.conftest import EchoModel
class EchoModel:
def execute(self, prompt, stream, response, conversation):
# from tests.conftest import MockKeyModel
class MockKeyModel:
def execute(self, prompt, stream, response, conversation, key):
# from tests.conftest import MockAsyncKeyModel
class MockAsyncKeyModel:
async def execute(self, prompt, stream, response, conversation, key):
# from tests.conftest import AsyncMockModel
class AsyncMockModel:
def __init__(self):
def enqueue(self, messages):
async def execute(self, prompt, stream, response, conversation):
# from tests.conftest import EmbedDemo
class EmbedDemo:
def __init__(self):
def embed_batch(self, texts):
# from tests.conftest import EmbedBinaryOnly
class EmbedBinaryOnly(EmbedDemo):
# from tests.conftest import EmbedTextOnly
class EmbedTextOnly(EmbedDemo):
# from tests.conftest import embed_demo
def embed_demo():
# from tests.conftest import mock_model
def mock_model():
# from tests.conftest import async_mock_model
def async_mock_model():
# from tests.conftest import mock_key_model
def mock_key_model():
# from tests.conftest import mock_async_key_model
def mock_async_key_model():
# from tests.conftest import register_embed_demo_model
def register_embed_demo_model(embed_demo, mock_model, async_mock_model):
# from tests.conftest import register_echo_model
def register_echo_model():
# from tests.conftest import mocked_openai_chat
def mocked_openai_chat(httpx_mock):
# from tests.conftest import mocked_openai_chat_returning_fenced_code
def mocked_openai_chat_returning_fenced_code(httpx_mock):
# from tests.conftest import stream_events
def stream_events():
# from tests.conftest import mocked_openai_chat_stream
def mocked_openai_chat_stream(httpx_mock):
# from tests.conftest import mocked_openai_completion
def mocked_openai_completion(httpx_mock):
# from tests.conftest import stream_completion_events
def stream_completion_events():
# from tests.conftest import mocked_openai_completion_logprobs_stream
def mocked_openai_completion_logprobs_stream(httpx_mock):
# from tests.conftest import mocked_openai_completion_logprobs
def mocked_openai_completion_logprobs(httpx_mock):
# from tests.conftest import mocked_localai
def mocked_localai(httpx_mock):
# from tests.conftest import collection
def collection():
# from tests.test_aliases import test_set_alias
def test_set_alias(model_id_or_alias):
# from tests.test_aliases import test_remove_alias
def test_remove_alias():
# from tests.test_aliases import test_cli_aliases_list
def test_cli_aliases_list(args):
# from tests.test_aliases import test_cli_aliases_list_json
def test_cli_aliases_list_json(args):
# from tests.test_aliases import test_cli_aliases_set
def test_cli_aliases_set(user_path, args, expected, expected_error):
# from tests.test_aliases import test_cli_aliases_path
def test_cli_aliases_path(user_path):
# from tests.test_aliases import test_cli_aliases_remove
def test_cli_aliases_remove(user_path):
# from tests.test_aliases import test_cli_aliases_remove_invalid
def test_cli_aliases_remove_invalid(user_path):
# from tests.test_aliases import test_cli_aliases_are_registered
def test_cli_aliases_are_registered(user_path, args):
# from tests.test_async import test_async_model
async def test_async_model(async_mock_model):
# from tests.test_async import test_async_model_conversation
async def test_async_model_conversation(async_mock_model):
# from tests.test_async import test_async_on_done
async def test_async_on_done(async_mock_model):
# from tests.test_async import test_async_conversation
async def test_async_conversation(async_mock_model):
# from tests.test_attachments import test_prompt_attachment
def test_prompt_attachment(mock_model, logs_db, attachment_type, attachment_content):
# from tests.test_chat import test_chat_basic
def test_chat_basic(mock_model, logs_db):
# from tests.test_chat import test_chat_system
def test_chat_system(mock_model, logs_db):
# from tests.test_chat import test_chat_options
def test_chat_options(mock_model, logs_db):
# from tests.test_chat import test_chat_multi
def test_chat_multi(mock_model, logs_db, input, expected):
# from tests.test_cli_openai_models import mocked_models
def mocked_models(httpx_mock):
# from tests.test_cli_openai_models import test_openai_models
def test_openai_models(mocked_models):
# from tests.test_cli_openai_models import test_openai_options_min_max
def test_openai_options_min_max():
# from tests.test_cli_openai_models import test_only_gpt4_audio_preview_allows_mp3_or_wav
def test_only_gpt4_audio_preview_allows_mp3_or_wav(httpx_mock, model, filetype):
# from tests.test_cli_openai_models import test_gpt4o_mini_sync_and_async
def test_gpt4o_mini_sync_and_async(monkeypatch, tmpdir, httpx_mock, async_, usage):
# from tests.test_cli_options import test_set_model_default_options
def test_set_model_default_options(user_path, args, expected_options, expected_error):
# from tests.test_cli_options import test_model_options_list_and_show
def test_model_options_list_and_show(user_path):
# from tests.test_cli_options import test_model_options_clear
def test_model_options_clear(user_path):
# from tests.test_embed import test_demo_plugin
def test_demo_plugin():
# from tests.test_embed import test_embed_huge_list
def test_embed_huge_list(batch_size, expected_batches):
# from tests.test_embed import test_embed_store
def test_embed_store(collection):
# from tests.test_embed import test_embed_metadata
def test_embed_metadata(collection):
# from tests.test_embed import test_collection
def test_collection(collection):
# from tests.test_embed import test_similar
def test_similar(collection):
# from tests.test_embed import test_similar_by_id
def test_similar_by_id(collection):
# from tests.test_embed import test_embed_multi
def test_embed_multi(with_metadata, batch_size, expected_batches):
# from tests.test_embed import test_collection_delete
def test_collection_delete(collection):
# from tests.test_embed import test_binary_only_and_text_only_embedding_models
def test_binary_only_and_text_only_embedding_models():
# from tests.test_embed_cli import test_embed_output_format
def test_embed_output_format(tmpdir, format_, expected, scenario):
# from tests.test_embed_cli import test_embed_errors
def test_embed_errors(args, expected_error):
# from tests.test_embed_cli import test_embed_store
def test_embed_store(user_path, metadata, metadata_error):
# from tests.test_embed_cli import test_embed_store_binary
def test_embed_store_binary(user_path):
# from tests.test_embed_cli import test_collection_delete_errors
def test_collection_delete_errors(user_path):
# from tests.test_embed_cli import test_similar_errors
def test_similar_errors(args, expected_error, user_path_with_embeddings):
# from tests.test_embed_cli import test_similar_by_id_cli
def test_similar_by_id_cli(user_path_with_embeddings):
# from tests.test_embed_cli import test_similar_by_id_cli_output_plain
def test_similar_by_id_cli_output_plain(user_path_with_embeddings, option):
# from tests.test_embed_cli import test_similar_by_content_cli
def test_similar_by_content_cli(tmpdir, user_path_with_embeddings, scenario):
# from tests.test_embed_cli import test_embed_multi_file_input
def test_embed_multi_file_input(tmpdir, use_stdin, prefix, prepend, filename, content):
# from tests.test_embed_cli import test_embed_multi_files_binary_store
def test_embed_multi_files_binary_store(tmpdir):
# from tests.test_embed_cli import test_embed_multi_sql
def test_embed_multi_sql(tmpdir, use_other_db, prefix, prepend):
# from tests.test_embed_cli import test_embed_multi_batch_size
def test_embed_multi_batch_size(embed_demo, tmpdir):
# from tests.test_embed_cli import multi_files
def multi_files(tmpdir):
# from tests.test_embed_cli import test_embed_multi_files
def test_embed_multi_files(multi_files, scenario, prepend):
# from tests.test_embed_cli import test_embed_multi_files_errors
def test_embed_multi_files_errors(multi_files, args, expected_error):
# from tests.test_embed_cli import test_embed_multi_files_encoding
def test_embed_multi_files_encoding(multi_files, extra_args, expected_error):
# from tests.test_embed_cli import test_default_embedding_model
def test_default_embedding_model():
# from tests.test_embed_cli import test_llm_embed_models_query
def test_llm_embed_models_query(user_path, args, expected_model_id):
# from tests.test_embed_cli import test_default_embed_model_errors
def test_default_embed_model_errors(user_path, default_is_set, command):
# from tests.test_embed_cli import test_duplicate_content_embedded_only_once
def test_duplicate_content_embedded_only_once(embed_demo):
# from tests.test_encode_decode import test_roundtrip
def test_roundtrip(array):
# from tests.test_fragments_cli import test_fragments_set_show_remove
def test_fragments_set_show_remove(user_path):
# from tests.test_keys import test_keys_in_user_path
def test_keys_in_user_path(monkeypatch, env, user_path):
# from tests.test_keys import test_keys_set
def test_keys_set(monkeypatch, tmpdir):
# from tests.test_keys import test_keys_get
def test_keys_get(monkeypatch, tmpdir):
# from tests.test_keys import test_keys_list
def test_keys_list(monkeypatch, tmpdir, args):
# from tests.test_keys import test_uses_correct_key
def test_uses_correct_key(mocked_openai_chat, monkeypatch, tmpdir):
# from tests.test_llm import test_version
def test_version():
# from tests.test_llm import test_llm_prompt_creates_log_database
def test_llm_prompt_creates_log_database(mocked_openai_chat, tmpdir, monkeypatch, custom_database_path):
# from tests.test_llm import test_llm_default_prompt
def test_llm_default_prompt(mocked_openai_chat, use_stdin, user_path, logs_off, logs_args, should_log):
# from tests.test_llm import test_llm_prompt_continue
def test_llm_prompt_continue(httpx_mock, user_path, async_):
# from tests.test_llm import test_extract_fenced_code
def test_extract_fenced_code(mocked_openai_chat_returning_fenced_code, args, expect_just_code):
# from tests.test_llm import test_openai_chat_stream
def test_openai_chat_stream(mocked_openai_chat_stream, user_path):
# from tests.test_llm import test_openai_completion
def test_openai_completion(mocked_openai_completion, user_path):
# from tests.test_llm import test_openai_completion_system_prompt_error
def test_openai_completion_system_prompt_error():
# from tests.test_llm import test_openai_completion_logprobs_stream
def test_openai_completion_logprobs_stream(mocked_openai_completion_logprobs_stream, user_path):
# from tests.test_llm import test_openai_completion_logprobs_nostream
def test_openai_completion_logprobs_nostream(mocked_openai_completion_logprobs, user_path):
# from tests.test_llm import test_openai_localai_configuration
def test_openai_localai_configuration(mocked_localai, user_path):
# from tests.test_llm import test_prompt_select_model_with_queries
def test_prompt_select_model_with_queries(mock_model, user_path, args, exit_code):
# from tests.test_llm import test_llm_models_options
def test_llm_models_options(user_path):
# from tests.test_llm import test_llm_models_async
def test_llm_models_async(user_path):
# from tests.test_llm import test_llm_models_filter
def test_llm_models_filter(user_path, args, expected_model_ids, unexpected_model_ids):
# from tests.test_llm import test_llm_user_dir
def test_llm_user_dir(tmpdir, monkeypatch):
# from tests.test_llm import test_model_defaults
def test_model_defaults(tmpdir, monkeypatch):
# from tests.test_llm import test_get_models
def test_get_models():
# from tests.test_llm import test_get_async_models
def test_get_async_models():
# from tests.test_llm import test_mock_model
def test_mock_model(mock_model):
# from tests.test_llm import Dog
class Dog(BaseModel):
# from tests.test_llm import test_schema
def test_schema(mock_model, use_pydantic):
# from tests.test_llm import test_model_environment_variable
def test_model_environment_variable(monkeypatch):
# from tests.test_llm import test_schema_via_cli
def test_schema_via_cli(mock_model, tmpdir, monkeypatch, use_filename):
# from tests.test_llm import test_schema_using_dsl
def test_schema_using_dsl(mock_model, tmpdir, monkeypatch, args, expected):
# from tests.test_llm import test_schema_async
async def test_schema_async(async_mock_model, use_pydantic):
# from tests.test_llm import test_mock_key_model
def test_mock_key_model(mock_key_model):
# from tests.test_llm import test_mock_async_key_model
async def test_mock_async_key_model(mock_async_key_model):
# from tests.test_llm import test_sync_on_done
def test_sync_on_done(mock_model):
# from tests.test_llm import test_schemas_dsl
def test_schemas_dsl():
# from tests.test_llm_logs import log_path
def log_path(user_path):
# from tests.test_llm_logs import schema_log_path
def schema_log_path(user_path):
# from tests.test_llm_logs import test_logs_text
def test_logs_text(log_path, usage):
# from tests.test_llm_logs import test_logs_json
def test_logs_json(n, log_path):
"Test that logs command correctly returns requested -n records"
# from tests.test_llm_logs import test_logs_response_only
def test_logs_response_only(args, log_path):
"Test that logs -r/--response returns just the last response"
# from tests.test_llm_logs import test_logs_extract_first_code
def test_logs_extract_first_code(args, log_path):
"Test that logs -x/--extract returns the first code block"
# from tests.test_llm_logs import test_logs_extract_last_code
def test_logs_extract_last_code(args, log_path):
"Test that logs --xl/--extract-last returns the last code block"
# from tests.test_llm_logs import test_logs_short
def test_logs_short(log_path, arg, usage):
# from tests.test_llm_logs import test_logs_path
def test_logs_path(monkeypatch, env, user_path):
# from tests.test_llm_logs import test_logs_filtered
def test_logs_filtered(user_path, model, path_option):
# from tests.test_llm_logs import test_logs_search
def test_logs_search(user_path, query, extra_args, expected):
# from tests.test_llm_logs import test_logs_schema
def test_logs_schema(schema_log_path, args, expected):
# from tests.test_llm_logs import test_logs_schema_data_ids
def test_logs_schema_data_ids(schema_log_path):
# from tests.test_llm_logs import fragments_fixture
def fragments_fixture(user_path):
# from tests.test_llm_logs import test_logs_fragments
def test_logs_fragments(fragments_fixture, fragment_refs, expected):
# from tests.test_llm_logs import test_logs_fragments_markdown
def test_logs_fragments_markdown(fragments_fixture):
# from tests.test_llm_logs import test_expand_fragment_json
def test_expand_fragment_json(fragments_fixture, arg):
# from tests.test_llm_logs import test_expand_fragment_markdown
def test_expand_fragment_markdown(fragments_fixture):
# from tests.test_llm_logs import test_logs_backup
def test_logs_backup(logs_db):
# from tests.test_migrate import test_migrate_blank
def test_migrate_blank():
# from tests.test_migrate import test_migrate_from_original_schema
def test_migrate_from_original_schema(has_record):
# from tests.test_migrate import test_migrations_with_legacy_alter_table
def test_migrations_with_legacy_alter_table():
# from tests.test_migrate import test_migrations_for_embeddings
def test_migrations_for_embeddings():
# from tests.test_migrate import test_backfill_content_hash
def test_backfill_content_hash():
# from tests.test_plugins import test_register_commands
def test_register_commands():
# from tests.test_plugins import test_register_template_loaders
def test_register_template_loaders():
# from tests.test_plugins import test_register_fragment_loaders
def test_register_fragment_loaders(logs_db):
# from tests.test_templates import test_template_evaluate
def test_template_evaluate(prompt, system, defaults, params, expected_prompt, expected_system, expected_error):
# from tests.test_templates import test_templates_list_no_templates_found
def test_templates_list_no_templates_found():
# from tests.test_templates import test_templates_list
def test_templates_list(templates_path, args):
# from tests.test_templates import test_templates_prompt_save
def test_templates_prompt_save(templates_path, args, expected, expected_error):
# from tests.test_templates import test_templates_error_on_missing_schema
def test_templates_error_on_missing_schema(templates_path):
# from tests.test_templates import test_execute_prompt_with_a_template
def test_execute_prompt_with_a_template(templates_path, mocked_openai_chat, template, input_text, extra_args, expected_model, expected_input, expected_error, expected_options):
# from tests.test_templates import test_execute_prompt_from_template_url
def test_execute_prompt_from_template_url(httpx_mock, template, expected):
# from tests.test_templates import test_execute_prompt_from_template_path
def test_execute_prompt_from_template_path():
# from tests.test_utils import test_simplify_usage_dict
def test_simplify_usage_dict(input_data, expected_output):
# from tests.test_utils import test_extract_fenced_code_block
def test_extract_fenced_code_block(input, last, expected):
# from tests.test_utils import test_schema_dsl
def test_schema_dsl(schema, expected):
# from tests.test_utils import test_schema_dsl_multi
def test_schema_dsl_multi():
# from tests.test_utils import test_truncate_string
def test_truncate_string(text, max_length, normalize_whitespace, keep_end, expected):
"Test the truncate_string function with various inputs and parameters."
# from tests.test_utils import test_test_truncate_string_keep_end
def test_test_truncate_string_keep_end(text, max_length, keep_end, prefix_len, expected_full):
"Test the specific behavior of the keep_end parameter."
# from tests.test_utils import test_maybe_fenced_code
def test_maybe_fenced_code(content: str, expected_fenced: bool):
# from tests.test_utils import test_backtick_count_adjustment
def test_backtick_count_adjustment(content: str, backtick_count: int):guess what this code does
This code listing appears to be an index or dump of Python functions, classes, and methods primarily belonging to a codebase related to large language models (LLMs). It covers a broad functionality set related to managing LLMs, embeddings, templates, plugins, logging, and command-line interface (CLI) utilities for interaction with language models.
Key features and purposes evident from the code listing:
-
LLM Model and Embeddings Management:
- Classes like
Model,AsyncModel,KeyModel,EmbeddingModel, and associated async variants represent interfaces and implementations for language models and embedding models. - Functions like
register_models,register_embedding_modelsallow plugins or modules to register new model implementations. - Facilities for managing model aliases, fetching models (sync or async), and default model selection.
- Classes like
-
CLI Commands and Utilities:
- Functions under
llm.cliprovide a command-line interface to interact with the language models, enabling prompt execution (prompt), chat conversations (chat), key management (keys_*), logs overview, fragments and templates management, model listing and management, embedding stores, and collections. - Support for advanced CLI features like schema validation, JSON output, streaming responses, and usage statistics.
- Functions under
-
Logging and Database Migrations:
- Various migration functions (
m001_create_tables,m002_foreign_key, etc.) for upgrading and maintaining database schemas used in logging prompts, responses, fragments, embeddings, and collections. - Log querying and formatted output for usage and prompt/response exploration.
- Various migration functions (
-
Templates and Fragment Management:
- Facilities to define prompt templates with variable substitution.
- Support for reusable prompt fragments which can be stored, resolved, and included in prompts.
-
Embedding Collections and Similarity Search:
CollectionandEntryclasses to manage embedding vectors, store them in a database, and retrieve similar items by vector or content.- Commands to embed multiple texts, find similar items, and manage embedding models.
-
Error Handling and Utility Functions:
- Exception classes like
ModelError,NeedsKeyException. - Utilities for encoding, decoding, calculating cosine similarity, extracting fenced code blocks from text.
- Helpers for attachments, content type detection, string truncation.
- Exception classes like
-
Plugin and Hook System:
- Functions for loading plugins, registering commands, and template/fragment loaders.
-
Testing Infrastructure:
- Test functions and mock classes for various components ensuring correctness and facilitating development.
This codebase forms the backend logic and CLI tooling of a framework that enables users to access multiple Large Language Models and embedding models via a unified interface. It supports extensibility via plugins, manages API keys and model aliases, handles prompt templates and fragments, performs embeddings storage and similarity search, facilitates chat and prompt sessions, logs all queries and responses for auditing, and provides comprehensive command-line tools.
The name hints like llm, references to OpenAI models and embedding models, key management, and CLI commands like llm 'some prompt', suggest this is a rich, extensible open-source tool or framework built around Large Language Models.
- The
promptfunction inllm.cliexecutes a prompt against a model, supporting options like streaming, attachments (images, audio), schema validation, system prompts, continuation of conversations, and extraction of code blocks from responses. - The
Collectionclass manages a named collection of embedding vectors stored in a SQLite database, supporting adding new embeddings, bulk embedding with batching, querying similar embeddings, and deleting collections.
If you want, I can explain some specific parts or describe how to use the CLI commands provided.
13,471 input, 781 output