Flowcept

https://flowcept.readthedocs.io/en/latest/
Admin
Flowcept is a lightweight framework for capturing, integrating, and querying distributed workflow...
Tokens:66,775
Snippets:534
Trust Score:9.6
Update:1 month ago
Show doc for...
Tokens:
Raw
### Install Redis on macOS via Homebrew

Source: https://flowcept.readthedocs.io/en/latest/_sources/setup.rst.txt

Installs Redis and starts it as a service on macOS using Homebrew.

```bash
brew install redis
brew services start redis
```

--------------------------------

### Full runnable example: versioned single-layer perceptron with Flowcept

Source: https://flowcept.readthedocs.io/en/latest/_sources/blob_versioned_single_layer_perceptron_example.rst.txt

Complete Python script demonstrating Flowcept's versioned dataset and model checkpoint saving, PyTorch checkpoint handling, and provenance report generation. Includes a deterministic setup function, a single-layer perceptron model, a shape args handler for serializable provenance, dataset generation with save_or_update_dataset, and a training/validation loop with checkpoint saving.

```python
import pickle
import random

import torch
import torch.nn as nn
import torch.optim as optim

from flowcept import Flowcept
from flowcept.instrumentation.flowcept_task import flowcept_task, get_current_context_task_id


def set_reproducibility(seed: int):
    """Simple deterministic setup. This is optional. Not required by Flowcept."""
    random.seed(seed)
    torch.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed_all(seed)
    if hasattr(torch, "use_deterministic_algorithms"):
        torch.use_deterministic_algorithms(True, warn_only=True)
    if hasattr(torch.backends, "cudnn"):
        torch.backends.cudnn.deterministic = True
        torch.backends.cudnn.benchmark = False
    return {
        "seed": seed,
        "torch_deterministic_algorithms": True,
        "torch_cudnn_deterministic": bool(getattr(torch.backends.cudnn, "deterministic", False))
        if hasattr(torch.backends, "cudnn")
        else False,
        "torch_cudnn_benchmark": bool(getattr(torch.backends.cudnn, "benchmark", False))
        if hasattr(torch.backends, "cudnn")
        else False,
    }


class SingleLayerPerceptron(nn.Module):
    def __init__(self, input_size=2):
        super().__init__()
        self.layer = nn.Linear(input_size, 1)

    def forward(self, x):
        return torch.sigmoid(self.layer(x))


def shape_args_handler(*args, **kwargs):
    """Store tensor arguments as shape metadata (for readable/serializable provenance)."""
    def _shape_key(name):
        return name if name.endswith("_shape") else f"{name}_shape"

    handled = {}
    for i, arg in enumerate(args):
        key = f"arg_{i}"
        if isinstance(arg, torch.Tensor):
            handled[_shape_key(key)] = tuple(arg.shape)
        else:
            handled[key] = arg
    for key, value in kwargs.items():
        if isinstance(value, torch.Tensor):
            handled[_shape_key(key)] = tuple(value.shape)
        else:
            handled[key] = value
    return handled


@flowcept_task(
    args_handler=shape_args_handler,
    output_names=["x_train_shape", "y_train_shape", "x_val_shape", "y_val_shape"],
)
def get_dataset(n_samples, split_ratio, reproducibility):
    """Generate synthetic binary-classification data and save dataset blob."""
    generator = torch.Generator().manual_seed(reproducibility["seed"])
    x = torch.cat(
        [
            torch.randn(n_samples // 2, 2, generator=generator) + 2,
            torch.randn(n_samples // 2, 2, generator=generator) - 2,
        ]
    )
    y = torch.cat([torch.zeros(n_samples // 2), torch.ones(n_samples // 2)]).unsqueeze(1)
    n_train = int(n_samples * split_ratio)
    x_train, x_val = x[:n_train], x[n_train:]
    y_train, y_val = y[:n_train], y[n_train:]

    Flowcept.db.save_or_update_dataset(
        object={"x_train": x_train, "y_train": y_train, "x_val": x_val, "y_val": y_val},
        task_id=get_current_context_task_id(),
        custom_metadata={"n_samples": n_samples, "split_ratio": split_ratio, **reproducibility},
        save_data_in_collection=True,
        pickle=True,
        control_version=True,
    )
    return x_train, y_train, x_val, y_val


def validate(model, criterion, x_val, y_val):
    model.eval()
    with torch.no_grad():
        outputs = model(x_val)
        loss = criterion(outputs, y_val)
        predictions = outputs.round()
        accuracy = (predictions.eq(y_val).sum().item()) / y_val.size(0)
    return loss.item(), accuracy


@flowcept_task(args_handler=shape_args_handler)
def train_and_validate(n_input_neurons, epochs, x_train, y_train, x_val, y_val, checkpoint_check=2):
    model = SingleLayerPerceptron(input_size=n_input_neurons)
    criterion = nn.BCELoss()
    optimizer = optim.SGD(model.parameters(), lr=0.1)

    best_val_loss = float("inf")
    ml_model_object_id = None
    torch_model_object_id = None
    current_task_id = get_current_context_task_id()

    for epoch in range(1, epochs + 1):
        model.train()
        outputs = model(x_train)
        loss = criterion(outputs, y_train)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

        if epoch % checkpoint_check == 0:
            val_loss, val_accuracy = validate(model, criterion, x_val, y_val)
            if val_loss < best_val_loss:
                best_val_loss = val_loss
                ml_model_object_id = Flowcept.db.save_or_update_ml_model(
                    model=model,
                    task_id=current_task_id,
                    custom_metadata={"epoch": epoch, "val_loss": val_loss, "val_accuracy": val_accuracy},
                    save_data_in_collection=True,
                    pickle=True,
                    control_version=True,
                )
                torch_model_object_id = Flowcept.db.save_or_update_torch_model(
                    model=model,
                    task_id=current_task_id,
                    custom_metadata={"epoch": epoch, "val_loss": val_loss, "val_accuracy": val_accuracy},
                    save_data_in_collection=True,
                    pickle=True,
                    control_version=True,
                )

    return ml_model_object_id, torch_model_object_id


if __name__ == "__main__":
    reproducibility = set_reproducibility(seed=42)
    x_train, y_train, x_val, y_val = get_dataset(
        n_samples=1000, split_ratio=0.8, reproducibility=reproducibility
    )
    ml_model_object_id, torch_model_object_id = train_and_validate(
        n_input_neurons=2, epochs=10, x_train=x_train, y_train=y_train, x_val=x_val, y_val=y_val
    )

    # Load the torch model checkpoint
    loaded_torch_model = Flowcept.db.load_torch_model(torch_model_object_id)

    # Generate provenance reports
    Flowcept.db.generate_workflow_provenance_report(
        report_type="markdown",
        output_file="workflow_provenance.md",
    )
    Flowcept.db.generate_workflow_provenance_report(
        report_type="pdf",
        output_file="workflow_provenance.pdf",
    )

```

--------------------------------

### Install Flowcept from source

Source: https://flowcept.readthedocs.io/en/latest/_sources/setup.rst.txt

Clones the Flowcept repository and installs it from source. After this, you can install specific dependencies similarly as with PyPI.

```bash
git clone https://github.com/ORNL/flowcept.git
cd flowcept
pip install .
```

--------------------------------

### flowcept.cli.start_services()

Source: https://flowcept.readthedocs.io/en/latest/cli-reference.html

Starts the Flowcept services. The parameters control which services are started.

```APIDOC
## flowcept.cli.start_services()

### Description
Starts the Flowcept services. The parameters control which services are started.

### Method
CLI function call

### Parameters
- **agent** (bool, optional) - Start the Flowcept FastMCP agent service.
- **webservice** (bool, optional) - Start the Flowcept FastAPI webservice.
- **ui** (bool, optional) - Start the React frontend service. This also starts the FastAPI webservice.
- **agent_gui** (bool, optional) - Start the legacy Streamlit agent GUI service.
- **webservice_host** (str, optional) - Host interface for the webservice. Defaults to settings.yaml `web_server.host`.
- **webservice_port** (str, optional) - Port for the webservice. Defaults to settings.yaml `web_server.port`.
- **ui_dir** (str, optional) - Path to the UI directory containing package.json (default: ui).
- **port** (int, optional) - Port for the legacy agent GUI.

### Returns
None
```

--------------------------------

### Start the Flowcept UI

Source: https://flowcept.readthedocs.io/en/latest/_sources/web_ui.rst.txt

Starts the React frontend and FastAPI webservice. Kills previous processes, starts the webservice in the background, and runs the frontend in the foreground with hot-reload. Press Ctrl+C to stop both.

```bash
flowcept --start --ui
```

--------------------------------

### Install Flowcept with common runtime bundle

Source: https://flowcept.readthedocs.io/en/latest/_sources/setup.rst.txt

Installs the 'extras' group, a convenience shortcut bundling the most common runtime dependencies. Intended for users who want a fairly complete environment, but not maximal. If you only need one feature, install it individually.

```bash
pip install flowcept[extras]
```

--------------------------------

### Install specific dependencies from source

Source: https://flowcept.readthedocs.io/en/latest/_sources/setup.rst.txt

Installs a specific optional dependency from the source repository, following the same pattern as the PyPI extras.

```bash
pip install .[optional_dependency_name]
```

--------------------------------

### Start Flowcept services: flowcept --start

Source: https://flowcept.readthedocs.io/en/latest/_sources/cli-reference.rst.txt

Starts one or more Flowcept services. --webservice starts the FastAPI service, --agent starts the FastMCP agent service, and --ui starts the React frontend and its dependent FastAPI webservice.

```shell
flowcept --start --webservice
flowcept --start --webservice --agent
flowcept --start --ui
flowcept --start --agent
flowcept --start --agent --webservice --ui
```

--------------------------------

### Install all optional dependencies

Source: https://flowcept.readthedocs.io/en/latest/_sources/setup.rst.txt

Installs all optional dependencies via the 'all' extra. Not recommended for users; intended only for Flowcept developers who need to test across all adapters and integrations.

```bash
pip install flowcept[all]
```

--------------------------------

### Install Flowcept with webservice extra

Source: https://flowcept.readthedocs.io/en/latest/_sources/web_ui.rst.txt

Installs Flowcept with the webservice extra to enable the web UI and FastAPI service.

```bash
pip install flowcept[webservice]
```

--------------------------------

### Install Flowcept with specific extras

Source: https://flowcept.readthedocs.io/en/latest/_sources/setup.rst.txt

Install Flowcept with individual optional dependencies. Cherry-pick only the extras relevant to your workflow.

```bash
pip install flowcept[mongo]           # MongoDB support
pip install flowcept[mlflow]          # MLflow adapter
pip install flowcept[dask]            # Dask adapter
pip install flowcept[tensorboard]     # TensorBoard adapter
pip install flowcept[rabbitmq]        # RabbitMQ message queue
pip install flowcept[kafka]           # Kafka message queue
pip install flowcept[nvidia]          # NVIDIA GPU runtime capture
pip install flowcept[telemetry]       # CPU/GPU/memory telemetry capture
pip install flowcept[lmdb]            # LMDB lightweight database
pip install flowcept[mqtt]            # MQTT support
pip install flowcept[llm_agent]       # MCP agent, LangChain, Streamlit integration
pip install flowcept[llm_google]      # Google GenAI + Flowcept agent support
pip install flowcept[llm_agent_audio] # MCP agent with audio enabled (tts).
pip install flowcept[dev]             # Developer dependencies (docs, tests, lint, etc.)
```

--------------------------------

### Start Redis and MongoDB with Docker Compose

Source: https://flowcept.readthedocs.io/en/latest/_sources/web_ui.rst.txt

Starts the required Redis and MongoDB services using the make target. MongoDB is recommended, Redis is required.

```bash
make services-mongo   # Redis + MongoDB
```

--------------------------------

### Install flowcept with PDF report support

Source: https://flowcept.readthedocs.io/en/latest/reporting.html

Install the optional dependency for PDF report generation using pip.

```bash
pip install flowcept[report_pdf]

```

--------------------------------

### Start MongoDB and Redis with Docker Compose

Source: https://flowcept.readthedocs.io/en/latest/web_ui.html

Starts Redis and MongoDB using Docker Compose, required for the instrumentation message queue and most UI features.

```bash
make services-mongo   # Redis + MongoDB

```

--------------------------------

### Install Flowcept with basic dependencies

Source: https://flowcept.readthedocs.io/en/latest/_sources/setup.rst.txt

Installs the minimal Flowcept package from PyPI, which does not include MongoDB, Redis, MCP, or adapter-specific dependencies.

```bash
pip install flowcept
```

--------------------------------

### Minimal PyTorch instrumentation example

Source: https://flowcept.readthedocs.io/en/latest/_sources/telemetry_capture.rst.txt

Minimal Python example to instrument a PyTorch model with Flowcept. Imports torch, torch.nn, and Flowcept's torch instrumentation.

```python
import torch
import torch.nn as nn
from flowcept import Flowcept
from flowcept.instrumentation.flowcept_torch import flowcept_torch
```

--------------------------------

### Lineage highlighting example queries

Source: https://flowcept.readthedocs.io/en/latest/agent.html

Example queries to ask the web chat agent to highlight provenance lineage. The agent resolves matching tasks via a Mongo-style filter and dims unrelated nodes in the Dataflow graph.

```text
"highlight the lineage of the slowest task"
"show me which tasks produced outputs that were later used by failed tasks"
"highlight the lineage of tasks where status is FINISHED and generated.accuracy exists"

```

--------------------------------

### Install Flowcept with webservice, llm_agent, and model provider extras

Source: https://flowcept.readthedocs.io/en/latest/_sources/web_ui.rst.txt

Installs Flowcept with the webservice extra plus llm_agent and a model provider extra for the LLM chat feature.

```bash
pip install flowcept[webservice,llm_agent]
```

--------------------------------

### Recommended settings initialization and profile

Source: https://flowcept.readthedocs.io/en/latest/_sources/setup.rst.txt

Creates a full settings file and applies the full-online profile.

```bash
flowcept --init-settings --full -y
flowcept --config-profile full-online -y
```

--------------------------------

### Get buffer as list of dictionaries

Source: https://flowcept.readthedocs.io/en/latest/api-reference.html

Retrieves the in-memory message buffer as a list of dictionaries. The example checks that the result is a list.

```python
>>> buf = flowcept.get_buffer()
>>> isinstance(buf, list)
True
```

--------------------------------

### Initialize full settings file

Source: https://flowcept.readthedocs.io/en/latest/setup.html

Creates a full settings file with all options at ~/.flowcept/settings.yaml.

```bash
flowcept --init-settings --full

```

--------------------------------

### Get buffer as pandas DataFrame

Source: https://flowcept.readthedocs.io/en/latest/api-reference.html

Retrieves the buffer as a normalized pandas DataFrame with dotted notation for nested keys. The example checks for a specific column.

```python
>>> df = flowcept.get_buffer(return_df=True)
>>> "generated.attention" in df.columns
True
```

--------------------------------

### Initialize settings: flowcept --init-settings

Source: https://flowcept.readthedocs.io/en/latest/_sources/cli-reference.rst.txt

Creates a settings file. Use --full to copy the sample settings file, and --config-profile to modify the existing file in place.

```shell
flowcept --init-settings
flowcept --init-settings --full
flowcept --config-profile full-online
```

--------------------------------

### Initialize minimal settings file

Source: https://flowcept.readthedocs.io/en/latest/_sources/setup.rst.txt

Creates ~/.flowcept/settings.yaml with minimal settings.

```bash
flowcept --init-settings
```

--------------------------------

### Distributed Task Script (distributed_tasks.py)

Source: https://flowcept.readthedocs.io/en/latest/_sources/prov_capture.rst.txt

This script is a task producer in a distributed setup. It reads the workflow ID from the command line, initializes Flowcept with persistence disabled, starts it, executes a task, and stops. It must be run after the persistence starter has created the workflow ID.

```python
import sys
from flowcept import Flowcept, flowcept_task
workflow_id = sys.argv[1]
@flowcept_task
def double(x):
    return 2 * x
# Initialize Flowcept
flowcept = Flowcept(
    workflow_id=workflow_id,
    start_persistence=False,
    check_safe_stops=False,
    save_workflows=False,
)
# This will only establish connection to the MQ, not to the DB.
flowcept.start()
# Example task
y = double(21)
print(f"Result: {y}")
flowcept.stop()
```

--------------------------------

### Query tasks and workflows with Flowcept.db

Source: https://flowcept.readthedocs.io/en/latest/prov_query.html

Typical usage pattern for querying tasks and workflows using the Flowcept.db API. It shows how to get tasks from the current workflow, query tasks by tag, and fetch a workflow object by ID. The example prints task details and workflow arguments.

```python
from flowcept import Flowcept

# query tasks for the current workflow
tasks = Flowcept.db.get_tasks_from_current_workflow()
print(f"Tasks captured in current workflow: {len(tasks)}")

# find all tasks marked with a "math" tag
math_tasks = Flowcept.db.task_query(filter={"tags": "math"})
for t in math_tasks:
    print(f"{t['task_id']} – {t['activity_id']}: {t['status']}")

# fetch a workflow object and inspect its arguments
wf = Flowcept.db.get_workflow_object(workflow_id="123e4567-e89b-12d3-a456-426614174000")
print(wf.workflow_args)
```

--------------------------------

### Minimal Flowcept instrumentation example (quickstart.py)

Source: https://flowcept.readthedocs.io/en/latest/_sources/quick_start.rst.txt

Save as quickstart.py and run with 'python quickstart.py'. Uses decorators @flowcept_task and @flowcept to capture provenance without a database or broker. After running, it generates a report and writes provenance to flowcept_buffer.jsonl.

```python
"""
Minimal example of Flowcept's instrumentation using decorators.
No DB, broker, or external service required.
"""
from flowcept import Flowcept, flowcept_task
from flowcept.instrumentation.flowcept_decorator import flowcept


@flowcept_task(output_names="o1")
def sum_one(i1):
    return i1 + 1


@flowcept_task(output_names="o2")
def mult_two(o1):
    return o1 * 2


@flowcept
def main():
    n = 3
    o1 = sum_one(n)
    o2 = mult_two(o1)
    print("Final output", o2)


if __name__ == "__main__":
    main()

    # print(Flowcept.read_buffer_file())  # inspect raw JSONL records if needed
    Flowcept.generate_report(print_markdown=True)
```

--------------------------------

### Recommended initialization and profile pattern

Source: https://flowcept.readthedocs.io/en/latest/setup.html

Recommended pattern to initialize full settings and apply the full-online profile. The -y flag auto-accepts prompts.

```bash
flowcept --init-settings --full -y
flowcept --config-profile full-online -y

```

--------------------------------

### flowcept.cli.start_redis()

Source: https://flowcept.readthedocs.io/en/latest/cli-reference.html

Start a Redis server using paths configured in settings. Looks up: mq: * bin : str (required) path to the redis-server executable * conf_file : str, optional (appended as the sole argument). Builds and runs the command via _run_command(cmd, check_output=True).

```APIDOC
## flowcept.cli.start_redis() → None

### Description
Start a Redis server using paths configured in settings. Looks up: mq: * bin : str (required) path to the redis-server executable * conf_file : str, optional (appended as the sole argument). Builds and runs the command via _run_command(cmd, check_output=True).

### Method
CLI function call

### Parameters
None

### Returns
None
```

--------------------------------

### flowcept.cli.start_mongo()

Source: https://flowcept.readthedocs.io/en/latest/cli-reference.html

Start a MongoDB server using paths configured in the settings file. Looks up: databases: mongodb: * bin : str (required) path to the mongod executable * db_path: str, required path to the db data directory * log_path : str, optional (adds –fork –logpath) * lock_file_path : str, optional (adds –pidfilepath). Builds and runs the startup command.

```APIDOC
## flowcept.cli.start_mongo() → None

### Description
Start a MongoDB server using paths configured in the settings file. Looks up: databases: mongodb: * bin : str (required) path to the mongod executable * db_path: str, required path to the db data directory * log_path : str, optional (adds –fork –logpath) * lock_file_path : str, optional (adds –pidfilepath). Builds and runs the startup command.

### Method
CLI function call

### Parameters
None

### Returns
None
```

--------------------------------

### Generate full settings with flowcept --init-settings --full

Source: https://flowcept.readthedocs.io/en/latest/_sources/quick_start.rst.txt

Generate a full configuration file for online services, adapters, telemetry, or deployment profiles. This is optional for the simplest offline path.

```bash
flowcept --init-settings --full
```

--------------------------------

### Show CLI help: flowcept --help

Source: https://flowcept.readthedocs.io/en/latest/_sources/cli-reference.rst.txt

Displays all available commands with their helper descriptions and arguments.

```shell
flowcept --help
```

--------------------------------

### Switch CLI profiles with flowcept commands

Source: https://flowcept.readthedocs.io/en/latest/_sources/default_user_guide.rst.txt

Initialize settings and switch between common runtime profiles using flowcept CLI. Prints which keys will change, asks for confirmation, writes to FLOWCEPT_SETTINGS_PATH or ~/.flowcept/settings.yaml. Use --init-settings to create a file, --config-profile to change mode, and adapter flags to add sections.

```shell
flowcept --init-settings --full
flowcept --config-profile full-online
flowcept --config-profile full-telemetry
flowcept --config-profile mq-only
flowcept --config-profile full-offline
flowcept --config-profile mq-only-no-flush
```
Tokens:
Raw