steamship.plugin package#

Subpackages#

Submodules#

steamship.plugin.capabilities module#

Plugin capabilities.

Capabilities are a concept communicated back and forth between plugins and client code via blocks, and they are meant to indicate client code’s need for certain levels of support of a range of features.

Clients can request NATIVE, BEST_EFFORT, or OPTIONAL support for features that a plugin may or may not support. Plugins are expected to parse this and fail-fast if the user has requested support for a feature that the plugin does not support, so that users are not e.g. billed for usage they can’t incorporate.

Capability requests can include other information on the request itself, but oftentimes indicate that certain blocks will be tagged in Steamship-native ways as part of the rest of the payload. For example, ConversationSupport is a capability that indicates the CHAT TagKind will be included in blocks that are part of the input, and the plugin is expected to incorporate these with a model that supports them.

In the case that a Plugin does not support behavior indicated by the Capability request, it will throw, listing the models that it could not support at the levels requested. Otherwise, when Plugins respond, they’ll include another block indicating at which level they served the requested capabilities.

class steamship.plugin.capabilities.Capability(*, name: str = None, request_level: RequestLevel = RequestLevel.NATIVE, **extra_data: Any)[source]#

Bases: BaseModel

Base class for all capabilities.

class Config[source]#

Bases: object

extra = 'allow'#

NAME: ClassVar[str]#

Name of the capability.

Each capability provides its own name. When capabilities are deserialized, they become this base Capability class, with extra fields appended to them. This is not included in the init path for a class by default, with the intention that classes define their names at a class-definition-level.

Make an effort to namespace these by organization, since this is technically extensible.

class Response(*, name: str = None, fulfilled_at: SupportLevel, **extra_data: Any)[source]#

Bases: BaseModel

Response regarding a specific capability served by the plugin.

Responses indicate the level at which they served the request-level.

class Config[source]#

Bases: object

extra = 'allow'#

fulfilled_at: SupportLevel#

name: str#

is_plugin_support_valid(support_level: SupportLevel | None) → Tuple[bool, Response | None][source]#

Checks if the plugin fulfills the capability request level for this specific capability.

Returns a basic Response if the plugin supports the capability and the support level is at least what the consumer asked for.

Parameters:: support_level – The level at which the Plugin asserts that it can support the requested capability
Returns:: a bool that indicates whether the Plugin support level is valid, plus an Optional Response for this capability’s support, if applicable.

name: str#

Name of the capability as an instance field.

The base class of capabilities doesn’t have a NAME, but can represent the unresolved Capabilities that clients are requesting. See CapabilityImpl for the setting of this field.

request_level: RequestLevel#

class steamship.plugin.capabilities.CapabilityImpl(*, name: str = None, request_level: RequestLevel = RequestLevel.NATIVE)[source]#

Bases: Capability

class Config[source]#

Bases: object

extra = 'forbid'#

class steamship.plugin.capabilities.CapabilityPluginRequest(*, requested_capabilities: List[Capability])[source]#

Bases: BaseModel

Model representing the text in a STEAMSHIP_PLUGIN_CAPABILITIES block when it is used as a request.

create_block(client: Client, file_id: str) → Block[source]#

classmethod from_block(block: Block) → CapabilityPluginRequest[source]#

requested_capabilities: List[Capability]#

to_block() → Block[source]#

class steamship.plugin.capabilities.CapabilityPluginResponse(*, capability_responses: List[Response])[source]#

Bases: BaseModel

Model representing the text in a STEAMSHIP_PLUGIN_CAPABILITIES block when it is used as a response.

capability_responses: List[Response]#

create_block(client: Client, file_id: str) → Block[source]#

classmethod from_block(block: Block) → CapabilityPluginResponse[source]#

to_block() → Block[source]#

class steamship.plugin.capabilities.ConversationSupport(*, name: str = None, request_level: RequestLevel = RequestLevel.NATIVE)[source]#

Bases: CapabilityImpl

This plugin supports conversations.

The content of the conversation will come across in other blocks on the request, using the CHAT TagKind.

NAME: ClassVar[str] = 'steamship.conversation_support'#

Name of the capability.

Make an effort to namespace these by organization, since this is technically extensible.

class steamship.plugin.capabilities.FunctionCallingSupport(*, name: str = None, request_level: RequestLevel = RequestLevel.NATIVE, functions: List[Tool])[source]#

Bases: CapabilityImpl

This plugin supports function calling.

Function definitions come across as a list of Tool objects. If the plugin determines a function should be called, it will return a FunctionCallInvocation block, and then will expect a FunctionCallResult block as part of the following request.

class FunctionCallInvocation(*, tool_name: str, args: Dict[str, int | str | float])[source]#

Bases: BaseModel

Describes a request from a plugin to invoke a function

tool_name specifies the name of a Tool that was provided in FunctionCallingSupport, and args to it will be mapped to their values in the args member. A FunctionCallResult block will be expected as part of the next request.

MIME_TYPE: ClassVar[MimeTypes] = 'application/vnd.steamship-block.function-calling-support-invocation+json'#

args: Dict[str, int | str | float]#: The names of arguments that the plugin is providing for the function call, mapped to their values

create_block(client: Client, file_id: str, tags: List[Tag] | None = None) → Block[source]#

classmethod from_block(block: Block) → FunctionCallInvocation[source]#

to_block() → Block[source]#

tool_name: str#: The name of the tool the plugin is requesting to call

class FunctionCallResult(*, tool_name: str, result: List[str])[source]#

Bases: BaseModel

Describes a result of a function call.

A block with this content will be expected after the Plugin requests a call with FunctionCallInvocation.

MIME_TYPE: ClassVar[MimeTypes] = 'application/vnd.steamship-block.function-calling-support-result+json'#

create_block(client: Client, file_id: str, tags: List[Tag] | None = None) → Block[source]#

classmethod from_block(block: Block) → FunctionCallResult[source]#

result: List[str]#: The result of the tool invocation. Each item in the list can be text, or a Block identifier.

result_items(client: Client) → Iterator[str | Block][source]#

to_block() → Block[source]#

tool_name: str#: The name of the tool for which the result of a function call is being provided

NAME: ClassVar[str] = 'steamship.function_calling_support'#

Name of the capability.

Make an effort to namespace these by organization, since this is technically extensible.

functions: List[Tool]#: A list of Tools which the LLM can choose from to execute.

class steamship.plugin.capabilities.RequestLevel(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Bases: Enum

Specifies how the Plugin should handle the request for support of a specific feature.

BEST_EFFORT = 'best-effort'#

The plugin may choose to make a best effort to support the requested feature. For example, the plugin may craft a prompt to a model in a way that attempts to support the feature and parse the output appropriately.

If the plugin supports the feature at a native level, but the request_level from the user was BEST_EFFORT, the plugin will still report NATIVE. It is not a requirement that plugins do this, and so many plugins may opt to only support their NATIVE levels.

DISABLE = 'disable'#

Actively disable this capability.

In the case that the plugin does not support this capability, no behavior change occurs. In the case that the plugin can’t disable this capability (it’s a core part of the experience), the plugin should

fast-fail.

In the case that the plugin supports but can opt not to use this capability, it won’t be used.

NATIVE = 'native'#: The plugin user is requesting that the plugin supports this feature as a first-class feature.

OPTIONAL = 'optional'#: The plugin may ignore this feature when serving content.

class steamship.plugin.capabilities.RequestedCapabilities(supported_levels: Mapping[Type[CapabilityType], SupportLevel])[source]#

Bases: object

extract_from_blocks(blocks: Iterable[Block]) → CapabilityPluginResponse | None[source]#

Find the block in a list that defines capability requests, and initialize this data structure with it.

It may be the case that there is no block indicating capability requests; Older clients may be passing blocks to the plugin without programmatic knowledge of capabilities.

Parameters:: blocks – A list of blocks that was passed as input to a plugin
Returns:: a CapabilityPluginResponse which indicates the level at which we are serving each request

get(typ: Type[CapabilityType], default: CapabilityType | None = None) → CapabilityType | None[source]#

load_requests(request: CapabilityPluginRequest) → CapabilityPluginResponse[source]#

Load requests from a CapabilityPluginRequest into this mapping and verify that we can support the requested capabilities.

Parameters:: request – The CapabilityPluginRequest provided to the Plugin
Returns:: a CapabilityPluginResponse which indicates the level at which we are serving each request

class steamship.plugin.capabilities.SupportLevel(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Bases: Flag

Flags for how plugins support capabilities.

NATIVE and BEST_EFFORT can be OR-ed together here, but don’t need to be: NATIVE implies at least BEST_EFFORT.

BEST_EFFORT = 2#: The plugin claims to support this capability, in a way that is not native to any model.

CAN_DISABLE = 4#: The plugin offers the ability to disable this behavior.

NATIVE = 1#: The plugin supports this capability natively.

class steamship.plugin.capabilities.SystemPromptSupport(*, name: str = None, request_level: RequestLevel = RequestLevel.BEST_EFFORT)[source]#

Bases: CapabilityImpl

This plugin supports system prompts separate from the per-request model prompt.

The system prompt will come across in other blocks on the request.

NAME: ClassVar[str] = 'steamship.system_prompt_support'#

Name of the capability.

Make an effort to namespace these by organization, since this is technically extensible.

request_level: RequestLevel#: If NATIVE, asserts that the model being used supports System Prompt in a first-class capacity. If BEST_EFFORT, the system prompt will be incorporated into the request in another way, usually via concatenation

with the prompt.

exception steamship.plugin.capabilities.UnsupportedCapabilityError(unsupported_capabilities: List[Capability])[source]#

Bases: Exception

Raised when support for a given feature is requested from a plugin in a non-optional way, but the plugin doesn’t support it

steamship.plugin.embedder module#

class steamship.plugin.embedder.Embedder(client: Steamship = None, config: Dict[str, Any] = None, context: InvocationContext = None)[source]#

Bases: PluginService[BlockAndTagPluginInput, EmbeddedItemsPluginOutput], ABC

abstract run(request: PluginRequest[BlockAndTagPluginInput]) → InvocableResponse[EmbeddedItemsPluginOutput][source]#

Runs the core operation implemented by this plugin: import, export, blockify, tag, etc.

This is the method that a Steamship Plugin implements to perform its main work.

run_endpoint(**kwargs) → InvocableResponse[EmbeddedItemsPluginOutput][source]#: Exposes the Embedder’s run operation to the Steamship Engine via the expected HTTP path POST /tag

steamship.plugin.file_importer module#

class steamship.plugin.file_importer.FileImporter(client: Steamship = None, config: Dict[str, Any] = None, context: InvocationContext = None)[source]#

Bases: PluginService[FileImportPluginInput, RawDataPluginOutput], ABC

abstract run(request: PluginRequest[FileImportPluginInput]) → InvocableResponse[RawDataPluginOutput][source]#

Runs the core operation implemented by this plugin: import, export, blockify, tag, etc.

This is the method that a Steamship Plugin implements to perform its main work.

run_endpoint(**kwargs) → InvocableResponse[RawDataPluginOutput][source]#: Exposes the File Importer’s run operation to the Steamship Engine via the expected HTTP path POST /import

steamship.plugin.generator module#

class steamship.plugin.generator.Generator(client: Steamship = None, config: Dict[str, Any] = None, context: InvocationContext = None)[source]#

Bases: PluginService[RawBlockAndTagPluginInput, RawBlockAndTagPluginOutput], ABC

abstract run(request: PluginRequest[RawBlockAndTagPluginInput]) → InvocableResponse[RawBlockAndTagPluginOutput][source]#

Runs the core operation implemented by this plugin: import, export, blockify, tag, etc.

This is the method that a Steamship Plugin implements to perform its main work.

run_endpoint(**kwargs) → InvocableResponse[RawBlockAndTagPluginOutput][source]#: Exposes the Tagger’s run operation to the Steamship Engine via the expected HTTP path POST /tag

upload_block_content_to_signed_url(block: Block) → Block[source]#: Recreate the block (create request) as a URL request, rather than direct content, since we can’t do a multipart file upload from here.

class steamship.plugin.generator.TrainableGenerator(client: Steamship = None, config: Dict[str, Any] = None, context: InvocationContext = None)[source]#

Bases: TrainablePluginService[RawBlockAndTagPluginInput, RawBlockAndTagPluginOutput], ABC

get_training_parameters_endpoint(**kwargs) → InvocableResponse[TrainingParameterPluginOutput][source]#: Exposes the Service’s get_training_parameters operation to the Steamship Engine via the expected HTTP path POST /getTrainingParameters

run_endpoint(**kwargs) → InvocableResponse[RawBlockAndTagPluginOutput][source]#: Exposes the Tagger’s run operation to the Steamship Engine via the expected HTTP path POST /generate

abstract run_with_model(request: PluginRequest[RawBlockAndTagPluginInput], model: TrainableModel) → InvocableResponse[RawBlockAndTagPluginOutput][source]#: Rather than implementing run(request), a TrainablePluginService implements run_with_model(request, model)

train_endpoint(**kwargs) → InvocableResponse[TrainPluginOutput][source]#: Exposes the Service’s train operation to the Steamship Engine via the expected HTTP path POST /train

steamship.plugin.request module#

class steamship.plugin.request.PluginRequest(*, data: T | None = None, context: PluginRequestContext | None = None, status: Task | None = None, isStatusCheck: bool = False)[source]#

Bases: GenericModel, Generic[T]

class Config[source]#

Bases: object

alias_generator() → str#

allow_population_by_field_name = True#

context: PluginRequestContext | None#

data: T | None#

is_status_check: bool#

status: Task | None#

class steamship.plugin.request.PluginRequestContext(*, pluginId: str = None, pluginHandle: str = None, pluginVersionId: str = None, pluginVersionHandle: str = None, pluginInstanceId: str = None, pluginInstanceHandle: str = None)[source]#

Bases: CamelModel

Contains the context in which

plugin_handle: str#

plugin_id: str#

plugin_instance_handle: str#

plugin_instance_id: str#

plugin_version_handle: str#

plugin_version_id: str#

steamship.plugin.streaming_generator module#

class steamship.plugin.streaming_generator.StreamingGenerator(client: Steamship = None, config: Dict[str, Any] = None, context: InvocationContext = None)[source]#

Bases: PluginService[RawBlockAndTagPluginInputWithPreallocatedBlocks, StreamCompletePluginOutput], ABC

abort_open_block_streams(blocks: List[Block])[source]#

abstract determine_output_block_types(request: PluginRequest[RawBlockAndTagPluginInput]) → InvocableResponse[BlockTypePluginOutput][source]#

determine_output_block_types_endpoint(**kwargs) → InvocableResponse[BlockTypePluginOutput][source]#

abstract run(request: PluginRequest[RawBlockAndTagPluginInputWithPreallocatedBlocks]) → InvocableResponse[StreamCompletePluginOutput][source]#

Runs the core operation implemented by this plugin: import, export, blockify, tag, etc.

This is the method that a Steamship Plugin implements to perform its main work.

run_endpoint(**kwargs) → InvocableResponse[StreamCompletePluginOutput][source]#

steamship.plugin.tagger module#

class steamship.plugin.tagger.Tagger(client: Steamship = None, config: Dict[str, Any] = None, context: InvocationContext = None)[source]#

Bases: PluginService[BlockAndTagPluginInput, BlockAndTagPluginOutput], ABC

abstract run(request: PluginRequest[BlockAndTagPluginInput]) → InvocableResponse[BlockAndTagPluginOutput][source]#

Runs the core operation implemented by this plugin: import, export, blockify, tag, etc.

This is the method that a Steamship Plugin implements to perform its main work.

run_endpoint(**kwargs) → InvocableResponse[BlockAndTagPluginOutput][source]#: Exposes the Tagger’s run operation to the Steamship Engine via the expected HTTP path POST /tag

class steamship.plugin.tagger.TrainableTagger(client: Steamship = None, config: Dict[str, Any] = None, context: InvocationContext = None)[source]#

Bases: TrainablePluginService[BlockAndTagPluginInput, BlockAndTagPluginOutput], ABC

get_training_parameters_endpoint(**kwargs) → InvocableResponse[TrainingParameterPluginOutput][source]#: Exposes the Service’s get_training_parameters operation to the Steamship Engine via the expected HTTP path POST /getTrainingParameters

run_endpoint(**kwargs) → InvocableResponse[BlockAndTagPluginOutput][source]#: Exposes the Tagger’s run operation to the Steamship Engine via the expected HTTP path POST /tag

abstract run_with_model(request: PluginRequest[BlockAndTagPluginInput], model: TrainableModel) → InvocableResponse[BlockAndTagPluginOutput][source]#: Rather than implementing run(request), a TrainablePluginService implements run_with_model(request, model)

train_endpoint(**kwargs) → InvocableResponse[TrainPluginOutput][source]#: Exposes the Service’s train operation to the Steamship Engine via the expected HTTP path POST /train

steamship.plugin.trainable_model module#

class steamship.plugin.trainable_model.TrainableModel[source]#

Bases: ABC, Generic[ConfigType]

Base class for trainable models.

Trainable models are not plugins. They are a thin wrapper around the state of a model designed to be used with the Steamship plugin system.

# State Management

100% of a TrainableModel’s state management should save to & read from a folder on disk via the methods save_to_folder and load_from_folder.

# Remote Saving and Loading

TrainableModel instances automatically save to a user’s Workspace on Steamship via save_remote method. They can load themselves from a user’s workspace via the load_remote method.

When saving a model, the caller provides handle, such as “V1” or “epoch_23”. This allows that particular checkpoint to be re-loaded. By default, every save operation also saves the model to the “default” checkpoint, overwriting it if it already existed. When a user loads a model without specifying a checkpoint, the “default” checkpoint will be used.

# Data Scope

A TrainableModel’s data is saved & loaded with respect to

The user’s active Workspace, and
The provided Plugin Instance within that workspace.

The active workspace is read from the Steamship client context, and the plugin_instance_id is supplied as a method argument on the save_remote and load_remote methods.

This organization enables a user to have arbitrarily many trained model instances of the same type colocated within a Workspace.

# Training

A training job is fully parameterized by the TrainPluginInput object.

# Result Reporting

A training job’s results are reported via the TrainPluginOutput object. These results include a reference to the save_remote output, but they do not include the model parameters themselves. For example, after training, one could write:

>>> archive_path_in_steamship = model.save_remote(..)
>>> output = TrainPluginOutput(archive_path_in_steamship=archive_path_in_steamship,
    ...
    )

That output is the ultimate return object of the training process, but the Plugin that owns this model need not wait for synchronous completion to update the Steamship Engine with intermediate results. It can use the Response.post_update to proactively stream results back to the server.

# Third-party / External Models

This model class is a convenient wrapper for models running on third party systems (e.g. Google’s AutoML). In such a case:

The train method would begin the job on the 3rd party system.
The save_to_folder method would write the Job ID and any other useful data to the checkpoint path
The load_from_folder method would read this Job ID from disk and obtain an authenticated client with the third party system.
Any run method the implementer created would ferry back results fetched from the third-party system.
Any status reporting in TrainPluginOutput would ferry back status fetched from the third-party system.

config: ConfigType = None#

abstract load_from_folder(checkpoint_path: Path)[source]#: Load 100% of the state of this model to the provided path.

classmethod load_from_local_checkpoint(checkpoint: ModelCheckpoint, config: ConfigType)[source]#

classmethod load_remote(client: Client, plugin_instance_id: str, checkpoint_handle: str | None = None, use_cache: bool = True, model_parent_directory: Path = None, plugin_instance_config: ConfigType = None)[source]#

receive_config(config: ConfigType)[source]#: Stores config from plugin instance, so it is accessible by model on load or train.

save_remote(client: Client, plugin_instance_id: str, checkpoint_handle: str | None = None, model_parent_directory: Path = None, set_as_default: bool = True) → str[source]#

abstract save_to_folder(checkpoint_path: Path)[source]#: Saves 100% of the state of this model to the provided path.

abstract train(input: PluginRequest[TrainPluginInput]) → InvocableResponse[TrainPluginOutput][source]#: Train or fine-tune the model, parameterized by the information in the TrainPluginInput object.

abstract train_status(input: PluginRequest[TrainPluginInput]) → InvocableResponse[TrainPluginOutput][source]#: Check on the status of an in-process training job, if it is running externally asynchronously.

steamship.plugin package#

Subpackages#

Submodules#

steamship.plugin.capabilities module#

steamship.plugin.embedder module#

steamship.plugin.file_importer module#

steamship.plugin.generator module#

steamship.plugin.request module#

steamship.plugin.streaming_generator module#

steamship.plugin.tagger module#

steamship.plugin.trainable_model module#

Module contents#