nat.builder.workflow#

Classes#

Workflow

Module Contents#

class Workflow(
*,
config: nat.data_models.config.Config,
entry_fn: nat.builder.function.Function[nat.builder.function_base.InputT, nat.builder.function_base.StreamingOutputT, nat.builder.function_base.SingleOutputT],
functions: dict[str, nat.builder.function.Function] | None = None,
function_groups: dict[str, nat.builder.function.FunctionGroup] | None = None,
llms: dict[str, nat.builder.llm.LLMProviderInfo] | None = None,
embeddings: dict[str, nat.builder.embedder.EmbedderProviderInfo] | None = None,
memory: dict[str, nat.memory.interfaces.MemoryEditor] | None = None,
object_stores: dict[str, nat.object_store.interfaces.ObjectStore] | None = None,
telemetry_exporters: dict[str, nat.observability.exporter.base_exporter.BaseExporter] | None = None,
retrievers: dict[str | None, nat.builder.retriever.RetrieverProviderInfo] | None = None,
ttc_strategies: dict[str, nat.experimental.test_time_compute.models.strategy_base.StrategyBase] | None = None,
context_state: nat.builder.context.ContextState,
)#

Bases: nat.builder.function_base.FunctionBase[nat.builder.function_base.InputT, nat.builder.function_base.StreamingOutputT, nat.builder.function_base.SingleOutputT]

config#
functions#
function_groups#
llms#
embeddings#
memory#
telemetry_exporters#
object_stores#
retrievers#
_exporter_manager#
ttc_strategies#
_entry_fn#
_context_state#
_saved_context#
property has_streaming_output: bool#
property has_single_output: bool#
async get_all_exporters() → dict[str, nat.observability.exporter.base_exporter.BaseExporter]#
property exporter_manager: nat.observability.exporter_manager.ExporterManager#
async run(
message: nat.builder.function_base.InputT,
runtime_type: nat.data_models.runtime_enum.RuntimeTypeEnum = RuntimeTypeEnum.RUN_OR_SERVE,
)#

Called each time we start a new workflow run. We’ll create a new top-level workflow span here.

async result_with_steps(
message: nat.builder.function_base.InputT,
to_type: type | None = None,
)#
static from_entry_fn(
*,
config: nat.data_models.config.Config,
entry_fn: nat.builder.function.Function[nat.builder.function_base.InputT, nat.builder.function_base.StreamingOutputT, nat.builder.function_base.SingleOutputT],
functions: dict[str, nat.builder.function.Function] | None = None,
function_groups: dict[str, nat.builder.function.FunctionGroup] | None = None,
llms: dict[str, nat.builder.llm.LLMProviderInfo] | None = None,
embeddings: dict[str, nat.builder.embedder.EmbedderProviderInfo] | None = None,
memory: dict[str, nat.memory.interfaces.MemoryEditor] | None = None,
object_stores: dict[str, nat.object_store.interfaces.ObjectStore] | None = None,
telemetry_exporters: dict[str, nat.observability.exporter.base_exporter.BaseExporter] | None = None,
retrievers: dict[str | None, nat.builder.retriever.RetrieverProviderInfo] | None = None,
ttc_strategies: dict[str, nat.experimental.test_time_compute.models.strategy_base.StrategyBase] | None = None,
context_state: nat.builder.context.ContextState,
) → Workflow[InputT, StreamingOutputT, SingleOutputT]#