Programming Guide

from tensorlake import tensorlake_function

image = Image()
  .run("pip install transformers")
  .build()

@tensorlake_function(image=image, input_encoder="json")
def my_function(data: str) -> int:
    return len(data)

from tensorlake import tensorlake_function, Image, GraphRequestContext, Retries


@tensorlake_function()
def quadruple(value: int) -> int:
    return value * 4


@tensorlake_function(
    # Use Ubuntu as a base image instead of the default Debian
    image=Image().base_image("ubuntu:latest"),
    # Use cloudpickle to deserialize function inputs
    input_encoding="cloudpickle",
    # Use cloudpickle to serialize function outputs
    output_encoding="cloudpickle",
    # Make my_secret available to the function as an environment variable
    secrets=["my_secret"],
    # Call quadruple with the output of this function
    next=quadruple,
    # Name of the function in the workflow
    name="measure string",
    # Description of the function in the workflow
    description="Measures the string using its length",
    # Retry the function twice if it fails
    retries=Retries(max_retries=2),
    # Function fails if it was running for more than 30 seconds and didn't report any progress
    timeout=30,
    # Pass the request context to the function as its first argument
    use_ctx=True,
    # Reuse previous function outputs if the function gets called with the same inputs
    cacheable=True,
    # 2 CPUs are available to the function
    cpu=2,
    # 4 GB of memory is available to the function
    memory=4,
    # 2 GB of ephemeral /tmp disk space is available to the function
    ephemeral_disk=2,
)
def string_length(ctx: GraphRequestContext, s: str) -> int:
    return len(s)

from large_model import load_large_model

class MyCompute(TensorlakeCompute):
    # The same attributes as for @tensorlake_function() decorator
    image = ...
    input_encoder = ...

    def __init__(self):
        # Run initialization code once on function container startup
        self.model = load_large_model()

    def run(self, data: str) -> int:
        return self.model.run(data)

from tensorlake import tensorlake_function, Retries

# Retry the function once if it failed
@tensorlake_function(retries=Retries(max_retries=1))
def my_function() -> int:
    raise Exception("Something went wrong")

from tensorlake import GraphRequestContext, tensorlake_function

@tensorlake_function(use_ctx=True)
def my_function(ctx: GraphRequestContext, data: str) -> int:
    # Set request progress to 1 out of 100.
    ctx.update_progress(1, 100)
    # Print the request information.
    print(f"Request ID: {ctx.request_id}, Graph name: {ctx.graph_name}, Graph version: {ctx.graph_version}")
    # Set a request key-value pair. It's availble to the function in the same request.
    ctx.request_state.set("my_function_data", data)
    # Set request progress to 100 out of 100.
    ctx.update_progress(100, 100)
    # Return the length of the previously stored string.
    return len(ctx.request_state.get("my_function_data"))

g = Graph(start_node=my_function, name="my_workflow", description="My workflow")
g.add_edge(my_function, my_function1)
g.add_edge(my_function, my_function2)

curl -X POST https://api.tensorlake.com/workflows/my_workflow \
  -H "Content-Type: application/json" \
  -d '{"data": "Hello, world!"}'

from tensorlake import RemoteGraph

g = RemoteGraph(name="my_workflow")

g.outputs(my_function1)

from tensorlake import tensorlake_function, Graph, Retries

# `retries` attribute is not set, uses default workflow retry policy.
@tensorlake_function()
def my_function() -> int:
    raise Exception("Something went wrong")

# Set default workflow retry policy to retry all its functions twice.
# If not set, the default workflow retry policy is to not retry the functions.
g = Graph(start_node=my_function, name="my_workflow", retries=Retries(max_retries=2))