Workflows are not Generally Available yet. Please contact us for early access.
Tensorlake Workflows enable orchestrating data ingestion and transformation. They are great for building long running or compute intensive data workflows which you wouldn’t normally build inside your application within the request-response lifecycle. You can go from code to production in a few minutes, and not spend time managing any infrastructure or pay for resources when you are not using them. Workflows are indefinitely durable by default, so failures in the middle won’t require re-running previous functions during retries.

Setup

  1. The Tensorlake package provides the SDK to build graphs and a CLI to deploy them.
pip install tensorlake
  1. Get an API key from the Tensorlake Dashboard.
export TENSORLAKE_API_KEY=<API_KEY>
Test wether the API key is working:
tensorlake auth status

Your First Workflow

Workflows are defined by writing Python functions and their data-flow is defined by connecting them as graphs. This allows parallel execution of disjoint parts of the workflows, map-reduce, dynamic routing, etc.
greeter_graph.py
from tensorlake.functions_sdk import Graph, tensorlake_function

@tensorlake_function(input_encoder="json")
def say_hello(name: str) -> str:
    return f"Hello, {name}!"

@tensorlake_function(output_encoder="json")
def say_goodbye(greeting: str) -> str:
    name = greeting.split(" ")[1]
    return f"Goodbye, {name}!"

graph = Graph(name="greeter", start_node=say_hello)
graph.add_edge(say_hello, say_goodbye)

Deploy a Graph

tensorlake deploy greeter_graph.py
That’s it! Your workflow is now deployed on Tensorlake Cloud and available at https://api.tensorlake.ai/v1/workflows/greeter.

Invoke a Graph

Once you have deployed your graph, it’s available as an HTTP endpoint. You can either invoke it via HTTP or use the Python SDK. Making a request to your graph will return a Request ID. You can use this request ID to get the output of the graph.
curl -X POST https://api.tensorlake.ai/workflows/greeter \
-H "Authorization: Bearer "$TENSORLAKE_API_KEY"" \
-H "Content-Type: application/json" \
-H "Accept: application/json" \
-d '{"name": "John"}'

# {"id":"beae8736ece31ef9"
You can then use the request ID to get the status of the request.
curl -X GET https://api.tensorlake.ai/workflows/greeter/requests/{request_id} \
-H "Authorization: Bearer "$TENSORLAKE_API_KEY"" \
If the graph is still running, you will get a response back with the request metadata with a status of “running”. At some point, the status will be changed to “complete”.

Get the Output of a Graph

Once the graph is complete, you can get the output of the graph of any function in your workflow. This is useful because often the intermediate stages of a workflow might have useful data that you want to use in your application, or you might want them for debugging purposes.
curl -X GET https://api.tensorlake.ai/workflows/greeter/requests/{request_id}/output/say_goodbye \
-H "Authorization: Bearer "$TENSORLAKE_API_KEY"" \
-H "Accept: application/json"
If the function hasn’t produced any output, you will get an empty response with HTTP status code of 204. The SDK returns a None value in this case. Generally, you would want to check if the request is complete before getting the output.

Test a Graph Locally

These Graphs can run locally on your laptop, so you can test them before deploying them to the cloud.
greeter_graph.py
if __name__ == "__main__":
    invocation_id = graph.run(name="John")
    output = graph.get_output(invocation_id, "say_goodbye")
    print(output)

Include Python Dependencies and Secrets

Most real world workflows will require python packages and access to secrets. For example, if you are building a workflow to extract personal information from Driving licenses using OpenAI’s structured output - you will need the OpenAI package, and your OpenAI API key.
structured_extraction.py
import os
import base64

import requests
from pydantic import BaseModel
from tensorlake.functions_sdk import Graph, tensorlake_function, Image, RequestException

image = Image().run("pip install openai pydantic requests")

class DrivingLicense(BaseModel):
    name: str
    date_of_birth: str
    address: str
    license_number: str
    license_expiration_date: str

@tensorlake_function(
    input_encoder="json", image=image, output_encoder="json", secrets=["OPENAI_API_KEY"]
)
def extract_driving_license_data(url: str) -> DrivingLicense:
    from openai import OpenAI

    # Download image from URL
    response = requests.get(url)
    try:
        response.raise_for_status()
    except Exception as e:
        raise RequestException(f"Failed to download image from URL: {url}: {e}")

    # Encode image as base64
    image_base64 = base64.b64encode(response.content).decode("utf-8")

    # Determine image format from content type or URL
    content_type = response.headers.get("content-type", "")
    if "jpeg" in content_type or "jpg" in content_type:
        image_format = "jpeg"
    elif "png" in content_type:
        image_format = "png"
    else:
        # Default to jpeg if can't determine
        image_format = "jpeg"

    openai = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

    response = openai.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "Extract the personal information from the driving license image.",
            },
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/{image_format};base64,{image_base64}"
                        },
                    }
                ],
            },
        ],
        response_format=DrivingLicense,
    )
    dl = response.choices[0].message.parsed
    return dl.model_dump()


graph = Graph(name="driving_license_extractor", start_node=extract_driving_license_data)
This function uses a custom image with some additional python packages, and it uses the OPENAI_API_KEY secret to authenticate with OpenAI. Building custom images allows you to install pretty much anything you want in your function. The Image api allows running any commands, or other standard commands Docker allows. The secrets argument is used to specify the secrets that are available to the functions. These secrets have to be specified before the functions are deployed. So lets deploy the secrets first, get your OpenAI API key and then run the following command:
tensorlake secrets set OPENAI_API_KEY=<YOUR_OPENAI_API_KEY>
Now you can deploy the graph.
tensorlake deploy structured_extraction.py
You should see the tensorlake stream build logs as your image is being built. Once the image is built, you can invoke the graph as before.
curl -N -X POST https://api.tensorlake.ai/workflows/driving_license_extractor \
-H "Authorization: Bearer "$TENSORLAKE_API_KEY"" \
-H "Content-Type: application/json" \
-H "Accept: application/json" \
-d '{"url": "https://tlake.link/dl"}'
You can build very complex and massively scalable near-real time data workflows with Tensorlake Workflows. Here are some of the next things to learn about: