# Releases
Source: https://launchpad.datalumina.com/docs/changelog/updates
Release notes
## What's Changed
### Bug fixes
* **Celery task discovery** now points at `launchpad.worker`, matching the packaged `app/launchpad/` module structure introduced in v3.4.0. Workers can discover tasks correctly after installing the project as a package.
* **Event persistence** now stores the validated event model directly on `Event.data`, avoiding an unnecessary JSON dump step before the repository writes the event.
* **Azure OpenAI provider setup** now uses pydantic-ai's `AzureProvider` with `AZURE_OPENAI_ENDPOINT`, `AZURE_OPENAI_API_KEY`, and `AZURE_OPENAI_API_VERSION` (default `2025-04-01-preview`).
### Provider cleanup
* **Mistral support was removed** from `ModelProvider` and Docker Compose environment forwarding to keep the provider list aligned with the currently supported runtime paths.
* **Google Vertex AI credential loading** now passes an empty string when `GOOGLE_APPLICATION_CREDENTIALS` is unset, making the expected configuration failure clearer.
### Dependencies and developer tooling
* **pydantic-ai** was bumped from `>=1.80` to `>=1.94`.
* **Pyright configuration** now points at the `app` package root and local `.venv`, improving editor/type-checker resolution for `launchpad.*` imports.
## What's Changed
### Project layout
The runtime code that used to live flat under `app/` is now a proper installable Python package: `app/launchpad/`. `pyproject.toml` declares hatchling as the build backend so `uv sync` installs the project in editable mode — no more `sys.path` tweaks in playground scripts, and IDE / pyright pick up the package tree natively.
* **Imports**: every `from core.…`, `from workflows.…`, etc. is now `from launchpad.core.…` / `from launchpad.workflows.…`. Same applies to `api`, `services`, `database`, `worker`, `utils`, `schemas`.
* **.env moved** to the repo root (was `app/.env`). `docker/.env` still lives under `docker/` for compose-level settings.
* **Alembic** stays colocated with the package at `app/launchpad/alembic/`; `./migrate.sh` and `./makemigration.sh` are unchanged but now run from `app/launchpad/`.
* **Docker** uses `WORKDIR /workspace`, installs the package editable via `uv pip install -e .`, and mounts `./app:/workspace/app` for hot reload. The Celery worker's `-A` target is `launchpad.worker.config`.
* **Playground scripts** dropped their `sys.path.append(...)` preamble; fixtures and `.env` now load from their new paths.
### Reference workflows grouped under `examples/`
Every shipped workflow now lives in its own package under `app/launchpad/workflows/examples//` with `schema.py`, `workflow.py`, a `nodes/` directory, a `request_examples/` folder, and a `README.md`. The `examples/` grouping makes it explicit that these are demos to read and replace — your own workflows belong at `app/launchpad/workflows//`. The `WorkflowRegistry` enum exposes five ready-to-run workflows:
* **`STREAMING`** → `ExampleStreamingWorkflow` (renamed from `StreamingExampleWorkflow`), backs `POST /v1/chat/completions`.
* **`QUICKSTART`** → `CustomerCareWorkflow`, concurrent analysis + routing + Jinja2 prompts colocated at `app/launchpad/workflows/examples/quickstart/prompts/`.
* **`LANGFUSE_TRACING`** *(new)* → `LangfuseTracingWorkflow`, a small moderation pipeline used as a tracing demo.
* **`PGVECTOR_RAG`** *(new)* → `RagExampleWorkflow`, two-node retrieval + generation on top of a `PgvectorRAGService` wrapper around `vecs` + OpenAI embeddings.
* **`NESTED_WORKFLOW`** *(new)* → `NestedWorkflow`, a deterministic parent/child workflow composition demo that reuses the same `TaskContext`.
### Core
* **Workflow composition** — `Workflow.run()` and `Workflow.run_async()` now accept an optional `context=` kwarg so a parent workflow can hand its `TaskContext` to a child workflow. The parent's node registry is preserved on return.
* **Nested workflows** — workflows can now call other workflows from inside a node by using `await ChildWorkflow().run_async(context=task_context)`. The new `NESTED_WORKFLOW` example demonstrates a parent workflow delegating reply drafting to a child workflow while reusing the same `TaskContext`.
* **`TaskContext`** gains `should_stop: bool` and a matching `stop_workflow()` helper, plus `trace_id: str | None` captured on entry when tracing is enabled.
* **`Node.cleanup()`** is now explicit on the base class and invoked by the orchestrator after each node run, including when an exception propagates.
* **`enable_tracing` defaults to `False`** — opt in per workflow instance. Invalid Langfuse credentials raise `LangfuseAuthenticationError` at construction time.
* **`AgentConfig.instructions`** replaces `system_prompt`; `AgentNode` instances attach dynamic context with `@self.agent.instructions` instead of `@self.agent.system_prompt`.
### Prompt management
* **Colocated prompts** — `PromptManager.get_prompt(..., prompts_dir=PROMPTS_DIR)` loads templates next to the workflow that owns them.
* **Frontmatter support** — `python-frontmatter` parses YAML metadata at the top of each `.j2` file; surfaced via `PromptManager.get_template_info`.
### Playground
* One script per reference workflow: `playground/{quickstart,nested_workflow,streaming,langfuse_tracing,pgvector_rag}.py`. The legacy `workflow_playground.py` and `send_event.py` scripts have been removed.
### Docker
* Supabase remains excluded by default in `docker-compose.yml`; uncomment `docker-compose.supabase.yml` when you need Studio, Auth, Realtime, Storage, or the Supabase gateway. Caddy remains opt-in for HTTPS deployments.
### Dependencies
* `langfuse>=3.10.5`, `jinja2>=3.1.6`, `python-frontmatter`, `vecs>=0.4`, `tiktoken>=0.12.0`
* Python `==3.13.13`
* `pydantic-ai>=1.80`
## What's Changed
### New Features
* **SSE Streaming with OpenAI-compatible API** - New `/v1/chat/completions` endpoint that streams responses using Server-Sent Events, following the OpenAI API specification
* **Workflow streaming support** - Added `run_stream_async()` method to the `Workflow` class for streaming workflow execution
* **AgentStreamingNode** - New node type for streaming LLM responses with `stream_text_deltas()` and `stream_structured_deltas()` methods
* **StreamingExampleWorkflow** - Working example with `TextStreamingNode` and `StructuredStreamingNode` demonstrating both plain text and structured output streaming
### Langfuse Integration
* **Native SDK integration** - Replaced OpenTelemetry-based tracing with the native Langfuse SDK
* **Configurable tracing** - Enable or disable tracing per workflow with `enable_tracing` parameter: `Workflow(enable_tracing=True)`
* **Automatic span creation** - Workflow and node executions are automatically traced when enabled
### Core Updates
* **Python version requirement** - Bumped minimum Python version to `>=3.13.7`
### Docker Infrastructure
* **Modular compose files** - Split Docker configuration into separate files for easier customization:
* `docker-compose.launchpad.yml` - Core application (api, celery, redis, db)
* `docker-compose.supabase.yml` - Full Supabase stack (studio, auth, realtime, storage, etc.)
* `docker-compose.caddy.yml` - Reverse proxy with automatic HTTPS
* **Updated Supabase images** - All Supabase services updated to latest versions:
* Studio: 2025.11.26-sha-8f096b5
* GoTrue (Auth): v2.183.0
* PostgREST: v13.0.7
* Realtime: v2.65.3
* Storage API: v1.32.0
* Postgres Meta: v0.93.1
* Edge Runtime: v1.69.25
* Logflare (Analytics): 1.26.13
* Supavisor (Connection Pooler): 2.7.4
### Dependencies
* Updated `pydantic-ai` from >=1.0.15 to >=1.26
### Files Changed
* `app/api/openai.py` (new)
* `app/core/nodes/agent_streaming_node.py` (new)
* `app/core/workflow.py`
* `app/workflows/streaming_example_workflow.py` (new)
* `app/workflows/streaming_example_workflow_nodes/` (new)
* `app/utils/event_stream_generator.py` (new)
* `pyproject.toml`
* `docker/docker-compose.yml` (restructured)
* `docker/docker-compose.launchpad.yml` (new)
* `docker/docker-compose.supabase.yml` (new)
* `docker/docker-compose.caddy.yml` (new)
## What's Changed
### Core Improvements
* **Standardized node output management** - Added `save_output()` and `get_output()` methods to handle node outputs in a standardized way using Pydantic models. Users can now access node data by simply providing the node class, eliminating the need to memorize specific keys for storing and retrieving node outputs.
* **Enhanced RouterNode functionality** - Added task context handling and improved output management capabilities
* **Refactored node and workflow logic** - Updated base node and workflow classes to support task context handling across the system
* **Code cleanup** - Removed unused OpenAI model imports from agent module
### Model Provider Updates
* **OpenAI integration** - Migrated from `OpenAIModel` to `OpenAIChatModel` for improved consistency
### Dependencies
* Updated `pydantic-ai` from >=0.7.5 to >=1.0.15
* Updated `alembic` dependency to >=1.16.4
* Updated `uv.lock` with latest package metadata
### Files Changed
* `app/core/nodes/agent.py`
* `app/core/nodes/base.py`
* `app/core/nodes/router.py`
* `app/core/workflow.py`
* `pyproject.toml`
* `uv.lock`
# Agent Node
Source: https://launchpad.datalumina.com/docs/core/agent-node
Integrate LLMs into workflows with AgentNode, backed by pydantic-ai.
The `AgentNode` class wraps a pydantic-ai `Agent` so LLM calls slot into the same Chain-of-Responsibility pattern as any other node. Subclasses implement two methods:
* **`get_agent_config()`** — returns an `AgentConfig` describing the model provider, model name, structured output type, tools, and instructions.
* **`process()`** — runs the underlying `Agent`, validates outputs against `OutputType`, and saves them with `save_output()`.
See [LLM Providers](/docs/tools/llm-providers) for per-provider configuration details.
## AgentConfig
```python theme={null}
@dataclass
class AgentConfig:
model_provider: ModelProvider
model_name: Union[
OpenAIModelName, AnthropicModelName, GeminiModelName, BedrockModelName
]
output_type: Any = str
instructions: Optional[str] = None
deps_type: Optional[Type[Any]] = None
name: str | None = None
model_settings: ModelSettings | None = None
retries: int = 1
output_retries: int | None = None
tools: List = field(default_factory=list)
builtin_tools: List = field(default_factory=list)
instrument: bool = True
```
Notable fields:
* **`instructions`** — the system prompt for the underlying pydantic-ai `Agent`. You can also register per-run context via the `@self.agent.instructions` decorator inside `process()`.
* **`output_type`** — a Pydantic model (subclass of `AgentNode.OutputType`) for structured output, or `str` for plain text.
* **`deps_type`** — a Pydantic model exposed via `RunContext` so tools and instruction callbacks can read dependencies the node computes at runtime.
* **`tools`, `builtin_tools`** — pydantic-ai tool definitions.
* **`instrument`** — defaults to `True`; set `False` to opt this node out of Langfuse instrumentation even when the workflow is traced.
## AgentNode base class
```python theme={null}
class AgentNode(Node, ABC):
class DepsType(BaseModel):
pass
class OutputType(BaseModel):
pass
def __init__(self, task_context: TaskContext = None):
super().__init__(task_context=task_context)
self.__async_client = AsyncClient()
agent_wrapper = self.get_agent_config()
self.agent = Agent(
model=self.__get_model_instance(
agent_wrapper.model_provider, agent_wrapper.model_name
),
output_type=agent_wrapper.output_type,
instructions=agent_wrapper.instructions,
deps_type=agent_wrapper.deps_type,
name=agent_wrapper.name,
model_settings=agent_wrapper.model_settings,
retries=agent_wrapper.retries,
output_retries=agent_wrapper.output_retries,
tools=agent_wrapper.tools,
builtin_tools=agent_wrapper.builtin_tools,
instrument=agent_wrapper.instrument,
)
self.agent.instrument_all()
@abstractmethod
def get_agent_config(self) -> AgentConfig:
pass
@abstractmethod
async def process(self, task_context: TaskContext) -> TaskContext:
pass
```
`AgentNode` runs asynchronously. Call `await self.agent.run(...)` or `async with self.agent.run_stream(...)` — pydantic-ai's `run_sync()` will not work inside the workflow's event loop.
## Implementation examples
### Without dependencies
```python theme={null}
class FilterSpamNode(AgentNode):
class OutputType(AgentNode.OutputType):
reasoning: str = Field(description="Reasoning for the spam classification.")
confidence: float = Field(ge=0, le=1)
is_human: bool = Field(description="True if the message is from a human.")
def get_agent_config(self) -> AgentConfig:
return AgentConfig(
instructions=(
"You are a helpful assistant that filters messages to determine "
"whether they are written by a human or are spam generated by a bot."
),
output_type=self.OutputType,
model_provider=ModelProvider.OPENAI,
model_name="gpt-5.4-mini",
)
async def process(self, task_context: TaskContext) -> TaskContext:
event: CustomerCareEventSchema = task_context.event
result = await self.agent.run(user_prompt=event.model_dump_json())
self.save_output(result.output)
return task_context
```
### With dependencies and dynamic instructions
Use `deps_type` + `@self.agent.instructions` to inject runtime context (for example, retrieval results in a RAG node) into the system prompt at call time:
```python theme={null}
class GenerationNode(AgentNode):
class DepsType(AgentNode.DepsType):
context: RetrievalResults
class OutputType(AgentNode.OutputType):
answer: str
sources: list[str]
confidence: float = Field(ge=0, le=1)
def get_agent_config(self) -> AgentConfig:
return AgentConfig(
instructions=(
"You are a helpful assistant that answers questions using the "
"retrieved documents."
),
output_type=GenerationNode.OutputType,
deps_type=GenerationNode.DepsType,
model_provider=ModelProvider.OPENAI,
model_name="gpt-5.4-mini",
)
async def process(self, task_context: TaskContext) -> TaskContext:
retrieval: RetrievalNode.OutputType = self.get_output(RetrievalNode)
deps = GenerationNode.DepsType(context=retrieval.results)
@self.agent.instructions
def add_rag_context(ctx: RunContext[GenerationNode.DepsType]) -> str:
return (
"Here are the documents I found for your query:\n"
f"{ctx.deps.context.model_dump_json(indent=2)}"
)
result = await self.agent.run(
user_prompt=task_context.event.query,
deps=deps,
)
self.save_output(result.output)
return task_context
```
## Key features
* **Type-safe outputs** — Pydantic `OutputType` validates what the model returns.
* **Flexible dependencies** — `DepsType` + `RunContext` plug structured context into tools and instructions.
* **Seven model providers** — switch between OpenAI, Azure OpenAI, Anthropic, Google Gemini, Google Vertex AI, Bedrock, and Ollama by changing one enum.
* **Dynamic instructions** — augment the system prompt per run via `@self.agent.instructions`.
* **Tool integration** — register pydantic-ai tools or built-in tools via `AgentConfig.tools` / `builtin_tools`.
* **Langfuse-ready** — tracing is on by default via `instrument=True`; the workflow decides whether traces are actually exported.
# Agent Streaming Node
Source: https://launchpad.datalumina.com/docs/core/agent-streaming-node
Learn how to build streaming LLM responses with the AgentStreamingNode class for real-time token delivery.
The `AgentStreamingNode` class extends `AgentNode` to support real-time streaming of LLM responses. Instead of waiting for the complete response, tokens are yielded as they're generated.
**Key difference from AgentNode:**
* `AgentNode.process()` returns `TaskContext`
* `AgentStreamingNode.process()` returns `AsyncIterator[Dict[str, Any]]`
## AgentStreamingNode Class Structure
```python theme={null}
class AgentStreamingNode(AgentNode, ABC):
def __init__(self, task_context: TaskContext = None):
super().__init__(task_context=task_context)
@abstractmethod
async def process(self, task_context: TaskContext) -> AsyncIterator[Dict[str, Any]]:
pass
async def stream_text_deltas(
self,
stream_result,
debounce_by: float = 0.01,
) -> AsyncIterator[dict]:
...
async def stream_structured_deltas(
self,
stream_result,
debounce_by: float = 0.01,
) -> AsyncIterator[dict]:
...
def completion_chunk(self, content: str) -> dict:
...
```
## Streaming Methods
### stream\_text\_deltas
Streams plain text responses, extracting only the new tokens (deltas) from each chunk:
```python theme={null}
async def stream_text_deltas(
self,
stream_result,
debounce_by: float = 0.01,
) -> AsyncIterator[dict]:
previous_text = ""
async for text_chunk in stream_result.stream_text(debounce_by=debounce_by):
if text_chunk.startswith(previous_text):
delta_text = text_chunk[len(previous_text):]
else:
delta_text = text_chunk
if not delta_text:
continue
previous_text = text_chunk
yield self.completion_chunk(delta_text)
```
**Parameters:**
* `stream_result` - The streaming result from `agent.run_stream()`
* `debounce_by` - Delay in seconds between updates (default: 0.01)
### stream\_structured\_deltas
Streams structured Pydantic model outputs:
```python theme={null}
async def stream_structured_deltas(
self,
stream_result,
debounce_by: float = 0.01,
) -> AsyncIterator[dict]:
async for chunk in stream_result.stream_output(debounce_by=debounce_by):
if chunk.model_dump():
yield self.completion_chunk(chunk.model_dump())
```
### completion\_chunk
Formats content into OpenAI-compatible completion chunks:
```python theme={null}
def completion_chunk(self, content: str) -> dict:
return {
"object": "chat.completion.chunk",
"model": "default",
"choices": [
{
"index": 0,
"delta": {"role": "assistant", "content": content},
"finish_reason": None,
}
],
}
```
## Implementation Examples
### Text Streaming Node
Stream plain text responses token by token:
```python theme={null}
from typing import AsyncIterator, Dict, Any
from launchpad.core.nodes.agent import AgentConfig, ModelProvider
from launchpad.core.nodes.agent_streaming_node import AgentStreamingNode
from launchpad.core.task import TaskContext
from launchpad.workflows.examples.streaming.schema import OpenAIChatSchema
class TextStreamingNode(AgentStreamingNode):
def get_agent_config(self) -> AgentConfig:
return AgentConfig(
model_provider=ModelProvider.OPENAI,
model_name="gpt-5.4-mini",
output_type=str,
)
async def process(self, task_context: TaskContext) -> AsyncIterator[Dict[str, Any]]:
event: OpenAIChatSchema = task_context.event
async with self.agent.run_stream(user_prompt=event.get_message()) as result:
async for chunk in self.stream_text_deltas(result):
yield chunk
```
### Structured Streaming Node
Stream structured outputs with multiple fields:
```python theme={null}
class StructuredStreamingNode(AgentStreamingNode):
class OutputType(AgentStreamingNode.OutputType):
thinking: str
reply: str
def get_agent_config(self) -> AgentConfig:
return AgentConfig(
model_provider=ModelProvider.OPENAI,
model_name="gpt-5.4-mini",
output_type=self.OutputType,
)
async def process(self, task_context: TaskContext) -> AsyncIterator[Dict[str, Any]]:
event: OpenAIChatSchema = task_context.event
async with self.agent.run_stream(user_prompt=event.get_message()) as result:
async for chunk in self.stream_structured_deltas(result):
yield chunk
```
## Using in Workflows
The `Workflow` class automatically detects `AgentStreamingNode` instances and yields their events directly:
```python theme={null}
# In workflow.run_stream_async()
if isinstance(node_instance, AgentStreamingNode):
async for stream_event in node_instance.process(task_context):
yield stream_event # Events flow directly to client
else:
task_context = await node_instance.process(task_context)
```
Use `run_stream_async()` instead of `run()` or `run_async()` when your workflow contains streaming nodes:
```python theme={null}
workflow = MyStreamingWorkflow(enable_tracing=True)
async for event in workflow.run_stream_async(event_data):
# Process each streaming event
print(event)
```
## Key Features
* **Delta extraction** - Only transmits new tokens, not accumulated text
* **Debouncing** - Configurable delay to batch rapid updates
* **OpenAI format** - Chunks follow the OpenAI streaming specification
* **Structured support** - Stream complex Pydantic models, not just text
* **Workflow integration** - Automatic detection and handling by the workflow engine
# Base Node
Source: https://launchpad.datalumina.com/docs/core/base-node
Learn about the foundational Node class that all other node types inherit from, and how to implement custom processing logic.
The `Node` class is the foundation for all other node types. Each specialized node implements the `process()` method, which contains the step’s core logic.
Use `process()` to define how the node handles data during execution. This design keeps implementations consistent and extensible.
## Node Class
The base `Node` class provides the essential structure for all workflow processing steps:
```python theme={null}
class Node(ABC):
def save_output(self, output: BaseModel):
self.task_context.nodes[self.node_name] = output
def get_output(self, node_class: Type["Node"]) -> Optional[OutputType]:
return self.task_context.nodes.get(node_class.__name__, None)
@property
def node_name(self) -> str:
return self.__class__.__name__
@abstractmethod
async def process(self, task_context: TaskContext) -> TaskContext:
pass
async def cleanup(self) -> None:
"""Release per-instance resources. Called even if process() raised."""
pass
```
**Key features:**
* Abstract base class for a consistent interface
* Node name property uses the class name
* Abstract `process` method for subclasses
* Async support for non-blocking operations
* Optional `cleanup()` hook for releasing clients/connections, called by the orchestrator after each node run (including when an exception propagates)
## Basic Implementation
Here is a simple example of a custom node:
```python theme={null}
class ValidateInputNode(Node):
async def process(self, task_context: TaskContext) -> TaskContext:
# Your custom processing logic here
return task_context
```
Always return the `task_context` to maintain data flow through the pipeline.
## Storing and Accessing Node Results
**Why store results?** Persist outputs so later nodes can use them.
### Storing Node Results
Use `save_output()` to store results in the task context:
```python theme={null}
class ValidateInputNode(Node):
class OutputType(BaseModel):
is_valid: bool
validation_score: float
def validate_input(self, event: ExampleEvent):
# Your validation logic here
return self.OutputType(is_valid=True, validation_score=0.95)
async def process(self, task_context: TaskContext) -> TaskContext:
event: ExampleEvent = task_context.event
result = self.validate_input(event)
# Store the result using save_output
self.save_output(result)
return task_context
```
### Accessing Node Results
Retrieve results from previous nodes using `get_output()`:
```python theme={null}
class CalculateDifferenceNode(Node):
async def process(self, task_context: TaskContext) -> TaskContext:
# Access results from the ValidateInputNode
validation_result: ValidateInputNode.OutputType = self.get_output(ValidateInputNode)
# Use the validation result in your processing
if validation_result and validation_result.is_valid:
# Process valid data
pass
return task_context
```
## Key benefits
* **Consistency** - A predictable interface across nodes
* **Flexibility** - Customize while maintaining structure
* **Composability** - Combine and reorder nodes freely
* **Testability** - Test nodes independently with mock contexts
# Concurrent Node
Source: https://launchpad.datalumina.com/docs/core/concurrent-node
Learn how to implement concurrent processing with ConcurrentNode to improve performance by running independent operations simultaneously.
This node replaces `ParallelNode` (since v3.0.0). It executes multiple child nodes concurrently when their operations do not depend on each other.
**Use cases:**
* **Independent Validations** - Running validation steps concurrently without dependencies
* **Parallel Transformations** - Applying independent transformations or checks simultaneously
* **Performance Optimization** - Reducing overall task completion time by leveraging parallelism
* **Guardrails Processing** - Running multiple guardrails or safety checks simultaneously
After implementing the child nodes (which can be regular `Node` or `AgentNode` instances), add them to the `NodeConfig` in the `WorkflowSchema` using the `concurrent_nodes` parameter.
## ConcurrentNode Class
```python theme={null}
class ConcurrentNode(Node, ABC):
"""
Base class for nodes that execute other nodes concurrently using asyncio.
This class provides a method to execute a list of nodes concurrently on a single thread,
using asyncio.gather. This ensures that I/O-bound operations can proceed in parallel
without blocking the main thread or event loop.
Subclasses must implement the `process` method to define the specific logic of the concurrent node.
"""
async def execute_nodes_concurrently(self, task_context: TaskContext):
node_config: NodeConfig = task_context.metadata["nodes"][self.__class__]
coroutines = [
node(task_context).process(task_context)
for node in node_config.concurrent_nodes
]
return await asyncio.gather(*coroutines)
@abstractmethod
async def process(self, task_context: TaskContext) -> TaskContext:
pass
```
## Implementation Example
```python theme={null}
class AnalyzeTicketNode(ConcurrentNode):
async def process(self, task_context: TaskContext) -> TaskContext:
await self.execute_nodes_concurrently(task_context)
return task_context
```
### WorkflowSchema Configuration
```python theme={null}
class CustomerCareWorkflow(Workflow):
workflow_schema = WorkflowSchema(
description="Customer care workflow with concurrent analysis",
event_schema=CustomerCareEventSchema,
start=AnalyzeTicketNode,
nodes=[
NodeConfig(
node=AnalyzeTicketNode,
connections=[TicketRouterNode],
description="Concurrent analysis of customer ticket",
concurrent_nodes=[
DetermineTicketIntentNode,
FilterSpamNode,
ValidateTicketNode,
],
),
],
)
```
## How It Works
1. **Node Configuration** - Configure the concurrent nodes in your WorkflowSchema by specifying them in the `concurrent_nodes` list
2. **Concurrent Execution** - When `execute_nodes_concurrently()` is called, it creates coroutines for each child node and runs them simultaneously using `asyncio.gather()`
3. **Result Collection** - All child nodes process the same `task_context` and can store their results independently using `save_output()` or `task_context.update_node()`
4. **Workflow Continuation** - After all concurrent nodes complete, the workflow continues to the next node in the pipeline
## Performance Considerations
When to use: I/O-bound operations; independent steps; reduce total time; multiple validations or analyses.
When not to use: dependent outputs; CPU-bound tasks; overhead outweighs benefits; strict sequential processing needed.
# Types of Nodes
Source: https://launchpad.datalumina.com/docs/core/nodes
Explore the four primary node types that serve as building blocks for workflow processing steps, each designed for specific use cases.
## Overview
Nodes are the building blocks of workflow processing. There are four primary node types, each designed for a specific part of data processing or control.
**Primary Node Types:**
* **[Node](#base-node)** - Performs basic processing tasks with custom logic
* **[AgentNode](#agent-node)** - Handles processing using Large Language Models (LLM)
* **[ConcurrentNode](#concurrent-node)** - Executes multiple nodes concurrently for better performance
* **[BaseRouter](#router-node)** - Directs data flow using conditional routing logic
## Node Architecture
Each node type serves a distinct purpose in the workflow ecosystem:
**Node Categories:**
**Processing Nodes:**
* **Node**: Custom processing logic for any computational task
* **AgentNode**: LLM-powered processing for AI-driven operations
**Control Nodes:**
* **ConcurrentNode**: Parallel execution of independent operations
* **BaseRouter**: Conditional branching based on processing results
## When to Use Each Node Type
1. **Node for custom logic**: Data validation or computations without AI
2. **AgentNode for AI**: Natural language understanding and generation
3. **ConcurrentNode for parallelism**: Independent operations running simultaneously
4. **BaseRouter for conditional flow**: Branch based on results or business rules
## Node Hierarchy
All specialized node types inherit from the base `Node` class, ensuring a consistent interface while providing specialized functionality for different use cases.
```mermaid theme={null}
graph TD
A[Node] --> B[AgentNode]
A --> C[ConcurrentNode]
A --> D[BaseRouter]
E[RouterNode] -. routing rule .-> D
```
## Design Principles
* **Single Responsibility** - Each node type has a clear, single responsibility aligned with its specific use case
* **Extensibility** - Abstract base classes allow for easy extension while maintaining consistent interfaces
* **Composability** - Nodes can be easily combined to create complex workflows leveraging each type's strengths
* **Consistency** - Workflow nodes share common patterns for context management, error handling, and result storage; `RouterNode` rules use the same `get_output()` / `save_output()` helpers for routing decisions
# Overview
Source: https://launchpad.datalumina.com/docs/core/overview
Learn about the core package that provides a flexible DAG-based workflow system for processing tasks through interconnected nodes.
The `core` package provides a flexible framework for defining and executing workflows. It implements a DAG‑based system where tasks move through processing nodes, and each node updates a shared task context.
## Package Structure
The `core` package consists of the following components:
**Core Components:**
* **Nodes** - Node implementations for workflow processing
* **Schema** - Schema definitions for workflow configuration
* **Task Context** - Task context definitions for state management
* **Validation** - Workflow validation utilities
* **Workflow** - Workflow orchestration and execution
```
app/launchpad/core/
├── nodes/ # Node implementations for workflow processing
├── schema.py # Schema definitions for workflow configuration
├── task.py # Task context definitions
├── validate.py # Workflow validation utilities
└── workflow.py # Workflow orchestration
```
## Key Features
**Key Features:**
* **Workflow orchestration**: Define processing pipelines with interconnected nodes
* **Nested workflows**: Call a child workflow from a node and keep using the same `TaskContext`
* **Task context management**: Centralized state enables data flow between nodes
* **Node-based architecture**: Modular, reusable processing components
* **Schema validation**: Built‑in checks ensure valid DAGs and clear structure
# Router Node
Source: https://launchpad.datalumina.com/docs/core/router-node
Implement dynamic workflow routing with BaseRouter and RouterNode classes to create conditional branching based on processing results.
This node enables dynamic routing by selecting the next node based on routing rules. Each rule is a `RouterNode`. If no rule matches, the router uses a fallback node.
**Router components:**
* **BaseRouter** - Orchestrates routing decisions and manages the routing process
* **RouterNode** - Individual routing rules implementing conditional logic
## Router Classes
### BaseRouter
```python theme={null}
class BaseRouter(Node):
async def process(self, task_context: TaskContext) -> TaskContext:
pass
def route(self, task_context: TaskContext) -> Node:
for route_node in self.routes:
route_node.task_context = task_context
next_node = route_node.determine_next_node(task_context)
if next_node:
return next_node
return self.fallback if self.fallback else None
```
### RouterNode
```python theme={null}
class RouterNode(ABC):
@abstractmethod
def determine_next_node(self, task_context: TaskContext) -> Optional[Node]:
pass
@property
def node_name(self):
return self.__class__.__name__
```
## How routing works
1. **Router Node Reached** - The workflow detects `is_router=True` on the current `NodeConfig`
2. **Route Evaluation** - The workflow calls `BaseRouter.route()`, which iterates through each `RouterNode` and calls `determine_next_node()`
3. **First Match Wins** - The first router rule that returns a node determines the next step
4. **Fallback Handling** - If no rules match, the router uses the fallback node
5. **Class Resolution** - The selected node instance is converted back to its class and used as the next workflow node
## Implementation example
### Main Router
```python theme={null}
class TicketRouterNode(BaseRouter):
def __init__(self):
self.routes = [
CloseTicketRouter(),
EscalationRouter(),
InvoiceRouter(),
]
self.fallback = GenerateResponseNode()
```
### Router rule examples
```python theme={null}
class CloseTicketRouter(RouterNode):
def determine_next_node(self, task_context: TaskContext) -> Optional[Node]:
output: FilterSpamNode.OutputType = task_context.nodes["FilterSpamNode"][
"result"
].output
if not output.is_human and output.confidence > 0.8:
return CloseTicketNode()
return None
class EscalationRouter(RouterNode):
def determine_next_node(self, task_context: TaskContext) -> Optional[Node]:
analysis = task_context.nodes["DetermineTicketIntentNode"]["result"].output
if analysis.intent.escalate or analysis.escalate:
return EscalateTicketNode()
return None
```
## Best practices
* **Order Matters** - Arrange router nodes in order of priority; first match wins
* **Clear Conditions** - Keep routing conditions explicit and easy to reason about
* **Fallback Strategy** - Provide a meaningful fallback node
* **Test Coverage** - Test all routing paths
## Configuration in WorkflowSchema
```python theme={null}
NodeConfig(
node=TicketRouterNode,
connections=[CloseTicketNode, EscalateTicketNode, ProcessInvoiceNode, GenerateResponseNode],
is_router=True,
description="Route tickets based on analysis results",
)
```
# Task Context
Source: https://launchpad.datalumina.com/docs/core/task-context
Learn how the TaskContext provides stateful data management throughout workflow execution, enabling seamless data sharing between nodes.
The task context is a stateful Pydantic model used throughout the workflow. It provides a single reference point accessible from any node, so relevant data can be stored and retrieved as needed.
## TaskContext Class
The `TaskContext` class serves as the central data container for workflow execution:
```python theme={null}
class TaskContext(BaseModel):
event: Any
nodes: Dict[str, Any] = Field(default_factory=dict)
metadata: Dict[str, Any] = Field(default_factory=dict)
should_stop: bool = Field(default=False)
trace_id: str | None = Field(default=None)
def update_node(self, node_name: str, **kwargs):
self.nodes[node_name] = {**self.nodes.get(node_name, {}), **kwargs}
def stop_workflow(self) -> None:
self.should_stop = True
```
**TaskContext attributes:**
* **`event`** — the original triggering event, parsed against the workflow's `event_schema`.
* **`nodes`** — results from each node's execution, keyed by class name.
* **`metadata`** — workflow-level metadata (the workflow orchestrator stores the node registry here under `metadata["nodes"]`).
* **`should_stop`** — set to `True` via `stop_workflow()` to halt execution cleanly after the current node finishes.
* **`trace_id`** — Langfuse trace ID captured on entry when `enable_tracing=True`; useful when you want to log or surface the trace URL back to the caller.
## The Event Attribute
The `event` attribute in `TaskContext` serves as the main entry point for workflow input data. When a workflow is initialized, the provided event is parsed according to the `event_schema` specified within the `WorkflowSchema`.
This mechanism supports many event formats because each workflow defines its own event schema. You can run multiple workflows for different inputs without changing the shared workflow infrastructure.
**Example WorkflowSchema:**
```python theme={null}
class ExampleWorkflow(Workflow):
workflow_schema = WorkflowSchema(
description="",
event_schema=ExampleEventSchema,
start=InitialNode,
nodes=[
NodeConfig(
node=InitialNode,
connections=[],
description="",
concurrent_nodes=[],
),
],
)
```
## Type Hinting
By default, the `event` attribute of `TaskContext` has the type `Any`, which means you won't get autocomplete or type checking when accessing its fields or methods.
However, since each workflow defines its own `event_schema`, you already know the expected structure of `event` within that workflow.
To benefit from IDE features like autocomplete and static type checking, explicitly type the `event` attribute when retrieving it from the `TaskContext`. This makes your code more readable and helps catch errors earlier.
**Example:**
```python theme={null}
event: ExampleEventSchema = task_context.event
```
**Implementation steps:**
1. **Define Event Schema** - Create a Pydantic model for your event structure
2. **Configure Workflow** - Set the event\_schema in your WorkflowSchema
3. **Type the Event** - Cast the event to your schema type in node processing
4. **Enjoy Type Safety** - Get full IDE support and compile-time error checking
## Stopping a workflow mid-flight
Any node can halt the run after it returns by calling `task_context.stop_workflow()`. The orchestrator checks `should_stop` between nodes and exits the loop cleanly, preserving everything saved so far. Typical uses: a guardrail node detects a prompt injection, or a router decides there is nothing left to do.
```python theme={null}
class GuardrailNode(Node):
async def process(self, task_context: TaskContext) -> TaskContext:
if self.is_blocked(task_context.event):
task_context.stop_workflow()
return task_context
```
## Best practices
* **Event schema design**: Use specific, well-described field names
* **Node result storage**: Store results under the node class name for consistency
* **Metadata usage**: Keep workflow-level configuration in metadata
* **Type safety**: Cast the event to your schema type in node processing
# Workflow Orchestration
Source: https://launchpad.datalumina.com/docs/core/workflow
Understand the DAG-based workflow system that manages node execution and data flow through directed processing pipelines.
The workflow system is built around the concept of a directed acyclic graph (DAG) where nodes represent processing steps and edges represent the flow of data between them. The `Workflow` class in `workflow.py` serves as the orchestrator, managing the execution flow and passing the task context between nodes.
## Workflow Class
The foundation of the workflow system is the abstract `Workflow` class:
```python theme={null}
class Workflow(ABC):
"""Abstract base class for defining processing workflows.
The Workflow class provides a framework for creating processing workflows
with multiple nodes and routing logic. Each workflow must define its structure
using a WorkflowSchema.
"""
```
**Key capabilities:**
* Abstract base class for concrete workflows
* Schema validation during initialization
* Node execution and routing management
* Task context passing between nodes
* Optional Langfuse tracing via `enable_tracing=True` (default `False`)
## Execution Methods
A `Workflow` instance exposes three entry points:
| Method | Signature | Use when |
| ------------------ | ---------------------------------------------- | ------------------------------------------------------------------------------------------------------ |
| `run` | `run(event=None, *, context=None)` | Calling from sync Python (Celery task, script). Wraps `run_async` in `asyncio.run`. |
| `run_async` | `await run_async(event=None, *, context=None)` | Calling from an existing event loop (FastAPI route, parent node delegating to a child workflow). |
| `run_stream_async` | `async for chunk in run_stream_async(event)` | Streaming nodes that yield SSE chunks (see the [SSE Streaming example](/docs/examples/sse-streaming)). |
`run` and `run_async` both return the final `TaskContext`. Pass `event=` for a fresh run, or `context=` to continue with an existing `TaskContext` — this is how one workflow composes another without losing accumulated state.
```python theme={null}
# Enable Langfuse tracing for a single invocation
workflow = CustomerCareWorkflow(enable_tracing=True)
# Standard entry from a sync caller
context = workflow.run(event_data)
# Inside a FastAPI handler
context = await workflow.run_async(event_data)
# Compose a child workflow, carrying the parent's context forward
child = InvoiceWorkflow()
context = await child.run_async(context=context)
```
When `enable_tracing=True` and Langfuse credentials are missing or invalid, the workflow constructor raises `LangfuseAuthenticationError`.
## Nested Workflows
Use nested workflows when one step in a larger workflow deserves its own workflow definition. A parent node can delegate to a child workflow and pass the current `TaskContext` forward:
```python theme={null}
class RunReplyWorkflowNode(Node):
class OutputType(BaseModel):
delegated: bool
workflow: str
async def process(self, task_context: TaskContext) -> TaskContext:
await ReplyDraftWorkflow().run_async(context=task_context)
self.save_output(
self.OutputType(delegated=True, workflow="ReplyDraftWorkflow")
)
return task_context
```
When `context=` is provided, the child workflow reuses the existing `TaskContext` instead of parsing a fresh event. That means:
* The child workflow can read the same `task_context.event` and previous node outputs.
* Outputs saved by child nodes remain available to later parent nodes.
* The workflow engine temporarily swaps in the child's node registry while the child runs, then restores the parent's registry afterward.
* `should_stop` is reset when entering the child workflow so a child can run even if a parent step previously stopped another branch.
This pattern is useful for reusable sub-processes like drafting a reply, extracting invoice data, running a review workflow, or grouping a specialized sequence behind a single parent node.
See the [Nested Workflow example](/docs/examples/nested-workflow) for a complete parent/child workflow implementation.
## Schema Definitions
The `schema.py` module defines the structure and configuration of workflows using Pydantic models.
### WorkflowSchema
```python theme={null}
class WorkflowSchema(BaseModel):
"""Schema definition for a complete workflow.
WorkflowSchema defines the overall structure of a processing workflow,
including its entry point and all constituent nodes.
"""
```
**Key attributes:**
* `description`: Optional description of the workflow's purpose
* `event_schema`: Pydantic model for validating incoming events
* `start`: The entry point Node class for the workflow
* `nodes`: List of NodeConfig objects defining the workflow structure
**Benefits:**
* **Type Safety**: Pydantic models ensure type validation
* **Documentation**: Built-in schema documentation
* **Validation**: Automatic validation of workflow structure
* **IDE Support**: Full autocomplete and type checking
### NodeConfig
```python theme={null}
class NodeConfig(BaseModel):
"""Configuration model for workflow nodes.
NodeConfig defines the structure and behavior of a single node within
a workflow, including its connections to other nodes and routing properties.
"""
```
**NodeConfig attributes:**
* `node`: The Node class to be instantiated
* `connections`: List of Node classes this node can connect to
* `is_router`: Flag indicating if this node performs routing logic
* `description`: Optional description of the node's purpose
* `concurrent_nodes`: Optional list of Node classes that can run concurrently
## Workflow Validation
The `validate.py` module provides validation logic for workflow schemas, ensuring they form valid directed acyclic graphs (DAGs) and have proper routing configurations.
**Validation Features:**
1. **DAG Validation** - Validates that the workflow forms a proper DAG with no cycles
2. **Reachability Check** - Ensures all nodes are reachable from the start node
3. **Router Validation** - Validates that only router nodes have multiple connections
## Workflow Example
Here's a complete example of a workflow implementation:
```python theme={null}
from launchpad.core.schema import WorkflowSchema, NodeConfig
from launchpad.core.workflow import Workflow
from launchpad.workflows.examples.quickstart.schema import CustomerCareEventSchema
from launchpad.workflows.examples.quickstart.nodes.analyze_ticket_node import AnalyzeTicketNode
from launchpad.workflows.examples.quickstart.nodes.close_ticket_node import CloseTicketNode
from launchpad.workflows.examples.quickstart.nodes.determine_intent_ticket_node import (
DetermineTicketIntentNode,
)
from launchpad.workflows.examples.quickstart.nodes.escalate_ticket_node import EscalateTicketNode
from launchpad.workflows.examples.quickstart.nodes.filter_spam import FilterSpamNode
from launchpad.workflows.examples.quickstart.nodes.generate_response_node import GenerateResponseNode
from launchpad.workflows.examples.quickstart.nodes.process_invoice_node import ProcessInvoiceNode
from launchpad.workflows.examples.quickstart.nodes.send_reply_node import SendReplyNode
from launchpad.workflows.examples.quickstart.nodes.ticket_router_node import TicketRouterNode
from launchpad.workflows.examples.quickstart.nodes.validate_ticket_node import ValidateTicketNode
class CustomerCareWorkflow(Workflow):
workflow_schema = WorkflowSchema(
description="Customer care ticket processing workflow",
event_schema=CustomerCareEventSchema,
start=AnalyzeTicketNode,
nodes=[
NodeConfig(
node=AnalyzeTicketNode,
connections=[TicketRouterNode],
concurrent_nodes=[
DetermineTicketIntentNode,
FilterSpamNode,
ValidateTicketNode,
],
),
NodeConfig(
node=TicketRouterNode,
connections=[
CloseTicketNode,
EscalateTicketNode,
GenerateResponseNode,
ProcessInvoiceNode,
],
is_router=True,
),
NodeConfig(
node=GenerateResponseNode,
connections=[SendReplyNode],
),
],
)
```
This example demonstrates a typical workflow pattern: concurrent analysis, routing based on results, and a terminal action node for the selected path.
# Langfuse Tracing
Source: https://launchpad.datalumina.com/docs/examples/langfuse-tracing
End-to-end example of a traced moderation workflow with Langfuse observability
This example shows how the Launchpad integrates with [Langfuse](https://langfuse.com) to trace every workflow step and every LLM call. It ships as `LangfuseTracingWorkflow` in `app/launchpad/workflows/examples/langfuse_tracing/` and is registered as `WorkflowRegistry.LANGFUSE_TRACING`.
For how tracing is wired into the core `Workflow` class, see [Langfuse Integration](/docs/tools/langfuse).
## What the workflow does
A simple moderation pipeline for user comments:
1. `ViolationDetectionNode` — an `AgentNode` that classifies whether a comment violates policy.
2. `ContextSummaryResult` — an `AgentNode` that summarizes the comment for the audit log.
3. `RemoveCommentNode` — a plain `Node` that deletes the comment when the previous step flagged it.
Each node runs inside its own Langfuse span when `enable_tracing=True`, so you can see timings, inputs, outputs, and LLM calls for the whole run in the Langfuse dashboard.
## Schema
```python theme={null}
class LangfuseTracingEventSchema(BaseModel):
event: str
timestamp: datetime
comment_id: str
thread_id: str
user_id: str
content: str
```
## Workflow definition
```python theme={null}
class LangfuseTracingWorkflow(Workflow):
workflow_schema = WorkflowSchema(
description="",
event_schema=LangfuseTracingEventSchema,
start=ViolationDetectionNode,
nodes=[
NodeConfig(
node=ViolationDetectionNode,
connections=[ContextSummaryResult],
),
NodeConfig(
node=ContextSummaryResult,
connections=[RemoveCommentNode],
),
NodeConfig(
node=RemoveCommentNode,
connections=[],
),
],
)
```
## Violation detection node
```python theme={null}
class ViolationDetectionNode(AgentNode):
class OutputType(AgentNode.OutputType):
comment_id: str
violation: bool
reason: Optional[str] = None
def get_agent_config(self) -> AgentConfig:
return AgentConfig(
instructions=(
"Determine whether the comment is a violation or not. If it is a "
"violation, provide a reason for violation. If it is not a "
"violation, provide a reason for non-violation."
),
output_type=self.OutputType,
deps_type=LangfuseTracingEventSchema,
model_provider=ModelProvider.OPENAI,
model_name="gpt-5.4-mini",
instrument=True,
)
async def process(self, task_context: TaskContext) -> TaskContext:
event: LangfuseTracingEventSchema = task_context.event
@self.agent.instructions
async def add_context() -> str:
return event.model_dump_json()
result = await self.agent.run(user_prompt=event.model_dump_json())
self.save_output(result.output)
return task_context
```
The other two nodes are thin — `ContextSummaryResult` follows the same pattern with a summarization prompt, and `RemoveCommentNode` reads `ViolationDetectionNode.OutputType` via `get_output()` and logs the deletion.
## Running the example
Add to `.env` (or your shell environment):
```bash theme={null}
LANGFUSE_PUBLIC_KEY=pk-lf-...
LANGFUSE_SECRET_KEY=sk-lf-...
LANGFUSE_BASE_URL=https://cloud.langfuse.com
```
```bash theme={null}
uv run playground/langfuse_tracing.py
```
The script loads `app/launchpad/workflows/examples/langfuse_tracing/request_examples/violation.json`, instantiates the workflow via `WorkflowRegistry.LANGFUSE_TRACING.value()`, and runs it.
Open the Langfuse dashboard. You should see a trace named `LangfuseTracingWorkflow` with child spans for each node (`ViolationDetectionNode`, `ContextSummaryResult`, `RemoveCommentNode`) and the underlying LLM generations.
The playground instantiates the workflow without arguments, which defaults to `enable_tracing=False`. To capture traces, update the script to `WorkflowRegistry.LANGFUSE_TRACING.value(enable_tracing=True)`, or instantiate `LangfuseTracingWorkflow(enable_tracing=True)` directly.
## Example event
`app/launchpad/workflows/examples/langfuse_tracing/request_examples/violation.json`:
```json theme={null}
{
"event": "comment_posted",
"timestamp": "2026-04-17T12:00:00Z",
"comment_id": "comment-123",
"thread_id": "thread-abc",
"user_id": "user-42",
"content": "This is a test comment that should be evaluated for policy violations."
}
```
# Nested Workflow
Source: https://launchpad.datalumina.com/docs/examples/nested-workflow
Compose Launchpad workflows by delegating to a child workflow with a shared TaskContext
This deterministic example shows one Launchpad workflow calling another workflow while reusing the same `TaskContext`. It is useful when a larger workflow needs to delegate a focused sub-process without losing the parent workflow's accumulated state.
The example ships as `NestedWorkflow` in `app/launchpad/workflows/examples/nested_workflow/` and is registered as `WorkflowRegistry.NESTED_WORKFLOW`.
## What it demonstrates
* Parent workflow composition with `await ReplyDraftWorkflow().run_async(context=task_context)`
* A child workflow that writes output into the same `TaskContext`
* A runnable example that does not require LLM, tracing, database, or vector-search credentials
* The `Workflow.run_async(context=...)` path added for workflow composition
## Workflow graph
```mermaid theme={null}
flowchart TD
A[RunReplyWorkflowNode] --> B[ReplyDraftWorkflow]
B --> C[DraftReplyNode]
C --> D[SendReplyNode]
```
## Parent workflow
```python theme={null}
from launchpad.core.schema import NodeConfig, WorkflowSchema
from launchpad.core.workflow import Workflow
from launchpad.workflows.examples.nested_workflow.nodes.run_reply_workflow_node import (
RunReplyWorkflowNode,
)
from launchpad.workflows.examples.nested_workflow.nodes.send_reply_node import (
SendReplyNode,
)
from launchpad.workflows.examples.nested_workflow.schema import (
NestedWorkflowEventSchema,
)
class NestedWorkflow(Workflow):
workflow_schema = WorkflowSchema(
description="Parent workflow that delegates reply drafting to a child workflow.",
event_schema=NestedWorkflowEventSchema,
start=RunReplyWorkflowNode,
nodes=[
NodeConfig(
node=RunReplyWorkflowNode,
connections=[SendReplyNode],
description="Calls ReplyDraftWorkflow with the shared context.",
),
],
)
```
## Delegating to the child workflow
The parent node passes the current task context into the child workflow. Because the context is shared, `DraftReplyNode` can save its output and `SendReplyNode` can read it later in the parent flow.
```python theme={null}
class RunReplyWorkflowNode(Node):
class OutputType(BaseModel):
delegated: bool
workflow: str
async def process(self, task_context: TaskContext) -> TaskContext:
await ReplyDraftWorkflow().run_async(context=task_context)
self.save_output(self.OutputType(delegated=True, workflow="ReplyDraftWorkflow"))
return task_context
```
## Run it
```bash theme={null}
uv run playground/nested_workflow.py
```
The script loads `app/launchpad/workflows/examples/nested_workflow/request_examples/billing_question.json`, runs `WorkflowRegistry.NESTED_WORKFLOW.value()`, and prints the final `TaskContext`.
# PGVector RAG
Source: https://launchpad.datalumina.com/docs/examples/pgvector-rag
Build Retrieval Augmented Generation workflows using PostgreSQL with the pgvector extension
Retrieval Augmented Generation (RAG) grounds LLM responses in documents you control, rather than relying only on what the model learned during training. This example ships as `RagExampleWorkflow` in `app/launchpad/workflows/examples/pgvector_rag/`.
Stay on `main`. Do not check out `example/pgvector-rag` or tags like `example/pgvector-rag-v3.3.0` — those are frozen pre-v3.4 snapshots and will not match these docs.
Instead of a dedicated vector database, you reuse the PostgreSQL instance the Launchpad already runs with the `pgvector` extension enabled.
## Why pgvector?
* **Simple setup** — no additional vector database; the default PostgreSQL service already has pgvector available.
* **Single source of truth** — vectors live next to your relational data, so retrieval queries can join against business tables.
* **Cost effective** — no separate vector database to operate or pay for.
## Components
The workflow is a two-node pipeline: `RetrievalNode` → `GenerationNode`.
| File | Purpose |
| ------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| `app/launchpad/workflows/examples/pgvector_rag/schema.py` | `RagExampleEventSchema` — accepts a single `query: str`. |
| `app/launchpad/workflows/examples/pgvector_rag/services.py` | `PgvectorRAGService` — wraps `vecs` + OpenAI embeddings; exposes `get_embedding`, `get_collection`, `upsert`, `parse_results`, `count_tokens`, `disconnect`. |
| `app/launchpad/workflows/examples/pgvector_rag/nodes/retrieval_node.py` | `RetrievalNode` — embeds the query, performs a cosine-distance lookup, and writes the hits to the context. |
| `app/launchpad/workflows/examples/pgvector_rag/nodes/generation_node.py` | `GenerationNode` — grounds an agent response in the retrieved chunks and returns `answer`, `sources`, `confidence`. |
| `app/launchpad/workflows/examples/pgvector_rag/workflow.py` | `RagExampleWorkflow` registered as `WorkflowRegistry.PGVECTOR_RAG`. |
## Retrieval
```python theme={null}
class RetrievalNode(Node):
class OutputType(Node.OutputType):
results: RetrievalResults
async def process(self, task_context: TaskContext) -> TaskContext:
rag_service = PgvectorRAGService()
collection = rag_service.get_collection()
event: RagExampleEventSchema = task_context.event
embedding = rag_service.get_embedding(event.query)
results = collection.query(
data=embedding,
limit=3,
measure="cosine_distance",
include_value=False,
include_metadata=True,
)
self.save_output(self.OutputType(results=rag_service.parse_results(results)))
rag_service.disconnect()
return task_context
```
## Generation
```python theme={null}
class GenerationNode(AgentNode):
class DepsType(AgentNode.DepsType):
context: RetrievalResults
class OutputType(AgentNode.OutputType):
answer: str
sources: list[str]
confidence: float = Field(ge=0, le=1)
def get_agent_config(self) -> AgentConfig:
return AgentConfig(
instructions=(
"You are a helpful assistant that answers questions using the "
"retrieved documents."
),
output_type=GenerationNode.OutputType,
deps_type=GenerationNode.DepsType,
model_provider=ModelProvider.OPENAI,
model_name="gpt-5.4-mini",
)
async def process(self, task_context: TaskContext) -> TaskContext:
retrieval_results: RetrievalNode.OutputType = self.get_output(RetrievalNode)
deps = GenerationNode.DepsType(context=retrieval_results.results)
@self.agent.instructions
def add_rag_context(ctx: RunContext[GenerationNode.DepsType]) -> str:
return (
"Here are the documents I found for your query:\n"
f"{ctx.deps.context.model_dump_json(indent=2)}"
)
result = await self.agent.run(user_prompt=task_context.event.query, deps=deps)
self.save_output(result.output)
return task_context
```
## Running the example
From `docker/`, run `./start.sh` to bring up the API, Celery worker, Redis, and PostgreSQL with pgvector. Supabase services and Caddy stay disabled unless you uncomment them in `docker-compose.yml`.
From `app/launchpad/`, run `./migrate.sh` so the pgvector extension and tables exist.
Use `PgvectorRAGService` directly to seed your collection:
```python theme={null}
from launchpad.workflows.examples.pgvector_rag.services import PgvectorRAGService
service = PgvectorRAGService()
text = "Vector search enables semantic retrieval."
embedding = service.get_embedding(text)
service.upsert([("doc-1", embedding, {"source": "handbook", "text": text})])
service.disconnect()
```
Use the dedicated playground script to exercise the whole pipeline against a sample query:
```bash theme={null}
uv run playground/pgvector_rag.py
```
The script loads `app/launchpad/workflows/examples/pgvector_rag/request_examples/query.json`, instantiates `RagExampleWorkflow`, and prints the generated answer with sources and confidence.
## Customization
* **Swap embedding model** — pass a different `embedding_model` to `PgvectorRAGService` (default `text-embedding-3-small`, 1536 dims).
* **Change retrieval** — adjust `limit`, `measure`, or metadata filters on `collection.query(...)` in `RetrievalNode`.
* **Change generation model** — update `model_provider` / `model_name` in `GenerationNode.get_agent_config`.
# SSE Streaming
Source: https://launchpad.datalumina.com/docs/examples/sse-streaming
Build real-time streaming chat applications with Server-Sent Events and OpenAI-compatible API
This example demonstrates how to implement true Server-Sent Events (SSE) streaming using the OpenAI-compatible `/v1/chat/completions` endpoint. Unlike simulated streaming, this delivers tokens to the client as they are generated.
The streaming endpoint follows the OpenAI API specification, making it compatible with existing OpenAI client libraries and tools.
Stay on `main`. There is no `example/chat` branch, and tags like `example/chat-v3.2.0` are a retired Next.js chatbot that will not match these docs. The current chat example is this SSE endpoint — there is no separate frontend resource.
## Why SSE Streaming?
* **Real-time delivery** - Tokens stream to the client as they're generated
* **Reduced perceived latency** - Users see responses immediately, not after full generation
* **Native browser support** - SSE works out of the box with EventSource API
* **OpenAI compatibility** - Drop-in replacement for OpenAI streaming endpoints
## How It Works
The streaming architecture connects your workflow directly to the HTTP response:
```mermaid theme={null}
sequenceDiagram
participant Client
participant API
participant Workflow
participant StreamingNode
Client->>API: POST /v1/chat/completions
API->>Workflow: run_stream_async()
loop For each node
Workflow->>StreamingNode: process()
loop For each token
StreamingNode-->>Workflow: yield chunk
Workflow-->>API: yield chunk
API-->>Client: data: {chunk}\n\n
end
end
API-->>Client: data: [DONE]\n\n
```
## The Streaming Endpoint
The endpoint accepts OpenAI-compatible chat completion requests and returns an SSE stream:
```python theme={null}
from fastapi import APIRouter
from starlette.responses import StreamingResponse
from launchpad.workflows.examples.streaming.schema import OpenAIChatSchema
from launchpad.utils.event_stream_generator import event_stream_generator
from launchpad.workflows.examples.streaming.workflow import ExampleStreamingWorkflow
router = APIRouter()
@router.post("/chat/completions", dependencies=[])
async def handle_chat_completion_streaming(data: OpenAIChatSchema) -> StreamingResponse:
workflow = ExampleStreamingWorkflow(enable_tracing=True)
workflow_stream = workflow.run_stream_async(data.model_dump())
return StreamingResponse(
event_stream_generator(workflow_stream),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
},
)
```
The endpoint is mounted under the `/v1` prefix by `app/launchpad/api/router.py`, so the full path is `POST /v1/chat/completions`.
## Example Workflow
The `ExampleStreamingWorkflow` (registered as `WorkflowRegistry.STREAMING`) demonstrates a two-node streaming pipeline:
```python theme={null}
from launchpad.core.schema import WorkflowSchema, NodeConfig
from launchpad.core.workflow import Workflow
from launchpad.workflows.examples.streaming.schema import OpenAIChatSchema
from launchpad.workflows.examples.streaming.nodes.text_streaming_node import TextStreamingNode
from launchpad.workflows.examples.streaming.nodes.structured_streaming_node import StructuredStreamingNode
class ExampleStreamingWorkflow(Workflow):
workflow_schema = WorkflowSchema(
description="SSE streaming example with text and structured output",
event_schema=OpenAIChatSchema,
start=TextStreamingNode,
nodes=[
NodeConfig(
node=TextStreamingNode,
connections=[StructuredStreamingNode],
),
NodeConfig(
node=StructuredStreamingNode,
connections=[],
),
],
)
```
## Streaming Node Examples
### Text Streaming
Stream plain text responses token by token:
```python theme={null}
from typing import AsyncIterator, Dict, Any
from launchpad.core.nodes.agent import AgentConfig, ModelProvider
from launchpad.core.nodes.agent_streaming_node import AgentStreamingNode
from launchpad.core.task import TaskContext
from launchpad.workflows.examples.streaming.schema import OpenAIChatSchema
class TextStreamingNode(AgentStreamingNode):
def get_agent_config(self) -> AgentConfig:
return AgentConfig(
model_provider=ModelProvider.OPENAI,
model_name="gpt-5.4-mini",
output_type=str,
)
async def process(self, task_context: TaskContext) -> AsyncIterator[Dict[str, Any]]:
event: OpenAIChatSchema = task_context.event
async with self.agent.run_stream(user_prompt=event.get_message()) as result:
async for chunk in self.stream_text_deltas(result):
yield chunk
```
### Structured Streaming
Stream structured Pydantic model outputs:
```python theme={null}
class StructuredStreamingNode(AgentStreamingNode):
class OutputType(AgentStreamingNode.OutputType):
thinking: str
reply: str
def get_agent_config(self) -> AgentConfig:
return AgentConfig(
model_provider=ModelProvider.OPENAI,
model_name="gpt-5.4-mini",
output_type=self.OutputType,
)
async def process(self, task_context: TaskContext) -> AsyncIterator[Dict[str, Any]]:
event: OpenAIChatSchema = task_context.event
async with self.agent.run_stream(user_prompt=event.get_message()) as result:
async for chunk in self.stream_structured_deltas(result):
yield chunk
```
## Run it
This example ships on `main` under `app/launchpad/workflows/examples/streaming/`. After [installation](/docs/getting-started/installation):
```bash theme={null}
uv run playground/streaming.py
```
## Testing the Endpoint
Use curl to test the streaming endpoint:
```bash theme={null}
curl -X POST http://localhost:8080/v1/chat/completions \
-H "Content-Type: application/json" \
-d '{
"model": "default",
"messages": [
{"role": "user", "content": "Hello, how are you?"}
]
}'
```
You'll see SSE events streaming in real-time:
```
data: {"object": "chat.completion.chunk", "model": "default", "choices": [{"index": 0, "delta": {"role": "assistant", "content": "Hello"}, "finish_reason": null}]}
data: {"object": "chat.completion.chunk", "model": "default", "choices": [{"index": 0, "delta": {"role": "assistant", "content": "!"}, "finish_reason": null}]}
data: [DONE]
```
## Key Features
* **Delta extraction** - Only sends new tokens, not the full accumulated text
* **Debouncing** - Configurable delay (default 10ms) to batch rapid updates
* **OpenAI chunk format** - Compatible with standard OpenAI client libraries
* **Multi-node streaming** - Chain multiple streaming nodes in a single workflow
* **Langfuse tracing** - Full observability with `enable_tracing=True`
# Celery Workers
Source: https://launchpad.datalumina.com/docs/framework/celery-workers
Implement background task processing with Celery for scalable GenAI event-driven architectures
## What is it
Celery is a distributed task queue system that allows you to run time-consuming operations asynchronously in the background.
It's particularly useful for I/O-bound operations, periodic tasks, and long-running processes.
## Why we use it
The Launchpad uses Celery because of the unique characteristics of GenAI workflows that rely heavily on LLM calls. Unlike traditional APIs with predictable response times, LLM calls can vary significantly in duration depending on factors like prompt complexity, model load, and request volume.
Running these workflows as background tasks ensures the main application remains responsive to receive new events and serve other endpoints, while Celery workers handle the time-intensive AI processing separately. This architecture allows us to scale processing capacity easily by spawning additional workers when demand increases.
Celery provides robust capabilities including task distribution, automatic retries, monitoring, and horizontal scaling, making it perfect for the unpredictable nature of GenAI workflow processing.
## Where we use it
Background task processing for workflows, a task is usually queued whenever an event is received through an endpoint.
## Further Reading
For advanced Celery configuration, monitoring, and deployment patterns, refer to the [official Celery documentation](https://docs.celeryproject.org/).
# Docker
Source: https://launchpad.datalumina.com/docs/framework/docker
Understand the modular Docker architecture for development and production deployments
The Launchpad uses a modular Docker Compose architecture that lets you include or exclude services based on your needs. All services run on a shared bridge network for seamless communication.
## Compose File Structure
The Docker setup is split into four compose files:
| File | Purpose |
| ------------------------------ | ------------------------------------------- |
| `docker-compose.yml` | Main orchestrator that includes other files |
| `docker-compose.launchpad.yml` | Core application services |
| `docker-compose.supabase.yml` | Supabase backend services |
| `docker-compose.caddy.yml` | Reverse proxy with automatic HTTPS |
### Main Compose File
The `docker-compose.yml` file controls which services to include. By default only the core Launchpad stack is on; Supabase and Caddy are commented out. Uncomment Supabase when you need Studio/Auth/Realtime/Storage, and uncomment Caddy when you need HTTPS in front of the API (typical for VPS deployments):
```yaml theme={null}
include:
- path: ./docker-compose.launchpad.yml
# - path: ./docker-compose.supabase.yml # Uncomment to enable Supabase services
# - path: ./docker-compose.caddy.yml # Uncomment to enable Caddy/HTTPS
networks:
default:
driver: bridge
external: true
name: "${PROJECT_NAME}-network"
```
## Core Application Services
The `docker-compose.launchpad.yml` file contains the essential application services:
| Service | Image | Port | Purpose |
| ------------------ | ---------------------------- | ---- | -------------------------- |
| **api** | Custom (Dockerfile.api) | 8080 | FastAPI application server |
| **celery\_worker** | Custom (Dockerfile.celery) | - | Async task processing |
| **redis** | redis:latest | 6379 | Message broker & cache |
| **db** | supabase/postgres:15.8.1.085 | 5432 | PostgreSQL database |
The API and Celery services mount the host `app/` into `/workspace/app` for live code reloading during development. The package is installed editable inside the image, so edits in the host `app/launchpad/` tree take effect without a rebuild.
## Supabase Services
The `docker-compose.supabase.yml` file is opt-in and provides the full Supabase stack:
| Service | Image | Purpose |
| ------------- | ---------------------- | ------------------------------ |
| **studio** | supabase/studio | Dashboard UI |
| **kong** | kong:2.8.1 | API gateway (ports 8000, 8443) |
| **auth** | supabase/gotrue | Authentication service |
| **rest** | postgrest/postgrest | Auto-generated REST API |
| **realtime** | supabase/realtime | WebSocket subscriptions |
| **storage** | supabase/storage-api | File storage |
| **imgproxy** | darthsim/imgproxy | Image transformations |
| **meta** | supabase/postgres-meta | Schema introspection |
| **functions** | supabase/edge-runtime | Deno edge functions |
| **analytics** | supabase/logflare | Log aggregation |
| **vector** | timberio/vector | Log collection |
| **supavisor** | supabase/supavisor | Connection pooling |
When Supabase is enabled, access the Studio dashboard at `http://localhost:8000` with credentials from `docker/.env`.
## Caddy Reverse Proxy
The `docker-compose.caddy.yml` file provides:
* Automatic HTTPS certificate management
* Reverse proxy to application services
* HTTP/2 support
* Ports: 80 (HTTP), 443 (HTTPS), 2019 (admin API)
## Including and Excluding Services
### Option 1: Edit docker-compose.yml
Uncomment the services you need:
```yaml theme={null}
include:
- path: ./docker-compose.launchpad.yml # Always include
# - path: ./docker-compose.supabase.yml # Enable Supabase
# - path: ./docker-compose.caddy.yml # Enable Caddy
```
### Option 2: Use CLI flags
Specify which compose files to use:
```bash theme={null}
# Default core stack
docker compose -f docker-compose.launchpad.yml up
# Core + Supabase
docker compose -f docker-compose.launchpad.yml \
-f docker-compose.supabase.yml up
# Whatever is enabled in docker-compose.yml
docker compose up
```
## Management Scripts
The `docker/` directory includes helper scripts:
### start.sh
Creates the network if needed and starts all services:
```bash theme={null}
cd docker && ./start.sh
```
### stop.sh
Stops all running containers:
```bash theme={null}
cd docker && ./stop.sh
```
### logs.sh
Interactive log viewer with service selection:
```bash theme={null}
cd docker && ./logs.sh
```
## Architecture Overview
```
┌─────────────────────────────────────────────────────────┐
│ Caddy (optional) │
│ :80, :443, :2019 │
└────────────────────────┬────────────────────────────────┘
│
┌────────────────────────▼────────────────────────────────┐
│ Kong API Gateway │
│ :8000, :8443 │
└───────┬─────────────┬─────────────┬─────────────────────┘
│ │ │
┌───────▼───────┐ ┌───▼───────┐ ┌───▼───────────────────┐
│ Auth/GoTrue │ │ PostgREST │ │ Storage / Realtime / │
│ :9999 │ │ :3000 │ │ Functions / etc. │
└───────┬───────┘ └─────┬─────┘ └───────────┬───────────┘
│ │ │
┌───────▼───────────────▼───────────────────▼─────────────┐
│ PostgreSQL Database │
│ supabase/postgres:15.8.1.085 │
│ :5432 (direct) / :6543 (pooled) │
└─────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────┐
│ Application Services │
│ ┌─────────────────┐ ┌──────────────┐ ┌───────────┐ │
│ │ FastAPI (:8080) │ │ Celery Worker│ │ Redis │ │
│ │ /workspace/app │ │ async tasks │ │ :6379 │ │
│ └─────────────────┘ └──────────────┘ └───────────┘ │
└─────────────────────────────────────────────────────────┘
All services connected via: ${PROJECT_NAME}-network
```
## Environment Variables
The Launchpad has two `.env` files because local Python runs and Docker Compose read configuration in different ways.
| File | Used by | Purpose |
| ----------------------- | ------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------- |
| `.env` at the repo root | Local Python commands, including playground scripts, unit tests, notebooks, and one-off scripts run with `uv` | Keeps local development credentials close to the Python process that loads them with `python-dotenv` |
| `docker/.env` | Docker Compose, containers, the database, and optional Supabase services | Feeds Compose interpolation, container environment variables, database credentials, ports, JWT/secrets, and Supabase-specific settings |
`docker/.env` is intentionally much larger than the root `.env` because self-hosted Supabase requires many variables for Auth, Realtime, Storage, Studio, Kong, analytics, and related services. Those values stay in `docker/.env` even when Supabase is commented out by default, so opting in later is a compose-file change rather than a config migration.
For local Python runs that connect to the Docker database through `localhost`, the root `.env` database user depends on whether Supabase is enabled. Use `DATABASE_USER=postgres` with the default Launchpad-only stack. When `docker-compose.supabase.yml` is included, port `5432` is served by Supavisor, so use `DATABASE_USER=postgres.`, for example `postgres.launchpad`.
Key environment variables in `docker/.env`:
| Category | Variables |
| ----------------- | ----------------------------------------------------------------- |
| **Project** | `PROJECT_NAME` |
| **Database** | `POSTGRES_PASSWORD`, `POSTGRES_DB`, `POSTGRES_PORT` |
| **JWT** | `JWT_SECRET`, `JWT_EXPIRY`, `ANON_KEY`, `SERVICE_ROLE_KEY` |
| **LLM Providers** | `OPENAI_API_KEY`, `ANTHROPIC_API_KEY`, `MISTRAL_API_KEY`, etc. |
| **Langfuse** | `LANGFUSE_PUBLIC_KEY`, `LANGFUSE_SECRET_KEY`, `LANGFUSE_BASE_URL` |
Copy `docker/.env.example` to `docker/.env` and configure your values before starting.
# FastAPI
Source: https://launchpad.datalumina.com/docs/framework/fastapi
Leverage FastAPI framework for endpoints, authentication, and HTTP communication in GenAI applications
## What is it
FastAPI is a modern, fast web framework for building APIs with Python 3.7+ based on standard Python type hints. It provides automatic API documentation, data validation, and serialization.
## Why we use it
The Launchpad uses FastAPI because of its tight integration with Pydantic for automatic data validation, built-in OpenAPI documentation generation, and excellent performance for asynchronous operations. Its dependency injection system makes it perfect for managing database sessions, authentication, and other shared resources.
## Where we use it
We just use it to create endpoints, which most of the times is used as the entry point for an event.
## Further Reading
For advanced FastAPI features, middleware, testing, and deployment options, refer to the [official FastAPI documentation](https://fastapi.tiangolo.com/).
# Supabase
Source: https://launchpad.datalumina.com/docs/framework/supabase
Leverage Supabase as a comprehensive backend-as-a-service platform
## What is it
Supabase is an open-source backend-as-a-service (BaaS) platform that provides developers with a complete set of backend tools including PostgreSQL database, authentication, real-time subscriptions, storage, and auto-generated APIs.
## Why we use it
The Launchpad can run the self-hosted Supabase services when a project needs them. They are excluded by default so the local stack stays smaller, but you can opt in without changing application code.
## Where we use it
By default the Launchpad starts the core application stack: FastAPI, Celery, Redis, and PostgreSQL. The PostgreSQL service uses the `supabase/postgres` image because it includes useful extensions like pgvector, but Supabase Studio, Auth, Realtime, Storage, and the gateway are not started unless you uncomment `docker-compose.supabase.yml` in `docker/docker-compose.yml`.
When enabled, the self-hosted Supabase services offer:
* **Authentication**: User management, OAuth providers, JWT tokens for secure API access
* **Real-time subscriptions**: Live updates for collaborative features, chat applications, or dashboard monitoring
* **Storage**: File uploads for documents, images, or model artifacts with integrated access control
* **Auto-generated APIs**: Instant REST endpoints based on database schema, this makes it easy to build custom frontends
* **Row Level Security**: Fine-grained access control policies for multi-tenant applications
## Further Reading
For advanced Supabase features, Edge Functions, and deployment strategies, refer to the [official Supabase documentation](https://supabase.com/docs).
# Installation Guide
Source: https://launchpad.datalumina.com/docs/getting-started/installation
Step-by-step installation for GenAI Launchpad
Before you begin, ensure you have completed the system requirements setup.
Open your terminal and navigate to your desired project directory:
```bash theme={null}
cd desired/project/path
```
Clone the GenAI Launchpad repository:
```bash theme={null}
git clone git@github.com:datalumina/genai-launchpad.git
```
Navigate to the project directory:
```bash theme={null}
cd genai-launchpad
```
Copy the example environment files:
```bash theme={null}
cp .env.example .env && cp docker/.env.example docker/.env
```
These files serve different runtimes:
| File | Used by | What belongs here |
| ------------- | ----------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------- |
| `.env` | Local Python runs from your machine, such as `uv run playground/...`, unit tests, and one-off scripts | LLM provider keys, Langfuse keys, and local app settings loaded by `python-dotenv` |
| `docker/.env` | Docker Compose, the containerized Launchpad stack, the database, and optional Supabase services | Compose project settings, database credentials, container ports, JWT/secrets, Docker-side provider keys, and optional Supabase configuration |
`docker/.env` is much larger mostly because self-hosted Supabase needs many settings, even though Supabase services are excluded by default.
Navigate to the Docker directory and start the containers:
```bash theme={null}
cd docker && ./start.sh
```
This command will:
* Build all Docker containers
* Start all services via Docker Compose
Verify the containers are running:
```bash theme={null}
docker ps
```
Return to the project root and sync the environment:
```bash theme={null}
cd ../
uv sync
```
`uv sync` creates and manages the `.venv`, installs the pinned Python version if needed, and installs the Launchpad package in editable mode.
Navigate to the package directory where `alembic.ini` and the migration scripts live:
```bash theme={null}
cd app/launchpad
```
Create a new migration (you'll be prompted for a description):
```bash theme={null}
./makemigration.sh
```
When prompted, enter a descriptive message like "init db" for your first migration.
Apply the migration:
```bash theme={null}
./migrate.sh
```
Supabase Studio, Auth, Realtime, Storage, and the Supabase gateway are excluded by default. To enable them, uncomment `docker-compose.supabase.yml` in `docker/docker-compose.yml`, then restart the Docker stack.
If you run Python locally against the Docker database, update the root `.env` database user from `postgres` to `postgres.`, for example `postgres.launchpad` with the default `docker/.env` settings. Supavisor requires the tenant-qualified username.
When enabled, Studio is available at [http://localhost:8000](http://localhost:8000) with the dashboard credentials from `docker/.env`.
Windows users should use either Git Bash or Ubuntu WSL (WSL preferred) for the best experience.
Open Git Bash or Ubuntu WSL and navigate to your desired project directory:
```bash theme={null}
cd desired/project/path
```
Clone the GenAI Launchpad repository:
```bash theme={null}
git clone git@github.com:datalumina/genai-launchpad.git
```
Navigate to the project directory:
```bash theme={null}
cd genai-launchpad
```
Copy the example environment files:
```bash theme={null}
cp .env.example .env && cp docker/.env.example docker/.env
```
These files serve different runtimes:
| File | Used by | What belongs here |
| ------------- | ----------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------- |
| `.env` | Local Python runs from your machine, such as `uv run playground/...`, unit tests, and one-off scripts | LLM provider keys, Langfuse keys, and local app settings loaded by `python-dotenv` |
| `docker/.env` | Docker Compose, the containerized Launchpad stack, the database, and optional Supabase services | Compose project settings, database credentials, container ports, JWT/secrets, Docker-side provider keys, and optional Supabase configuration |
`docker/.env` is much larger mostly because self-hosted Supabase needs many settings, even though Supabase services are excluded by default.
Navigate to the Docker directory and start the containers:
```bash theme={null}
cd docker && ./start.sh
```
This command will:
* Build all Docker containers
* Start all services via Docker Compose
Verify the containers are running:
```bash theme={null}
docker ps
```
Return to the project root and sync the environment:
```bash theme={null}
cd ../
uv sync
```
`uv sync` creates and manages the `.venv`, installs the pinned Python version if needed, and installs the Launchpad package in editable mode.
Navigate to the package directory where `alembic.ini` and the migration scripts live:
```bash theme={null}
cd app/launchpad
```
Create a new migration (you'll be prompted for a description):
```bash theme={null}
./makemigration.sh
```
When prompted, enter a descriptive message like "init db" for your first migration.
Apply the migration:
```bash theme={null}
./migrate.sh
```
Supabase Studio, Auth, Realtime, Storage, and the Supabase gateway are excluded by default. To enable them, uncomment `docker-compose.supabase.yml` in `docker/docker-compose.yml`, then restart the Docker stack.
If you run Python locally against the Docker database, update the root `.env` database user from `postgres` to `postgres.`, for example `postgres.launchpad` with the default `docker/.env` settings. Supavisor requires the tenant-qualified username.
When enabled, Studio is available at [http://localhost:8000](http://localhost:8000) with the dashboard credentials from `docker/.env`.
## Exercise workflows from the playground
The `playground/` directory contains one script per reference workflow. Each script loads a matching JSON fixture from the workflow's `request_examples/` folder, runs the workflow in-process, and prints the final `TaskContext`. They are the fastest way to iterate on a workflow without running the full API + Celery stack.
All examples ship on `main`. Ignore `example/*` branches and tags (`example/pgvector-rag-v3.3.0`, `example/chat-v3.2.0`, and similar) — they are frozen pre-v3.4 snapshots and will not match these docs.
| Script | Workflow | What it does |
| -------------------------------- | -------------------------- | ----------------------------------------------------------------------------------------------------- |
| `playground/quickstart.py` | `CustomerCareWorkflow` | Runs a customer-care ticket end-to-end (concurrent analysis → routing → reply). |
| `playground/nested_workflow.py` | `NestedWorkflow` | Runs a parent workflow that delegates reply drafting to a child workflow with the same `TaskContext`. |
| `playground/streaming.py` | `ExampleStreamingWorkflow` | Iterates `run_stream_async` and prints each streamed chunk. |
| `playground/langfuse_tracing.py` | `LangfuseTracingWorkflow` | Runs the moderation pipeline; set `LANGFUSE_*` env vars to capture traces. |
| `playground/pgvector_rag.py` | `RagExampleWorkflow` | Runs retrieval + generation against the pgvector collection. |
Run any of them with `uv`:
```bash theme={null}
uv run playground/quickstart.py
```
# System Requirements
Source: https://launchpad.datalumina.com/docs/getting-started/requirements
Prerequisites for running GenAI Launchpad
Before you begin, ensure you have the following tools installed on your system:
## Development Environment
While we recommend one of the above IDEs for the best development experience, any code editor will work with GenAI Launchpad.
Free code editor with excellent Python support
AI-powered code editor built on VS Code
Professional Python IDE with advanced features
## Required Software
Required for containerized deployment and local development environment.
Make sure Docker Desktop is running before proceeding with the installation.
Version control system for cloning the repository and managing code. [Download Git](https://git-scm.com/downloads)
Fast Python package installer and resolver. [Installation Guide](https://docs.astral.sh/uv/getting-started/installation/#__tabbed_1_1)
UV is significantly faster than pip and provides better dependency resolution for Python projects.
The backend is pinned to Python 3.13.13, matching the container image and `pyproject.toml`. `uv sync` will install a matching interpreter automatically if you do not already have one.
## System Specifications
**Minimum Requirements:**
* 8GB RAM (16GB recommended)
* 10GB free disk space
* Modern CPU with virtualization support
* Stable internet connection for downloading dependencies
# End-to-End Testing
Source: https://launchpad.datalumina.com/docs/quickstart/e2e-testing
Exercise the full API + Celery + database path with a single HTTP call
After validating a workflow with the playground, run it through the full stack — API, database, Celery worker, and Langfuse if enabled — to confirm the pieces work together.
## Hit the `/events` endpoint
The generic events endpoint stores the payload, queues `process_incoming_event`, and returns HTTP 202 immediately. `get_workflow_type()` in `app/launchpad/api/events.py` currently returns `WorkflowRegistry.QUICKSTART.name`, so every event posted to `/events/` runs through `CustomerCareWorkflow`.
### curl
```bash theme={null}
curl -X POST http://localhost:8080/events/ \
-H "Content-Type: application/json" \
-d @app/launchpad/workflows/examples/quickstart/request_examples/invoice.json
```
Expected response:
```json theme={null}
{"message": "process_incoming_event started `` "}
```
with HTTP status **202 Accepted**.
### Python
```python theme={null}
import json
from pathlib import Path
import requests
event_path = Path("app/launchpad/workflows/examples/quickstart/request_examples/invoice.json")
payload = json.loads(event_path.read_text())
response = requests.post("http://localhost:8080/events/", json=payload)
print(response.status_code, response.text)
assert response.status_code == 202
```
## Prerequisites
* **Docker running** — `docker ps` shows the api, celery\_worker, redis, and db services.
* **API reachable** at `http://localhost:8080`.
* **Celery worker running** — logs show it picking up `process_incoming_event`.
* **Migrations applied** — run `./migrate.sh` from `app/launchpad/` once.
## Monitor results
Run `docker compose logs celery_worker` from `docker/` and look for the `process_incoming_event` task completing.
Query the `events` table with your preferred PostgreSQL client. Each row holds the raw payload in `data` and the final `TaskContext` in `task_context` once the worker finishes.
If you enabled `docker-compose.supabase.yml`, open Studio at [http://localhost:8000](http://localhost:8000), authenticate with the dashboard credentials from `docker/.env`, and inspect the `events` table in the Table Editor.
## Troubleshooting
* **Nothing processes** — check Celery worker logs; `docker compose logs celery_worker` will show import errors immediately.
* **422 from FastAPI** — the endpoint accepts any JSON, but the workflow raises a Pydantic validation error when the payload does not match `CustomerCareEventSchema`. Inspect the Celery logs for the stack trace.
* **Stuck on an older workflow** — update `get_workflow_type()` in `app/launchpad/api/events.py` to return a different `WorkflowRegistry.*.name`, or add a dedicated router module under `app/launchpad/api/` that mounts your workflow on its own path.
# Create API Endpoint
Source: https://launchpad.datalumina.com/docs/quickstart/endpoint
Build a FastAPI endpoint to receive events and trigger workflows
You now set up an endpoint to receive events. The endpoint validates input, persists the event, and enqueues processing.
## Endpoint Implementation
The implementation below creates a simple POST endpoint without authentication (for demonstration purposes):
```python theme={null}
import json
from http import HTTPStatus
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from starlette.responses import Response
from launchpad.database.event import Event
from launchpad.database.repository import GenericRepository
from launchpad.database.session import db_session
from launchpad.worker.config import celery_app
from launchpad.workflows.workflow_registry import WorkflowRegistry
router = APIRouter()
@router.post("/", dependencies=[])
def handle_event(
data: dict,
session: Session = Depends(db_session),
) -> Response:
repository = GenericRepository(session=session, model=Event)
raw_event = data.model_dump(mode="json")
event = Event(data=raw_event, workflow_type=get_workflow_type())
repository.create(obj=event)
task_id = celery_app.send_task(
"process_incoming_event",
args=[str(event.id)],
)
return Response(
content=json.dumps({"message": f"process_incoming_event started `{task_id}` "}),
status_code=HTTPStatus.ACCEPTED,
)
def get_workflow_type() -> str:
return WorkflowRegistry.QUICKSTART.name
```
This router is mounted at `/events` by `app/launchpad/api/router.py`, so the full path is `POST /events/`. The body is accepted as a raw `dict` and validated later by the workflow against `CustomerCareEventSchema` (the `event_schema` on `CustomerCareWorkflow`). This keeps a single endpoint usable for multiple registered workflows.
## Key Actions Performed
FastAPI parses the request body as JSON, then stores it. The workflow itself validates against `CustomerCareEventSchema` when it runs, raising a validation error if the payload is malformed.
The event is stored with raw JSON, the workflow type from the registry, a generated ID, and timestamps.
A Celery task named `process_incoming_event` is queued with the event ID and returns a task ID for tracking.
The endpoint returns HTTP 202 (Accepted) to indicate asynchronous processing.
## Understanding the Components
* **GenericRepository**: Provides database operations (create, read, update, delete) for any model
* **db\_session**: FastAPI dependency that provides a database session for the request
* **celery\_app**: The Celery application instance for queuing background tasks
* **WorkflowRegistry**: Enum containing all registered workflows in the system
## Response Status Code
The endpoint returns **HTTP 202 (Accepted)** rather than 200 (OK) because the workflow processing happens asynchronously, the client is informed that the request is queued, and the actual processing will happen in the background.
## Security Considerations
**Security Note**: This example endpoint has no authentication for simplicity. In production, you should add authentication middleware, implement rate limiting, validate API keys or JWT tokens, and use HTTPS for secure communication.
# Overview
Source: https://launchpad.datalumina.com/docs/quickstart/overview
Build your first GenAI workflow - a customer care automation system
This quickstart shows the Launchpad in action through a customer care example. You follow the steps end‑to‑end without writing new code, and you see how the same steps map to your own client workflows.
## General Development Flow
Create a Pydantic model that defines the structure of your incoming data
Build a FastAPI endpoint that receives events and triggers workflow processing
Design and implement the workflow logic using various node types
## What We're Building
In this quickstart, we'll explore a customer care automation system that:
Processes incoming support tickets through intelligent analysis
Detects and filters out spam messages automatically
Makes smart routing decisions based on ticket content
Creates appropriate AI-powered responses to customer queries
## Getting Started
The fully implemented quickstart workflow ships on `main` under `app/launchpad/workflows/examples/quickstart/`. After installation, run it directly from the playground:
```bash theme={null}
uv run playground/quickstart.py
```
## What You'll Learn
By the end, you understand how to structure event schemas, create API endpoints that trigger workflows, build multi‑node pipelines, integrate AI models for processing, test locally and end‑to‑end, and monitor results in the database or optional Supabase Studio.
# Define Your Schema
Source: https://launchpad.datalumina.com/docs/quickstart/schema
Create a Pydantic schema to define your data structure
You start every workflow by defining the structure of the incoming event. A Pydantic schema becomes the single source of truth for validation, documentation, and serialization.
## Why Schemas Matter
Schemas give you automatic type validation, inline documentation through field descriptions, and straightforward JSON serialization, so you can accept reliable inputs and publish clear API docs without extra boilerplate.
## Customer Care Event Schema
For our customer care use case, we've defined the following schema:
```python theme={null}
from datetime import datetime, timezone
from pydantic import BaseModel, Field
from uuid import UUID, uuid4
class CustomerCareEventSchema(BaseModel):
ticket_id: UUID = Field(
default_factory=uuid4,
description="Unique identifier for the ticket"
)
timestamp: datetime = Field(
default_factory=lambda: datetime.now(timezone.utc),
description="Time when the ticket was created",
)
from_email: str = Field(
...,
description="Email address of the sender"
)
to_email: str = Field(
...,
description="Email address of the recipient"
)
sender: str = Field(
...,
description="Name or identifier of the sender"
)
subject: str = Field(
...,
description="Subject of the ticket"
)
body: str = Field(
...,
description="The body of the ticket"
)
```
## Schema Components Explained
### Automatic ID Generation
The `ticket_id` field uses `default_factory=uuid4` to automatically generate a unique identifier for each ticket when created.
### Timestamp Management
The `timestamp` field automatically captures the creation time in UTC, ensuring consistent time tracking across different timezones.
### Required Fields
Fields marked with `...` (ellipsis) are required and must be provided when creating an event instance.
### Field Descriptions
Each field includes a description that serves as inline documentation and helps with API documentation generation.
## File Location
Colocate each workflow's schema in its own package (`app/launchpad/workflows//schema.py`) so the input model lives next to the workflow that validates against it. Truly cross-workflow schemas can go under `app/launchpad/schemas/`.
# Testing Your Workflow
Source: https://launchpad.datalumina.com/docs/quickstart/testing
Run workflows locally before involving the full API + Celery stack
Test locally first. Each reference workflow ships with a playground script and a matching set of request examples so you can validate logic without running the full API stack.
## Local Testing Setup
* **Request examples**: each workflow keeps its fixtures under `app/launchpad/workflows//request_examples/`. For the quickstart, the files are in `app/launchpad/workflows/examples/quickstart/request_examples/` — `invoice.json`, `policy_question.json`, `product.json`, `prompt_injection.json`, `refund.json`, `service_desk.json`, and `spam.json`.
* **Playground scripts**: one-per-workflow scripts in `playground/` that load a fixture, run the workflow in-process, and print the final `TaskContext`.
## Running the quickstart playground
```bash theme={null}
uv run playground/quickstart.py
```
The script is roughly:
```python theme={null}
event_path = project_root / "app/launchpad/workflows/examples/quickstart/request_examples/invoice.json"
with open(event_path) as f:
event = json.load(f)
workflow = WorkflowRegistry.QUICKSTART.value()
output = workflow.run(event)
print(output.model_dump_json(indent=2))
```
Swap the `event_path` filename to exercise any of the other fixtures.
## Benefits of local testing
### Immediate feedback
You see results instantly without waiting for async processing or checking databases.
### Easy debugging
Add breakpoints and step through your workflow logic in your IDE.
### No extra dependencies
You can test without running Celery workers or the full Supabase stack — only the services the workflow itself needs (for example, the quickstart does not need the database).
### Rapid iteration
Modify and test workflow logic quickly without restarting services.
## Adding your own test events
Add a new file to `app/launchpad/workflows//request_examples/` with your test payload.
Ensure the JSON structure matches the workflow's `event_schema` exactly — `Workflow.run()` parses it with `event_schema(**event)` and will surface validation errors.
Update `event_path` in the matching `playground/.py` script (or run a quick one-off Python snippet).
# Workflow Definition
Source: https://launchpad.datalumina.com/docs/quickstart/workflow-definition
Create the workflow structure and register it in the system
Now we'll define an empty Workflow schema that serves as the container for your workflow logic. This is where you'll specify which event schema to use and how nodes connect together.
## Creating the Workflow Class
Your own workflows live in their own packages under `app/launchpad/workflows//`, with `schema.py`, `workflow.py`, and a `nodes/` directory. The shipped reference workflows are grouped under `app/launchpad/workflows/examples/` so it's clear they're demos you can replace — the quickstart lives at `app/launchpad/workflows/examples/quickstart/workflow.py`:
```python theme={null}
from launchpad.core.schema import WorkflowSchema, NodeConfig
from launchpad.core.workflow import Workflow
from launchpad.workflows.examples.quickstart.schema import CustomerCareEventSchema
class CustomerCareWorkflow(Workflow):
workflow_schema = WorkflowSchema(
description="",
event_schema=CustomerCareEventSchema,
start=...,
nodes=[],
)
```
At this point, the workflow is just a skeleton. We'll add the actual node implementations and connections in the next steps.
## Workflow Schema Components
* **description**: A human-readable description of what the workflow does
* **event\_schema**: The Pydantic schema that defines the expected input data structure
* **start**: The name of the first node to execute in the workflow
* **nodes**: List of all nodes that will be part of this workflow
## Registering the Workflow
To make your workflow available to the system, you need to register it in the `WorkflowRegistry`.
Add a new entry to `app/launchpad/workflows/workflow_registry.py`. The Launchpad ships with five reference workflows under `workflows/examples/` already registered:
```python theme={null}
from enum import Enum
from launchpad.workflows.examples.streaming.workflow import ExampleStreamingWorkflow
from launchpad.workflows.examples.quickstart.workflow import CustomerCareWorkflow
from launchpad.workflows.examples.langfuse_tracing.workflow import (
LangfuseTracingWorkflow,
)
from launchpad.workflows.examples.nested_workflow.workflow import (
NestedWorkflow,
)
from launchpad.workflows.examples.pgvector_rag.workflow import RagExampleWorkflow
class WorkflowRegistry(Enum):
STREAMING = ExampleStreamingWorkflow
QUICKSTART = CustomerCareWorkflow
LANGFUSE_TRACING = LangfuseTracingWorkflow
PGVECTOR_RAG = RagExampleWorkflow
NESTED_WORKFLOW = NestedWorkflow
```
**Important**: The workflow registry enum name (e.g., `QUICKSTART`) is how the rest of the system — for example, the generic `/events` endpoint and Celery worker — looks a workflow up. Choose a descriptive name that clearly indicates the workflow's purpose.
## Why Use a Registry?
The `WorkflowRegistry` enum provides several benefits:
### Centralized Management
All workflows are registered in one place, making it easy to see what workflows are available in your system.
### Type Safety
Using an enum ensures you can only reference workflows that actually exist, preventing typos and runtime errors.
### Easy Discovery
IDEs can autocomplete workflow names, making development faster and reducing errors.
### Dynamic Loading
The system can dynamically load and execute workflows based on the registry entry.
**Tip**: Keep your workflow names consistent with their purpose. For example, use `CUSTOMER_CARE` for customer support workflows, `ORDER_PROCESSING` for e-commerce workflows, etc.
# Workflow Implementation
Source: https://launchpad.datalumina.com/docs/quickstart/workflow-implementation
Build the complete customer care workflow with AI-powered nodes
Now you implement the workflow steps. Each step is a **node**. This example demonstrates reusable patterns you can apply across client projects.
## Workflow Steps
Run three concurrent analyses:
* Determine ticket intent
* Check for spam
* Validate information sufficiency
Make intelligent routing decisions:
* Close spam tickets
* Escalate urgent issues
* Process specific requests (invoices, refunds)
* Generate responses for general queries
Execute the appropriate action based on routing decision
All nodes are located under `app/launchpad/workflows/examples/quickstart/nodes/`, and Jinja2 prompts live in `app/launchpad/workflows/examples/quickstart/prompts/`.
## Step 1: Concurrent Analysis
### AnalyzeTicketNode
Since our three analysis operations are independent, we use `ConcurrentNode` to run them simultaneously:
```python theme={null}
class AnalyzeTicketNode(ConcurrentNode):
async def process(self, task_context: TaskContext) -> TaskContext:
await self.execute_nodes_concurrently(task_context)
return task_context
```
**Tip**: Use concurrency when independent AI calls reduce total processing time.
### DetermineTicketIntentNode
This `AgentNode` uses AI to classify the ticket intent:
```python theme={null}
class CustomerIntent(str, Enum):
GENERAL_QUESTION = "general/question"
PRODUCT_QUESTION = "product/question"
BILLING_INVOICE = "billing/invoice"
REFUND_REQUEST = "refund/request"
@property
def escalate(self) -> bool:
return self in {self.REFUND_REQUEST}
class DetermineTicketIntentNode(AgentNode):
class OutputType(AgentNode.OutputType):
reasoning: str = Field(
description="Explain your reasoning for the intent classification"
)
intent: CustomerIntent
confidence: float = Field(
ge=0, le=1, description="Confidence score for the intent"
)
escalate: bool = Field(
description="Flag to indicate if the ticket needs escalation"
)
class DepsType(AgentNode.DepsType):
from_email: str = Field(..., description="Email address of the sender")
sender: str = Field(..., description="Name or identifier of the sender")
subject: str = Field(..., description="Subject of the ticket")
body: str = Field(..., description="The body of the ticket")
def get_agent_config(self) -> AgentConfig:
return AgentConfig(
instructions=PromptManager.get_prompt(
"ticket_analysis", prompts_dir=PROMPTS_DIR
),
output_type=self.OutputType,
deps_type=self.DepsType,
model_provider=ModelProvider.OPENAI,
model_name="gpt-5.4-mini",
)
async def process(self, task_context: TaskContext) -> TaskContext:
event: CustomerCareEventSchema = task_context.event
deps = self.DepsType(
from_email=event.from_email,
sender=event.sender,
subject=event.subject,
body=event.body,
)
@self.agent.instructions
def add_ticket_context() -> str:
return deps.model_dump_json(indent=2)
result = await self.agent.run(
user_prompt=event.model_dump_json(indent=2),
)
self.save_output(result.output)
return task_context
```
`PROMPTS_DIR` is defined at the top of the module as `Path(__file__).parent.parent / "prompts"` so the loader resolves templates colocated with the workflow.
### FilterSpamNode
Detects spam messages using AI:
```python theme={null}
class FilterSpamNode(AgentNode):
class OutputType(AgentNode.OutputType):
reasoning: str = Field(
description="Explain your reasoning for spam detection"
)
confidence: float = Field(
ge=0, le=1,
description="Confidence score for the classification"
)
is_human: bool = Field(
description="True if human-written, False if spam"
)
def get_agent_config(self) -> AgentConfig:
return AgentConfig(
instructions="You are a helpful assistant that filters messages...",
output_type=self.OutputType,
deps_type=None,
model_provider=ModelProvider.OPENAI,
model_name="gpt-5.4-mini",
)
async def process(self, task_context: TaskContext) -> TaskContext:
event: CustomerCareEventSchema = task_context.event
result = await self.agent.run(
user_prompt=event.model_dump_json(),
)
self.save_output(result.output)
return task_context
```
### ValidateTicketNode
Verifies if the ticket contains actionable information:
```python theme={null}
class ValidateTicketNode(AgentNode):
class OutputType(AgentNode.OutputType):
reasoning: str = Field(
description="Reasoning for actionability determination"
)
confidence: float = Field(
ge=0, le=1,
description="Confidence score"
)
is_actionable: bool = Field(
description="True if ticket is actionable"
)
def get_agent_config(self) -> AgentConfig:
return AgentConfig(
instructions="Review tickets for actionable information...",
output_type=self.OutputType,
deps_type=None,
model_provider=ModelProvider.OPENAI,
model_name="gpt-5.4-mini",
)
async def process(self, task_context: TaskContext) -> TaskContext:
event: CustomerCareEventSchema = task_context.event
result = await self.agent.run(
user_prompt=event.model_dump_json(),
)
self.save_output(result.output)
return task_context
```
## Step 2: Intelligent Routing
### TicketRouterNode
The router examines analysis results and selects the next action:
```python theme={null}
class TicketRouterNode(BaseRouter):
def __init__(self):
self.routes = [
CloseTicketRouter(),
EscalationRouter(),
InvoiceRouter(),
]
self.fallback = GenerateResponseNode()
```
* **CloseTicketRouter**: Closes spam tickets automatically when confidence > 0.8
* **EscalationRouter**: Escalates urgent or sensitive issues to human agents
* **InvoiceRouter**: Routes billing-related requests to invoice processing
### Router Implementation
```python theme={null}
class CloseTicketRouter(RouterNode):
def determine_next_node(self, task_context: TaskContext) -> Optional[Node]:
filter_spam_node: FilterSpamNode.OutputType = self.get_output(FilterSpamNode)
if not filter_spam_node.is_human and filter_spam_node.confidence > 0.8:
return CloseTicketNode()
return None
class EscalationRouter(RouterNode):
def determine_next_node(self, task_context: TaskContext) -> Optional[Node]:
intent_node: DetermineTicketIntentNode.OutputType = self.get_output(
DetermineTicketIntentNode
)
if intent_node.intent.escalate or intent_node.escalate:
return EscalateTicketNode()
return None
class InvoiceRouter(RouterNode):
def determine_next_node(self, task_context: TaskContext) -> Optional[Node]:
intent_node: DetermineTicketIntentNode.OutputType = self.get_output(
DetermineTicketIntentNode
)
if intent_node.intent == CustomerIntent.BILLING_INVOICE:
return ProcessInvoiceNode()
return None
```
## Step 3: Action Nodes
### GenerateResponseNode
Creates AI-powered responses for customer queries:
```python theme={null}
class GenerateResponseNode(AgentNode):
class OutputType(AgentNode.OutputType):
reasoning: str = Field(description="The reasoning for the response")
response: str = Field(description="The response to the ticket")
confidence: float = Field(
ge=0, le=1,
description="Confidence score for response quality"
)
def get_agent_config(self) -> AgentConfig:
return AgentConfig(
instructions=PromptManager.get_prompt(
"customer_ticket_response", prompts_dir=PROMPTS_DIR
),
output_type=self.OutputType,
deps_type=None,
model_provider=ModelProvider.OPENAI,
model_name="gpt-5.4-mini",
)
async def process(self, task_context: TaskContext) -> TaskContext:
event: CustomerCareEventSchema = task_context.event
result = await self.agent.run(
user_prompt=event.model_dump_json(),
)
self.save_output(result.output)
return task_context
```
### SendReplyNode
Delivers the generated response:
```python theme={null}
class SendReplyNode(Node):
async def process(self, task_context: TaskContext) -> TaskContext:
logging.info("Sending reply:")
output: GenerateResponseNode.OutputType = self.get_output(GenerateResponseNode)
logging.info(output.response)
return task_context
```
## Complete Workflow Schema
```python theme={null}
class CustomerCareWorkflow(Workflow):
workflow_schema = WorkflowSchema(
description="Customer care ticket processing workflow",
event_schema=CustomerCareEventSchema,
start=AnalyzeTicketNode,
nodes=[
NodeConfig(
node=AnalyzeTicketNode,
connections=[TicketRouterNode],
description="Analyze ticket concurrently",
concurrent_nodes=[
DetermineTicketIntentNode,
FilterSpamNode,
ValidateTicketNode,
],
),
NodeConfig(
node=TicketRouterNode,
connections=[
CloseTicketNode,
EscalateTicketNode,
GenerateResponseNode,
ProcessInvoiceNode,
],
description="Route based on analysis",
is_router=True,
),
NodeConfig(
node=GenerateResponseNode,
connections=[SendReplyNode],
description="Generate AI response",
),
],
)
```
# Database Migrations
Source: https://launchpad.datalumina.com/docs/tools/alembic
Manage database schema changes systematically with Alembic migrations
## What is it
Alembic is a lightweight database migration tool that manages database schema changes over time. Think of it as **"version control for your database"** - it tracks changes and lets you upgrade or rollback your database structure safely.
## Why we use it
The Launchpad uses Alembic because it integrates seamlessly with our SQLAlchemy models, automatically detects schema changes, and ensures all environments stay synchronized.
## Where we use it
We use Alembic for all database schema changes in the project. Here's how it works:
**Basic Commands:**
* Create Migration: `./makemigration.sh`
* Apply Migration: `./migrate.sh`
**Complete Workflow:**
1. Modify your SQLAlchemy model (e.g., in `database/event.py`)
2. Generate migration: `./makemigration.sh` → Enter migration message
3. Review the generated migration in `alembic/versions/`
4. Apply the migration: `./migrate.sh`
For Alembic to detect your models, you must import them in `alembic/env.py`
```python theme={null}
# alembic/env.py
from launchpad.database.session import Base
from launchpad.database.event import * # Import Event model for autogenerate
# Import additional SQLAlchemy models here when you add them.
```
**Project Structure:**
```
app/launchpad/
├── alembic.ini # Alembic configuration
├── alembic/
│ ├── env.py # Environment setup (import models here!)
│ └── versions/ # Generated migration files
├── makemigration.sh # Create migrations
└── migrate.sh # Apply migrations
```
**Development Commands:**
* Apply migrations: `./migrate.sh`
* Check current version: `alembic current`
* View history: `alembic history`
* Rollback: `alembic downgrade -1`
Always review generated migrations before applying - autogenerate is smart but not perfect!
## Further Reading
For advanced Alembic features and detailed configuration options, refer to the [official Alembic documentation](https://alembic.sqlalchemy.org/).
# Langfuse Integration
Source: https://launchpad.datalumina.com/docs/tools/langfuse
Monitor and trace every step of your GenAI workflows with comprehensive observability
Langfuse is an open-source observability platform for LLM applications that provides tracing, monitoring, and debugging. The integration is built into the Launchpad's core using the native Langfuse SDK.
You can **self-host** for full data control and privacy, which is useful when sensitive data must stay within your infrastructure.
## Why Langfuse?
* **Complete Tracing**: Track every workflow step, node execution, and LLM call
* **Performance Monitoring**: Monitor response times, costs, and success rates
* **Debug Issues**: Detailed logs and traces for troubleshooting failures
Datalumina uses this integration in production to monitor and trace workflows.
## Quick Setup
Create a free account at [langfuse.com](https://langfuse.com) and get your API keys
Add the keys to the environment file for the runtime you are using. Use the root `.env` for local Python runs such as playground scripts and tests. Use `docker/.env` when the workflow runs inside the Docker stack.
```bash theme={null}
LANGFUSE_PUBLIC_KEY=pk-lf-...
LANGFUSE_SECRET_KEY=sk-lf-...
LANGFUSE_BASE_URL=https://cloud.langfuse.com # Or your self-hosted URL
```
Pass `enable_tracing=True` when initializing your workflow:
```python theme={null}
workflow = MyWorkflow(enable_tracing=True)
result = workflow.run(event_data)
```
Run the dedicated playground and check your Langfuse dashboard for traces:
```bash theme={null}
uv run playground/langfuse_tracing.py
```
See the [Langfuse Tracing example](/docs/examples/langfuse-tracing) for the full workflow.
## How It Works
The Langfuse integration uses the native Langfuse SDK to create spans around workflow and node execution:
```python theme={null}
from langfuse import get_client
class Workflow(ABC):
def __init__(self, enable_tracing: bool = False):
if enable_tracing:
langfuse = get_client()
if langfuse.auth_check():
self.langfuse = langfuse
else:
raise LangfuseAuthenticationError(
"Failed to authenticate with Langfuse."
)
```
When tracing is enabled:
* A parent span is created for the entire workflow execution
* Each node gets its own child span with inputs and outputs
* LLM calls within AgentNodes are automatically instrumented
* Errors are captured with full context
## Enabling and Disabling Tracing
Tracing is controlled per-workflow instance:
Tracing defaults to **off** (`enable_tracing=False`) to avoid surprise network calls in tests and local runs. Opt in per workflow instance:
```python theme={null}
# Enable tracing for this invocation
workflow = ExampleStreamingWorkflow(enable_tracing=True)
# Default: no traces sent
workflow = ExampleStreamingWorkflow()
```
If `enable_tracing=True` but Langfuse credentials are missing or invalid, the workflow will raise a `LangfuseAuthenticationError`.
## Core Integration Features
* **Automatic Tracing**: Every workflow execution is automatically traced when enabled
* **Node-Level Visibility**: Individual node executions, inputs, and outputs are captured
* **LLM Call Tracking**: All LLM interactions including prompts, responses, and metadata
* **Error Monitoring**: Failed executions with full stack traces and context
* **Streaming Support**: SSE streaming workflows are fully traced
## Dashboard Features
### Workflow Traces
View complete workflow execution paths with timing, inputs, and outputs for each node.
### Performance Analytics
Monitor average response times, success rates, and cost analysis across workflows.
### LLM Usage Tracking
Track token usage, model performance, and costs across different LLM providers.
### Debug Information
Detailed error logs with full context when workflows fail or perform unexpectedly.
# LLM Providers
Source: https://launchpad.datalumina.com/docs/tools/llm-providers
Configure OpenAI, Azure OpenAI, Anthropic, Google, Bedrock, and Ollama via AgentNode
`AgentNode` picks a model backend via the `ModelProvider` enum in `app/launchpad/core/nodes/agent.py`. Switching providers is a one-line change in the node's `get_agent_config()`; credentials come from environment variables loaded by `python-dotenv` at import time.
```python theme={null}
from launchpad.core.nodes.agent import AgentConfig, ModelProvider
class MyNode(AgentNode):
def get_agent_config(self) -> AgentConfig:
return AgentConfig(
model_provider=ModelProvider.OPENAI,
model_name="gpt-5.4-mini",
output_type=self.OutputType,
)
```
All providers are reached through [pydantic-ai](https://ai.pydantic.dev). `AgentConfig.instrument=True` (the default) wires each call into Langfuse when `enable_tracing=True` on the workflow.
## Supported providers
### OpenAI — `ModelProvider.OPENAI`
Uses `OpenAIResponsesModel`. Good default for new workflows.
| Env var | Purpose |
| ---------------- | ------------------------ |
| `OPENAI_API_KEY` | Standard OpenAI API key. |
```python theme={null}
AgentConfig(model_provider=ModelProvider.OPENAI, model_name="gpt-5.4-mini")
```
### Azure OpenAI — `ModelProvider.AZURE_OPENAI`
Routes through pydantic-ai's `AzureProvider` with `OpenAIResponsesModel`.
| Env var | Purpose |
| -------------------------- | ----------------------------------------------- |
| `AZURE_OPENAI_ENDPOINT` | Azure resource endpoint. |
| `AZURE_OPENAI_API_KEY` | Resource API key. |
| `AZURE_OPENAI_API_VERSION` | API version (defaults to `2025-04-01-preview`). |
Set `model_name` to the Azure deployment name you want to call.
```python theme={null}
AgentConfig(model_provider=ModelProvider.AZURE_OPENAI, model_name="gpt-5-mini")
```
### Anthropic — `ModelProvider.ANTHROPIC`
| Env var | Purpose |
| ------------------- | ------------------ |
| `ANTHROPIC_API_KEY` | Anthropic API key. |
```python theme={null}
AgentConfig(model_provider=ModelProvider.ANTHROPIC, model_name="claude-sonnet-4-6")
```
### Google Gemini — `ModelProvider.GOOGLE_GEMINI`
Uses `GoogleModel` with the standard `GoogleProvider` (API-key auth).
| Env var | Purpose |
| ---------------- | --------------- |
| `GOOGLE_API_KEY` | Gemini API key. |
```python theme={null}
AgentConfig(model_provider=ModelProvider.GOOGLE_GEMINI, model_name="gemini-2.5-pro")
```
### Google Vertex AI — `ModelProvider.GOOGLE_VERTEX_AI`
Uses a service account to authenticate against Vertex AI.
| Env var | Purpose |
| -------------------------------- | --------------------------------------------- |
| `GOOGLE_APPLICATION_CREDENTIALS` | Absolute path to a service account JSON file. |
| `GOOGLE_VERTEX_AI_LOCATION` | Region (defaults to `europe-west1`). |
```python theme={null}
AgentConfig(model_provider=ModelProvider.GOOGLE_VERTEX_AI, model_name="gemini-2.5-pro")
```
### AWS Bedrock — `ModelProvider.BEDROCK`
Creates a `boto3` `bedrock-runtime` client and passes it to `BedrockConverseModel`.
| Env var | Purpose |
| ------------------------------- | ----------------------------- |
| `BEDROCK_AWS_ACCESS_KEY_ID` | AWS access key. |
| `BEDROCK_AWS_SECRET_ACCESS_KEY` | AWS secret. |
| `BEDROCK_AWS_REGION` | AWS region hosting the model. |
```python theme={null}
AgentConfig(
model_provider=ModelProvider.BEDROCK,
model_name="anthropic.claude-sonnet-4-6-v1:0",
)
```
### Ollama — `ModelProvider.OLLAMA`
Uses `OpenAIChatModel` with the pydantic-ai `OllamaProvider`. Ideal for local development against an `ollama serve` instance.
| Env var | Purpose |
| ----------------- | ------------------------------------------------------------------------------------------------ |
| `OLLAMA_BASE_URL` | Full base URL, e.g. `http://localhost:11434/v1`. Required; the node raises `KeyError` otherwise. |
```python theme={null}
AgentConfig(model_provider=ModelProvider.OLLAMA, model_name="llama3.2")
```
## Other AgentConfig knobs
`AgentConfig` forwards common pydantic-ai fields so most tuning happens in one place:
* `instructions` — static system prompt (can be augmented with `@self.agent.instructions` for per-run context).
* `output_type` — return a plain `str` or a `BaseModel` subclass for structured output.
* `deps_type` — a Pydantic model containing dependencies exposed via `RunContext` inside tools and instruction callbacks.
* `tools`, `builtin_tools` — pydantic-ai tool definitions.
* `model_settings` — a `ModelSettings` object to override temperature, max tokens, etc.
* `retries`, `output_retries` — retry behavior on model errors and validation failures.
* `instrument` — defaults to `True`; set to `False` to opt a node out of Langfuse instrumentation even when the workflow has tracing enabled.
# Prompt Management
Source: https://launchpad.datalumina.com/docs/tools/prompt-management
Create and manage dynamic prompts using Jinja templates with YAML frontmatter
Jinja2 lets you keep large, dynamic prompts out of Python strings. The Launchpad ships a small `PromptManager` service (`app/launchpad/services/prompt_loader.py`) that loads `.j2` templates, parses their YAML frontmatter, and renders them with runtime variables using `StrictUndefined` so missing variables fail loudly.
## Why templates
* **Readable structure** — sections like "Role", "Context", "Examples" stay visible in the template instead of buried in string concatenation.
* **Reuse** — Jinja includes, blocks, and macros let you share common snippets across prompts.
* **Conditional content** — show a section only when relevant data is available, or swap tone per user tier.
* **Variable injection** — interpolate runtime data without manual string formatting.
* **Frontmatter metadata** — each template can declare its own `description`, `author`, and any custom fields via YAML frontmatter at the top of the file.
## Template locations
`PromptManager` supports two layouts — use whichever fits the workflow:
| Layout | Path | When to use |
| ----------------- | ------------------------------------------------------ | ---------------------------------------------------------------------------------------------------------------------- |
| Shared prompts | `app/launchpad/prompts/.j2` | Templates reused across many workflows. This is the default; no `prompts_dir` argument needed. |
| Colocated prompts | `app/launchpad/workflows//prompts/.j2` | Templates that belong to a single workflow. Pass `prompts_dir=` so the loader resolves them next to the workflow code. |
The `quickstart` workflow uses the colocated layout — see `app/launchpad/workflows/examples/quickstart/prompts/ticket_analysis.j2`.
## Example template
`app/launchpad/workflows/examples/quickstart/prompts/ticket_analysis.j2`:
```jinja theme={null}
---
description: A template for analyzing incoming {{ pipeline | default('customer support') }} tickets
author: TechGear AI Team
---
You're an AI assistant named {{ name | default('Emma') }}, working for {{ company | default('TechGear') }}.
Your goal is to analyze incoming {{ pipeline | default('support') }} tickets and classify their intent.
# CONTEXT
You will be provided with the following information from a {{ pipeline | default('support') }} ticket:
- Sender: The name or identifier of the person who sent the ticket
- Subject: The subject line of the ticket
- Body: The main content of the ticket
# TASK
Your task is to analyze the ticket and determine its primary intent. You should also provide a confidence score for your classification and explain your reasoning.
```
The lines between `---` markers are YAML frontmatter. They are stripped before the template is rendered and are available via `PromptManager.get_template_info`.
## Using a template from a node
```python theme={null}
from pathlib import Path
from launchpad.core.nodes.agent import AgentNode, AgentConfig, ModelProvider
from launchpad.services.prompt_loader import PromptManager
PROMPTS_DIR = Path(__file__).parent.parent / "prompts"
class DetermineTicketIntentNode(AgentNode):
def get_agent_config(self) -> AgentConfig:
return AgentConfig(
instructions=PromptManager.get_prompt(
"ticket_analysis",
prompts_dir=PROMPTS_DIR,
name="Emma",
company="TechGear",
),
output_type=self.OutputType,
model_provider=ModelProvider.OPENAI,
model_name="gpt-5.4-mini",
)
```
Drop `prompts_dir=PROMPTS_DIR` for shared templates in `app/launchpad/prompts/`.
## Inspecting metadata and variables
`PromptManager.get_template_info(name)` returns the template's metadata plus the set of variables referenced in the body — useful when wiring templates into a registry or generating documentation:
```python theme={null}
info = PromptManager.get_template_info("ticket_analysis")
# {
# "name": "ticket_analysis",
# "description": "A template for analyzing incoming customer support tickets",
# "author": "TechGear AI Team",
# "variables": ["pipeline", "name", "company"],
# "frontmatter": {...},
# }
```
`StrictUndefined` is enabled. Any variable referenced in the template but not passed to `get_prompt` raises `UndefinedError` — prefer Jinja `default(...)` filters for optional fields.
## Further reading
* [Jinja2 documentation](https://jinja.palletsprojects.com/) — filters, macros, inheritance.
* [`python-frontmatter`](https://python-frontmatter.readthedocs.io/) — YAML frontmatter parsing used under the hood.
# Deploying on a VPS
Source: https://launchpad.datalumina.com/docs/tutorials/deploying-on-vps
Step-by-step guide to deploying GenAI Launchpad on a production Linux server
This tutorial provides a step-by-step process for deploying the GenAI Launchpad on a production-ready Linux server.
## Prerequisites
Before proceeding, ensure you have the following:
* A remote server running a Linux distribution (Ubuntu recommended)
* SSH access to the server
* (Optional) A custom domain with access to DNS settings
## Deployment
Log in to your remote server via SSH:
```bash theme={null}
ssh username@remote-ip-address
```
Docker is required to run the Launchpad. Follow the [official installation instructions](https://docs.docker.com/engine/install/) for your Linux distribution.
For Ubuntu, execute the following commands:
### Set up Docker's APT repository
```bash theme={null}
sudo apt-get update
sudo apt-get install -y ca-certificates curl
sudo install -m 0755 -d /etc/apt/keyrings
sudo curl -fsSL https://download.docker.com/linux/ubuntu/gpg -o /etc/apt/keyrings/docker.asc
sudo chmod a+r /etc/apt/keyrings/docker.asc
echo "deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.asc] \
https://download.docker.com/linux/ubuntu \
$(. /etc/os-release && echo "${UBUNTU_CODENAME:-$VERSION_CODENAME}") stable" | \
sudo tee /etc/apt/sources.list.d/docker.list > /dev/null
sudo apt-get update
```
### Install Docker
```bash theme={null}
sudo apt-get install -y docker-ce docker-ce-cli containerd.io docker-buildx-plugin docker-compose-plugin
```
### Verify installation
Ensure Docker is installed correctly by running:
```bash theme={null}
sudo docker run hello-world
```
To deploy the Launchpad, clone the repository onto your server. First, add an SSH key to your GitHub repository.
Replace the repository URL below with your own private GenAI Launchpad repository.
### Generate an SSH key
Generate a new SSH key pair:
```bash theme={null}
ssh-keygen -f ~/.ssh/github
```
Follow the prompts. If you do not wish to set a passphrase, press `Enter` to skip.
Print the public key:
```bash theme={null}
cat ~/.ssh/github.pub
```
Copy the output, as it will be needed in the next step.
### Add the SSH key to GitHub
1. Navigate to your GitHub repository.
2. Open **Settings**.
3. In the left menu, go to **Security > Deploy keys**.
4. Click **Add deploy key** and provide:
* **Title:** `genai-launchpad-prod`
* **Key:** Paste the copied public key.
5. Click **Add key**.
### Clone the repository
Run the following command to clone the repository to the server:
```bash theme={null}
cd /opt && git clone git@github.com:datalumina/genai-launchpad.git
```
Configure the environment file for the runtime you are deploying.
* `.env` at the repo root is for local Python development: playground scripts, unit tests, notebooks, and other commands run with your local `uv`/Python environment.
* `docker/.env` is for Docker Compose, the containerized Launchpad stack, the database, and optional Supabase services. This is the important file for a VPS deployment.
`docker/.env` is much larger mostly because self-hosted Supabase needs many variables for Auth, Realtime, Storage, Studio, Kong, and related services, even though Supabase is excluded by default.
Replace default credentials with secure values before deploying to production.
This step is optional. Skip if you don't need HTTPS or don't have a custom domain.
If you intend to use HTTPS, you must have a domain and configure the appropriate DNS settings.
### Configure domain
1. Create an **A record** in your domain's DNS settings, pointing to your server's IP address.
2. Modify the `.env` file in the `docker/` directory to include:
```dotenv theme={null}
CADDY_DOMAIN=https://yourdomain.com
```
### Open required ports
Ensure ports 80 and 443 are open on your server's firewall:
* **Port 80**: Required for Let's Encrypt certificate validation (ACME challenge)
* **Port 443**: Required for HTTPS traffic
If using `ufw`:
```bash theme={null}
sudo ufw allow 80/tcp
sudo ufw allow 443/tcp
```
Navigate to the `docker/` directory:
```bash theme={null}
cd /opt/genai-launchpad/docker
```
Run the startup script:
```bash theme={null}
./start.sh
```
# Introduction
Source: https://launchpad.datalumina.com/docs/welcome/introduction
A production-ready foundation for event-driven AI applications
You use GenAI Launchpad to start client projects on a solid production foundation without spending weeks on setup. It bridges the gap between proof‑of‑concept AI integrations and production systems by providing a robust, scalable architecture so you can focus on delivering outcomes instead of rebuilding infrastructure. The Launchpad is designed for solo developers and freelancers who need production‑ready patterns they can reuse across multiple client engagements.
**What's New in v3.4.1**
* Celery task discovery now matches the packaged `launchpad.worker` module path
* Azure OpenAI configuration uses pydantic-ai's `AzureProvider`
* The provider docs now reflect the currently supported runtime providers
* `pydantic-ai` is bumped to `>=1.94`
[View full changelog →](/docs/changelog/updates)
## Core Concept: Workflows and Nodes
Everything in GenAI Launchpad is built around one pattern: **Workflows** execute **Nodes** that pass data through a **TaskContext**.
```python theme={null}
class MyWorkflow(Workflow):
workflow_schema = WorkflowSchema(
event_schema=MyEventSchema,
start=AnalyzeNode,
nodes=[
NodeConfig(node=AnalyzeNode, connections=[GenerateNode]),
NodeConfig(node=GenerateNode, connections=[]),
],
)
# Run it
workflow = MyWorkflow(enable_tracing=True)
result = workflow.run({"message": "Hello, world!"})
```
* **Workflow**: Orchestrates execution of connected nodes
* **Node**: A processing unit (fetch data, call an LLM, route logic)
* **TaskContext**: Pydantic model passed between nodes containing event data and outputs
* **WorkflowSchema**: Defines the structure—which node starts, how they connect
## Multi-Provider LLM Support
GenAI Launchpad uses [PydanticAI](https://ai.pydantic.dev/) for LLM access. Switch providers by changing one line:
```python theme={null}
class MyAgentNode(AgentNode):
def get_agent_config(self) -> AgentConfig:
return AgentConfig(
model_provider=ModelProvider.OPENAI, # or ANTHROPIC, BEDROCK, OLLAMA, etc.
model_name="gpt-5.4-mini",
output_type=MyOutputSchema,
)
```
Supported providers: OpenAI, Azure OpenAI, Anthropic, Google Gemini, Google Vertex AI, AWS Bedrock, and Ollama.
## Production Infrastructure
The stack is pre-configured and ready to deploy:
* **FastAPI** - API endpoints that receive events
* **Celery + Redis** - Background task processing
* **PostgreSQL + pgvector** - Event persistence, results storage, and vector search
* **Supabase** - Optional Auth, realtime, storage, and Studio services
* **Langfuse** - LLM observability and tracing
* **Alembic** - Database migrations
* **Docker + Caddy** - Containerized deployment; Caddy is opt-in for HTTPS
Events flow through: API → Database → Celery Worker → Workflow → Results stored.
## What GenAI Launchpad Is Not
**Not an Agent Framework**: While you can build agent-like systems using our workflow architecture, GenAI Launchpad isn't primarily an agent framework like AutoGPT, CrewAI, or LangGraph. Instead, it provides the infrastructure to build any type of AI application. These frameworks can be integrated into the Launchpad alongside or instead of PydanticAI.
**Not Opinionated About AI Logic**: We don't dictate how you implement your AI logic:
* Use our built-in workflow system
* Integrate LangChain or LlamaIndex
* Build custom solutions
**Not a Closed System**: Every component is replaceable:
* Swap Redis for RabbitMQ
* Use different model providers
* Implement custom workflow processors
## Use Cases
The workflow/node pattern works well for:
* **Document Processing** - Chain nodes: extract → analyze → summarize → store
* **Customer Support** - Route node determines intent, specialized nodes handle responses
* **Content Generation** - Sequential nodes for research → outline → draft → refine
* **Data Pipelines** - Concurrent nodes process multiple data sources in parallel
Each use case maps naturally to a workflow with connected nodes, giving you traceability and the ability to modify individual steps without rewriting everything.
# License
Source: https://launchpad.datalumina.com/license
License terms and conditions for using the GenAI Launchpad
# Datalumina Launchpad License Agreement
**Copyright (c) 2024 Datalumina Solutions B.V.**
This License Agreement ("Agreement") is entered into between Datalumina Solutions B.V. ("the Company"),
whose contact information is [info@datalumina.com](mailto:info@datalumina.com), and you, the user ("Licensee"), regarding the use
of the GenAI Launchpad (the "Product") provided by the Company. By downloading, accessing, or using
the Product, Licensee agrees to be bound by the terms and conditions of this Agreement.
## 1. Grant of License
Subject to the terms and conditions of this Agreement, Datalumina Solutions B.V. grants Licensee
a non-exclusive, non-transferable, and non-sublicensable Individual License to use the GenAI
Launchpad for the following purposes:
* Create unlimited projects, both non-commercial and commercial.
* Build and develop applications or solutions for personal use, client work, or internal business purposes.
* Modify, customize, and expand upon the code to suit specific project needs.
## 2. Restrictions
Licensee shall not:
* Resell or redistribute the GenAI Launchpad as a standalone product or as part of a template,
package, or course where the primary value is in the project template itself.
* Remove, alter, or obscure any copyright, trademark, or other proprietary notices from the
GenAI Launchpad.
* Use the GenAI Launchpad in any way that violates applicable laws, regulations, or third-party rights.
* Sub-license, rent, lease, or transfer the GenAI Launchpad or any rights granted under this Agreement.
## 3. Ownership and Intellectual Property
Datalumina Solutions B.V. retains all ownership and intellectual property rights in and to the
GenAI Launchpad. This Agreement does not grant Licensee any ownership rights in the GenAI
Launchpad, beyond the right to use it as outlined in Section 1.
## 4. Warranty and Disclaimer
THE GENAI LAUNCHPAD IS PROVIDED "AS IS" WITHOUT WARRANTY OF ANY KIND, EITHER EXPRESS OR IMPLIED,
INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
PURPOSE, OR NONINFRINGEMENT.
## 5. Limitation of Liability
TO THE MAXIMUM EXTENT PERMITTED BY APPLICABLE LAW, DATALUMINA SOLUTIONS B.V. SHALL NOT BE LIABLE
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, CONSEQUENTIAL, OR PUNITIVE DAMAGES ARISING OUT OF
OR RELATING TO THE USE OR INABILITY TO USE THE GENAI LAUNCHPAD, EVEN IF DATALUMINA SOLUTIONS B.V.
HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH DAMAGES.
## 6. Governing Law and Jurisdiction
This Agreement shall be governed by and construed in accordance with the laws of the Netherlands,
without regard to its conflict of law principles. Any dispute arising out of or in connection with
this Agreement shall be subject to the exclusive jurisdiction of the courts located in the Netherlands.
## 7. Entire Agreement
This Agreement constitutes the entire agreement between Licensee and Datalumina Solutions B.V.
concerning the subject matter herein and supersedes all prior or contemporaneous agreements,
representations, warranties, and understandings.
***
**Last updated:** October 31, 2024
**Datalumina Solutions B.V.**\
**Contact Information:** [info@datalumina.com](mailto:info@datalumina.com)