Workflow Templates and Reusability Patterns
Defining every workflow from scratch is inefficient. Templates allow you to define a workflow once (e.g., "customer onboarding") and instantiate it multiple times with different parameters (different welcome emails, different service tiers). This article covers workflow templating, parameterization, versioning, and composition—the patterns that allow teams to scale without duplicating work.
What Is a Workflow Template?
A workflow template is a parameterized workflow definition. Instead of hardcoding values, you define placeholders (parameters) that are filled in when the workflow is instantiated. For example:
# Template: customer_onboarding
name: customer_onboarding
version: "1.0"
parameters:
- name: customer_tier
type: string
default: "basic"
- name: welcome_email_subject
type: string
default: "Welcome to Our Service!"
- name: notification_channel
type: string
default: "email"
steps:
- name: fetch_customer
type: tool
tool: fetch_customer_api
args:
customer_id: "{{ workflow.customer_id }}"
- name: send_welcome
type: tool
tool: send_email
args:
to: "{{ step.fetch_customer.email }}"
subject: "{{ params.welcome_email_subject }}"
body: "Welcome to our {{ params.customer_tier }} plan!"
- name: log_event
type: tool
tool: log_analytics
args:
event_name: "customer_onboarded"
tier: "{{ params.customer_tier }}"
channel: "{{ params.notification_channel }}"
When instantiated with different parameters, the same template produces different workflows:
# Instantiation A: Premium customer, Slack notification
instance_a = {
"customer_id": "CUST-111",
"customer_tier": "premium",
"welcome_email_subject": "Welcome to Premium!",
"notification_channel": "slack",
}
# Instantiation B: Free customer, email only
instance_b = {
"customer_id": "CUST-222",
"customer_tier": "free",
"welcome_email_subject": "Welcome!",
"notification_channel": "email",
}
Implementing Parameterized Templates
from typing import Any, Dict, List, Optional
from jinja2 import Template
import yaml
class WorkflowTemplate:
"""
A parameterized workflow template.
"""
def __init__(
self,
name: str,
version: str,
definition: Dict[str, Any],
):
self.name = name
self.version = version
self.definition = definition
self.parameters = definition.get("parameters", [])
def instantiate(self, parameters: Dict[str, Any]) -> Dict[str, Any]:
"""
Create a concrete workflow from this template.
Args:
parameters: Dict of parameter name -> value.
Returns:
A concrete workflow definition with parameters substituted.
"""
# Validate required parameters.
for param_def in self.parameters:
if param_def.get("required", False) and param_def["name"] not in parameters:
raise ValueError(f"Missing required parameter: {param_def['name']}")
# Fill in defaults.
if param_def["name"] not in parameters:
parameters[param_def["name"]] = param_def.get("default")
# Deep copy the template definition.
concrete = yaml.safe_load(yaml.safe_dump(self.definition))
# Recursively substitute parameters in the definition.
self._substitute_parameters(concrete, parameters)
return concrete
def _substitute_parameters(self, obj: Any, parameters: Dict[str, Any]):
"""Recursively substitute parameters in an object."""
if isinstance(obj, dict):
for key, value in obj.items():
if isinstance(value, str) and "{{" in value:
# This is a Jinja2 template; render it.
template = Template(value)
obj[key] = template.render(params=parameters)
else:
self._substitute_parameters(value, parameters)
elif isinstance(obj, list):
for i, item in enumerate(obj):
if isinstance(item, str) and "{{" in item:
template = Template(item)
obj[i] = template.render(params=parameters)
else:
self._substitute_parameters(item, parameters)
# Example: Load and instantiate a template
template_def = {
"name": "customer_onboarding",
"version": "1.0",
"parameters": [
{"name": "welcome_subject", "type": "string", "default": "Welcome!"},
{"name": "customer_tier", "type": "string", "required": True},
],
"steps": [
{
"name": "send_email",
"type": "tool",
"subject": "{{ params.welcome_subject }}",
"body": "Enjoy {{ params.customer_tier }}!",
}
],
}
template = WorkflowTemplate("customer_onboarding", "1.0", template_def)
concrete_workflow = template.instantiate({
"welcome_subject": "Premium Welcome!",
"customer_tier": "premium",
})
print(concrete_workflow["steps"][0]["subject"])
# Output: Premium Welcome!
Workflow Versioning
As templates evolve, you need to version them. Older workflows should continue running with the version they were defined with; new workflows use the latest version.
class TemplateRegistry:
"""
A registry of workflow templates, supporting multiple versions.
"""
def __init__(self):
self.templates: Dict[str, Dict[str, WorkflowTemplate]] = {} # name -> version -> template
def register(self, template: WorkflowTemplate):
"""Register a template."""
if template.name not in self.templates:
self.templates[template.name] = {}
self.templates[template.name][template.version] = template
print(f"Registered template {template.name}:{template.version}")
def get(self, name: str, version: str = None) -> WorkflowTemplate:
"""
Get a template by name and version.
If version is None, return the latest.
"""
if name not in self.templates:
raise KeyError(f"Template not found: {name}")
versions = self.templates[name]
if version is None:
# Return latest version.
version = max(versions.keys(), key=lambda v: self._parse_version(v))
if version not in versions:
raise KeyError(f"Template version not found: {name}:{version}")
return versions[version]
@staticmethod
def _parse_version(v: str) -> tuple:
"""Parse a version string into a tuple for comparison."""
return tuple(map(int, v.split(".")))
# Example: Register templates with versions
registry = TemplateRegistry()
template_v1 = WorkflowTemplate("approval_flow", "1.0", {
"name": "approval_flow",
"steps": [
{"name": "fetch_data", "type": "tool"},
{"name": "approve", "type": "approval_gate"},
],
})
template_v2 = WorkflowTemplate("approval_flow", "2.0", {
"name": "approval_flow",
"steps": [
{"name": "fetch_data", "type": "tool"},
{"name": "validate_data", "type": "llm"}, # New step in v2
{"name": "approve", "type": "approval_gate"},
],
})
registry.register(template_v1)
registry.register(template_v2)
# Old workflows use v1, new workflows use v2.
old_workflow_def = registry.get("approval_flow", "1.0").definition
new_workflow_def = registry.get("approval_flow").definition # Latest version (2.0)
Template Composition: Sub-Workflows
Complex workflows are built by composing simpler templates:
class CompositeTemplate:
"""
A template that composes multiple sub-templates.
"""
def __init__(
self,
name: str,
version: str,
sub_templates: List[tuple[str, Dict[str, Any]]], # (template_name, parameters)
):
self.name = name
self.version = version
self.sub_templates = sub_templates
def instantiate(
self,
registry: TemplateRegistry,
parameters: Dict[str, Any],
) -> List[Dict[str, Any]]:
"""
Instantiate all sub-templates and compose them.
Args:
registry: The template registry.
parameters: Parameters for this composite workflow.
Returns:
A list of concrete workflows (sub-workflows).
"""
concrete_workflows = []
for sub_template_name, sub_params in self.sub_templates:
# Resolve parameter placeholders in sub-params.
resolved_params = {}
for key, value in sub_params.items():
if isinstance(value, str) and "{{" in value:
template = Template(value)
resolved_params[key] = template.render(
params=parameters,
global_params=parameters,
)
else:
resolved_params[key] = value
# Get the sub-template and instantiate it.
sub_template = registry.get(sub_template_name)
concrete = sub_template.instantiate(resolved_params)
concrete_workflows.append(concrete)
return concrete_workflows
# Example: Composite "customer_lifecycle" = onboarding + send_email + cleanup
composite = CompositeTemplate(
name="customer_lifecycle",
version="1.0",
sub_templates=[
("customer_onboarding", {"customer_tier": "{{ params.tier }}"}),
("send_email", {"recipient": "{{ params.email }}"}),
("log_event", {"event_name": "lifecycle_complete"}),
],
)
registry = TemplateRegistry() # Populated with sub-templates
lifecycle_workflows = composite.instantiate(
registry,
{"tier": "premium", "email": "[email protected]"},
)
Template Configuration Store
Store templates in a database or cloud storage:
from abc import ABC, abstractmethod
class TemplateStore(ABC):
"""Abstract interface for storing and retrieving templates."""
@abstractmethod
async def save(self, template: WorkflowTemplate):
pass
@abstractmethod
async def load(self, name: str, version: str) -> WorkflowTemplate:
pass
@abstractmethod
async def list_versions(self, name: str) -> List[str]:
pass
class DatabaseTemplateStore(TemplateStore):
"""Store templates in a database."""
def __init__(self, db):
self.db = db
async def save(self, template: WorkflowTemplate):
"""Save a template."""
await self.db.collection("templates").insert_one({
"name": template.name,
"version": template.version,
"definition": template.definition,
"created_at": datetime.utcnow(),
})
async def load(self, name: str, version: str) -> WorkflowTemplate:
"""Load a template."""
doc = await self.db.collection("templates").find_one({
"name": name,
"version": version,
})
if not doc:
raise KeyError(f"Template not found: {name}:{version}")
return WorkflowTemplate(
name=doc["name"],
version=doc["version"],
definition=doc["definition"],
)
async def list_versions(self, name: str) -> List[str]:
"""List all versions of a template."""
docs = await self.db.collection("templates").find({
"name": name,
}).sort("version", -1).to_list(length=None)
return [doc["version"] for doc in docs]
# Usage:
store = DatabaseTemplateStore(db)
await store.save(template)
loaded = await store.load("customer_onboarding", "1.0")
Key Takeaways
- Workflow templates are parameterized definitions that can be instantiated with different values.
- Use Jinja2 templates to substitute parameters in workflow definitions.
- Version templates so old workflows continue using their original version.
- Compose complex workflows by chaining simpler templates.
- Store templates in a database or versioned repository for teams to discover and reuse.
- Validate required parameters at instantiation time; provide sensible defaults.
Frequently Asked Questions
Can I update a template without breaking existing workflows?
Yes, if you increment the version. Existing workflows reference a specific version and continue using it. New workflows use the new version. This prevents breaking changes.
How do I share templates across teams?
Publish templates to a central registry (database, artifact repository, template library). Include documentation (README), examples, and version history. Make discovery easy via search/tagging.
Can a template refer to another template's output?
Yes, if templates are composed. A composite template chains sub-templates and passes outputs from one sub-template as inputs to the next. Alternatively, a template can use a "call sub-workflow" step that waits for the sub-workflow to complete.
What if a template has optional parameters?
Mark parameters as required: False and provide a default value. At instantiation, omit the parameter and it uses the default. This makes templates flexible for different use cases.
How do I test a template before deploying it?
Instantiate the template with test parameters, execute it in a staging environment, and verify the output. Use unit tests on template rendering (parameter substitution) and integration tests on the full workflow.