Helix User Guide¶
Helix is a high-performance AI orchestration platform for Go that combines agentic AI, RAG pipelines, and enterprise integration patterns into a single, unified runtime. Native Go, lock-free, and AI-native.
Table of Contents¶
- Quick Start
- Core Concepts
- CLI Reference
- AI-First YAML DSL
- Workflow Definition (YAML)
- Go Fluent DSL
- Components Reference
- Enterprise Integration Patterns
- Resilience Patterns
- Data Transformation (Starlang)
- REST DSL and OpenAPI
- AI Agents
- RAG Pipelines
- Connectors (440+)
- Security
- Cluster and High Availability
- Configuration Reference
- Observability
- Performance Tuning
- Cookbook
- Large File & Streaming Support
1. Quick Start¶
Install¶
Helix ships as a single self-contained binary. Put helix on your $PATH and point it at your license file:
Create a Project¶
This creates:
- helix.yaml — configuration
- flow.yaml — example route
Run¶
Minimal Flow¶
workflows:
- id: hello-world
trigger:
uri: timer://hello?period=5s
steps:
- transform:
constant: "Hello from Helix!"
- log:
message: "Message received"
- target: log://output?showBody=true
This fires every 5 seconds, sets the body to a string, logs it, and sends it to the log component.
2. Core Concepts¶
Exchange¶
The Exchange is the fundamental data envelope that flows through workflows. Every message passing through Helix is wrapped in an Exchange.
┌─────────────────────────────────────┐
│ Exchange │
│ │
│ ID: "exchange-42" │
│ Pattern: InOnly / InOut │
│ Created: 2026-03-05T10:00:00Z │
│ RouteID: "order-processing" │
│ │
│ ┌───────────────────────────────┐ │
│ │ In (Message) │ │
│ │ Body: any (the payload) │ │
│ │ Headers: map[string]any │ │
│ └───────────────────────────────┘ │
│ │
│ Properties: map[string]any │
│ Exception: error (if failed) │
└─────────────────────────────────────┘
- Body — the message payload (any Go type)
- Headers — per-message metadata (Content-Type, routing keys, etc.)
- Properties — per-exchange metadata (survives the whole route)
- Exception — set when processing fails
Workflow¶
A Workflow defines a message flow: where messages come from, what steps to apply, and where to send them to.
Component¶
A Component is a factory for endpoints. Each component has a scheme (e.g., kafka, http, postgres) and knows how to create Producers (send messages) and Consumers (receive messages).
URI Format¶
All endpoints use URI notation:
Examples:
kafka://orders?brokers=kafka:9092&group=order-service
http://api.example.com/v1/users?method=POST&timeout=10s
postgres://orders/insert?dsn=postgres://localhost:5432/db
timer://tick?period=5s&repeatCount=100
3. CLI Reference¶
| Command | Description |
|---|---|
run <file\|dir> |
Run flow definitions (YAML, JSON, or directory) |
run --watch <file\|dir> |
Run with hot reload (watches for file changes) |
debug <file\|dir> |
Run with verbose debug logging |
validate <file> |
Validate a flow definition without running |
export <file> |
Export a flow as a Dockerfile |
components |
List all available components |
init <name> |
Initialize a new Helix project |
version |
Show version |
Hot Reload¶
File changes are detected automatically. Workflows are atomically replaced — in-flight exchanges complete before the old route stops.
Export to Docker¶
helix export flow.yaml --format=docker --output=Dockerfile
docker build -t my-flow .
docker run my-flow
AI-First YAML DSL¶
Helix provides five top-level YAML primitives. The AI-first primitives (agents:, tools:, pipelines:, mcps:) sit alongside the traditional workflows: tag. These are the recommended way to define AI agents, schema-validated tools, data pipelines, and MCP servers.
agents: — AI Agents¶
Define standalone AI agents with a trigger, model, tools, and memory:
agents:
- id: support-bot
type: react
trigger: http-server://0.0.0.0:8080/chat
model:
provider: anthropic
model: claude-sonnet-4-20250514
system_prompt: "You are a customer support agent."
tools:
- name: search_docs
type: function
description: "Search the knowledge base"
memory:
enabled: true
type: sliding-window
window_size: 20
approval_policy: tagged
approval_tools: [refund-payment]
Agent tool types:
| Type | Description |
|---|---|
function |
Generic function tool (Go code) |
tool |
Reference a top-level tools: or pipelines: definition by ID |
a2a |
Call an external A2A agent with url: "..." |
mcp |
Import tools from an external MCP server |
connector |
Invoke any Helix connector by URI (439 integrations: HTTP REST, SaaS, databases, messaging) |
Tools of type a2a, mcp, and connector support an auth: block for outbound authentication (bearer, api-key, oauth2, mtls). See the AI Agents Guide -- Tool Types Reference for full YAML examples.
See the AI Agents Guide for all agent types and configuration options.
tools: — Self-Describing Tools¶
Define tools with mandatory input and output schemas. Every field -- description, input, and output -- is required; the parser rejects tools missing any of them. Tools auto-generate a tool://{id} URI (no from: needed), validate input before execution, and are auto-discoverable by MCP with full schemas.
tools:
- id: search
description: "Search knowledge base"
input:
query: { type: string, description: "Search query", required: true }
limit: { type: integer, description: "Max results", default: 10 }
output:
results: { type: array, description: "Matching documents" }
total: { type: integer, description: "Total matches" }
steps:
- transform:
starlang: |
q = payload["query"]
limit = payload.get("limit", 10)
write({"results": [], "total": 0}, "application/json")
Mandatory fields:
| Field | Purpose |
|---|---|
id |
Unique identifier; becomes the tool://{id} URI |
description |
Human-readable description; exposed to MCP clients |
input |
Input schema with field types, descriptions, required flags, and defaults |
output |
Output schema; stored in x-helix-output-schema extension |
steps |
Processing pipeline (same step types as workflows) |
How it works:
- The parser validates that
description,input, andoutputare all present - A
tool://searchendpoint is auto-generated (nofrom:needed) - When invoked, input is validated against the
input:schema before any step runs - MCP discovers the tool with full input and output schema metadata
- The output schema is available via the
x-helix-output-schemaextension
Tools can be exposed to MCP clients, called from workflows via target: tool://search, or used as agent function tools.
pipelines: — Data Processing Flows¶
Define RAG ingestion, ETL, or other data processing flows:
pipelines:
- id: knowledge-base
source: file://./docs
steps:
- rag-chunk:
strategy: semantic
threshold: 0.5
- rag-embed:
provider: openai
model: text-embedding-3-small
- rag-store:
store: pgvector
connection: "postgres://user:pass@db:5432/rag"
- rag-retrieve:
strategy: hybrid
top_k: 10
- rag-synthesize:
mode: compact
model:
provider: openai
model: gpt-4o
output: text
See the RAG Guide for all pipeline stages and options.
mcps: — MCP Servers¶
Expose agents, pipelines, and workflows as MCP tools for Claude Desktop, Cursor, or any MCP client:
When expose is omitted, all agents, pipelines, and workflows are exposed. See the MCP Guide for transport options, Claude Desktop setup, and advanced configuration.
workflows: — Integration Workflows¶
The traditional workflows: tag remains the right choice for enterprise integration patterns, Kafka consumers, content-based routing, and any flow that is not purely an AI agent or data pipeline:
workflows:
- id: order-router
trigger:
uri: kafka://orders?brokers=localhost:9092&group=helix
steps:
- choice:
when:
- starlang: 'read(payload, "application/json")["priority"] == "high"'
steps:
- target: kafka://high-priority-orders?brokers=localhost:9092
otherwise:
steps:
- target: kafka://default-orders?brokers=localhost:9092
All five tags can coexist in the same YAML file. Internally, agents:, tools:, and pipelines: are reified into workflows by the DSL loader.
4. Workflow Definition (YAML)¶
Structure¶
workflows:
- id: route-name # unique identifier
description: "What this workflow does"
singleton: true # only run on one cluster node (optional)
auto-startup: true # start automatically (default: true)
trigger:
uri: scheme://path?params # where messages come from
auth: # optional authentication (runs before steps)
type: jwt # apikey, jwt, basic, oauth2
jwt:
secret: "${env:JWT_SECRET}"
issuer: my-auth-server
steps: # processing pipeline
- step1: { ... }
- step2: { ... }
error-handler: # optional error handling
dead-letter-channel: log://dlq
max-redeliveries: 3
redelivery-delay: "1s"
backoff-multiplier: 2.0
Basic Steps¶
to — Send to Endpoint¶
log — Log a Message¶
transform — Transform the Body¶
# Starlang expression
- transform:
starlang: |
payload.upper()
# From file
- transform:
starlang-file: transforms/enrich.star
# Constant
- transform:
constant: '{"status": "ok"}'
set-header — Set a Header¶
- set-header:
name: Content-Type
constant: application/json
- set-header:
name: priority
starlang: |
"high" if payload.total > 1000 else "normal"
set-body — Set the Body¶
remove-header / remove-headers¶
- remove-header: X-Internal-Token
- remove-headers:
pattern: "X-Debug-*"
except:
- X-Debug-RequestId
marshal / unmarshal — Data Format Conversion¶
validate — Validate Message¶
bean — Call a Registered Service¶
process — Custom Processing¶
# Reference a registered processor
- process:
ref: myCustomProcessor
# Inline Starlang processor
- process:
starlang: |
payload["processed_at"] = now()
payload["version"] = 2
wire-tap — Copy to Side Channel¶
enrich / poll-enrich — Data Enrichment¶
# Synchronous enrichment
- enrich:
uri: http://user-service/api/users/${header.userId}
strategy: mergeBodyStrategy
# Poll-based enrichment with timeout
- poll-enrich:
uri: file://incoming/?include=*.csv
timeout: "10s"
convert-body — Type Conversion¶
set-property / remove-property / remove-properties¶
- set-property:
name: stage
constant: "validated"
- remove-property: tempData
- remove-properties:
pattern: "temp.*"
except: [tempImportant]
set-exchange-pattern — Change Exchange Pattern¶
throw-exception — Raise an Error¶
to-dynamic — Dynamic Endpoint Resolution¶
script — Inline Starlang Script¶
do-try — Try/Catch/Finally Error Handling¶
- do-try:
steps:
- target: http://external-api/process
do-catch:
- exceptions: ["timeout", "connection"]
steps:
- log: { message: "Retryable error caught" }
do-finally:
- log: { message: "Cleanup complete" }
5. Go Fluent DSL¶
For programmatic route building:
package main
import (
"github.com/arvenza/helix/pkg/dsl"
"github.com/arvenza/helix/pkg/engine"
)
func main() {
h := engine.NewContext("my-app")
// Register components...
h.AddComponent("direct", direct.New())
h.AddComponent("kafka", kafka.New())
// Define workflows using the fluent builder
route := dsl.NewRoute().
ID("order-router").
Description("Routes orders by priority").
From("kafka://orders?brokers=localhost:9092&group=order-svc").
Transform(`read(payload, "application/json")`).
Choice().
When(`payload.priority == "high"`).
To("direct://fast-lane").
When(`payload.priority == "low"`).
To("seda://batch-queue").
Otherwise().
To("direct://normal-processing").
End().
Build()
// Reify and add to engine
reifier := dsl.NewReifier(h)
rt, _ := reifier.Reify(route)
h.AddRoute(rt)
h.Start(context.Background())
}
Builder Methods Summary¶
| Category | Methods |
|---|---|
| Lifecycle | NewRoute(), ID(), Description(), From(), Build() |
| Routing | To(), Choice().When().Otherwise().End(), Filter().End() |
| Transform | Transform(), TransformFile(), TransformConstant() |
| Headers | SetHeader(), SetHeaderConstant(), RemoveHeader(), RemoveHeaders() |
| Body | SetBody() |
| Processing | Process(), ProcessStarlang(), Bean(), Validate() |
| Logging | Log(), LogLevel() |
| Data | Marshal(), Unmarshal() |
| Orchestration | Split(), SplitParallel(), Aggregate(), Multicast() |
| Routing | RecipientList(), RoutingSlip(), DynamicRouter(), LoadBalancer() |
| Enrichment | Enrich(), EnrichWithStrategy(), PollEnrich(), WireTap() |
| Resilience | CircuitBreaker(), Throttle(), Delay(), Retry(), Timeout(), Loop() |
| Advanced | OnCompletion(), StepGroup(), Normalize() |
| Error | OnException(), OnExceptionWithRetry() |
All nested builders (Choice, Split, Filter, etc.) use .End() to return to the parent scope.
6. Components Reference¶
Core Components¶
| Scheme | Type | Producer | Consumer | Description |
|---|---|---|---|---|
direct |
Sync | Yes | Yes | Synchronous in-process messaging |
seda |
Async | Yes | Yes | Channel-based async messaging with backpressure |
timer |
Trigger | No | Yes | Periodic message generation |
cron |
Trigger | No | Yes | Cron-expression scheduling |
mock |
Test | Yes | Yes | Test assertions and verification |
log |
Sink | Yes | No | Structured logging output |
direct://name¶
Synchronous, in-process endpoint. Producer calls consumer directly.
seda://name¶
Asynchronous in-process with Go channels.
| Parameter | Default | Description |
|---|---|---|
bufferSize |
1000 | Channel buffer capacity |
concurrency |
1 | Worker goroutine pool size |
overflowPolicy |
fail | fail, block, drop, newest |
timer://name¶
| Parameter | Default | Description |
|---|---|---|
period |
1s | Interval between fires |
delay |
0 | Initial delay |
repeatCount |
0 | Total fires (0 = infinite) |
fixedRate |
true | Fixed-rate vs fixed-delay |
cron://name¶
trigger: { uri: "cron://daily-report?schedule=0 9 * * *" }
trigger: { uri: "cron://frequent?schedule=@every 30s" }
Supports: 5-field cron expressions, @yearly, @monthly, @weekly, @daily, @hourly, @every <duration>.
mock://name¶
Rich assertion endpoint for testing.
In Go tests:
mock := mock.New()
endpoint := mock.GetEndpoint("result")
endpoint.ExpectedMessageCount(5)
endpoint.ExpectedBodiesReceived("a", "b", "c")
err := endpoint.AssertIsSatisfied(5 * time.Second)
if err != nil {
t.Fatal(err)
}
Network Components¶
http://host/path (Client)¶
| Parameter | Default | Description |
|---|---|---|
method |
GET | HTTP method |
timeout |
30s | Request timeout |
Override via headers: HelixHttpMethod, HelixHttpPath.
http-server://addr/path (Server)¶
Creates an HTTP listener. Request body becomes the exchange body; exchange body after processing becomes the HTTP response.
grpc://host:port/service.Method¶
| Parameter | Default | Description |
|---|---|---|
protoFile |
Path to .proto file (optional) | |
plaintext |
false | Use h2c instead of TLS |
ws://host/path (WebSocket Client) / ws-server://addr/path (Server)¶
# Client
- target: ws://ws-service:8080/stream
# Server
trigger:
uri: ws-server://0.0.0.0:9090/events
Features: auto-reconnect with backoff, text/binary frames, broadcast, client targeting via ws.clientId header.
Messaging Components¶
kafka://topic¶
trigger:
uri: kafka://orders?brokers=kafka1:9092,kafka2:9092&group=order-service&autoOffsetReset=earliest
steps:
- target: kafka://processed-orders?brokers=kafka1:9092
| Parameter | Default | Description |
|---|---|---|
brokers |
localhost:9092 | Comma-separated broker list |
group |
Consumer group ID (required for consumer) | |
autoOffsetReset |
latest | earliest or latest |
Headers set by consumer: kafka.KEY, kafka.TOPIC, kafka.PARTITION, kafka.OFFSET, kafka.TIMESTAMP.
nats://subject¶
| Parameter | Default | Description |
|---|---|---|
servers |
nats://localhost:4222 | Server list |
queue |
Queue group for load-balanced consumption |
amqp://queue¶
trigger:
uri: amqp://order-queue?url=amqp://guest:guest@rabbit:5672/&exchange=orders&routingKey=new&durable=true
| Parameter | Default | Description |
|---|---|---|
url |
amqp://guest:guest@localhost:5672/ | Broker URL |
exchange |
Exchange name | |
routingKey |
Routing key | |
durable |
false | Queue durability |
autoAck |
false | Auto-acknowledge |
redis://channel¶
# Pub/Sub
trigger: { uri: "redis://notifications?addr=redis:6379" }
- target: "redis://alerts?addr=redis:6379"
# Key-Value
- target: "redis://cache-key?addr=redis:6379&operation=set"
- target: "redis://cache-key?addr=redis:6379&operation=get"
| Parameter | Default | Description |
|---|---|---|
addr |
localhost:6379 | Redis address |
password |
Auth password | |
db |
0 | Database number |
operation |
pubsub | pubsub, set, get, del, lpush, rpush, lpop, rpop, hset, hget |
sqs://queue-name¶
Database Components¶
postgres://table/operation¶
# Insert
- target: "postgres://orders/insert?dsn=postgres://user:pass@localhost:5432/mydb"
# Query
trigger:
uri: "postgres://orders/query?dsn=postgres://localhost:5432/mydb&pollInterval=30s"
# Custom SQL via header
- set-header:
name: postgres.sql
constant: "SELECT * FROM orders WHERE status = 'pending' LIMIT 100"
- target: "postgres://orders/query?dsn=postgres://localhost:5432/mydb"
| Parameter | Default | Description |
|---|---|---|
dsn |
postgres://localhost:5432 | Connection string |
operation |
query | query, insert, update, delete, execute |
pollInterval |
30s | Consumer polling interval |
Headers: postgres.sql (override query), postgres.where (filter), postgres.rowCount, postgres.rowsAffected.
mongodb://collection/operation¶
Operations: find, findOne, insert, update, delete, aggregate. Consumer uses change streams.
elasticsearch://index/operation¶
Storage Components¶
file://directory¶
# Consumer — poll for new files
trigger:
uri: "file://incoming/?include=*.csv&recursive=true&poll=10s&move=.processed"
# Producer — write files
- target: "file://output/"
| Parameter | Default | Description |
|---|---|---|
include |
Glob pattern (e.g., *.json) |
|
recursive |
false | Recurse subdirectories |
poll |
5s | Polling interval |
move |
Move processed files to this directory | |
delete |
false | Delete after processing |
Headers: HelixFileName, HelixFilePath, HelixFileLength.
s3://bucket/key¶
# Upload
- target: "s3://my-bucket/reports/daily.csv?region=us-east-1&accessKey=${AWS_KEY}&secretKey=${AWS_SECRET}"
# Download
- target: "s3://my-bucket/data.json?operation=getObject"
# List (consumer polls for new objects)
trigger:
uri: "s3://my-bucket?operation=listObjects&pollInterval=60s"
Operations: putObject, getObject, listObjects, deleteObject, copyObject. Works with any S3-compatible storage (MinIO, etc.) via endpoint parameter.
ftp://host/path / sftp://host/path¶
- target: "ftp://ftp.example.com:21/uploads?username=user&password=pass"
- target: "sftp://sftp.example.com:22/data?username=user&privateKey=/path/to/key"
AI / LLM Components¶
openai://operation¶
# Chat completion
- target: "openai://chat?model=gpt-4o&apiKey=${OPENAI_API_KEY}&temperature=0.7&maxTokens=4096"
# Embeddings
- target: "openai://embeddings?model=text-embedding-3-small&apiKey=${OPENAI_API_KEY}"
Set system prompt via systemPrompt header. Token usage reported in openai.usage.* headers.
anthropic://chat¶
ollama://operation¶
# Local LLM
- target: "ollama://chat?model=llama3&baseURL=http://localhost:11434"
- target: "ollama://embeddings?model=nomic-embed-text"
300-second timeout for local model inference.
agent://agent-id¶
# Pre-registered agent
- target: agent://support-agent
# Ad-hoc agent trigger URI
- target: "agent://ad-hoc?type=react&model=openai&modelName=gpt-4o&apiKey=${OPENAI_API_KEY}&systemPrompt=You are helpful"
Seven agent types: react, supervisor, plan-execute, deep, sequential, parallel, loop.
Vector Database Components¶
milvus://collection¶
- target: "milvus://documents/insert?addr=localhost:19530"
- target: "milvus://documents/search?addr=localhost:19530&topK=10"
qdrant://collection¶
- target: "qdrant://knowledge-base/upsert?url=http://localhost:6333"
- target: "qdrant://knowledge-base/search?url=http://localhost:6333&topK=5"
pinecone://index¶
Other Components¶
docker://operation¶
- target: "docker://containers/list?host=unix:///var/run/docker.sock"
- target: "docker://containers/create"
- target: "docker://images/pull"
Operations: containers/list, containers/create, containers/start, containers/stop, containers/logs, containers/inspect, images/list, images/pull, exec.
smtp://host¶
- set-header: { name: To, constant: "team@example.com" }
- set-header: { name: Subject, constant: "Daily Report" }
- target: "smtp://mail.example.com:587?from=noreply@example.com&username=user&password=pass&starttls=true"
slack://mode¶
# Webhook
- target: "slack://webhook?url=https://hooks.slack.com/services/xxx"
# Bot API
- target: "slack://channel?token=xoxb-xxx&channel=C1234567"
# Events API (consumer)
trigger:
uri: "slack://events?token=xoxb-xxx&signingSecret=xxx&port=3000"
websocket¶
# Server (receive connections)
trigger: { uri: ws-server://0.0.0.0:9090/events }
# Client (connect out)
- target: ws://echo.websocket.org/
mcp-server:// and mcp://¶
# Expose workflows as MCP tools
trigger:
uri: "mcp-server://0.0.0.0:3000?transport=http&name=helix-tools"
# Call an external MCP server
- target: "mcp://localhost:3001/search-tool?transport=http"
a2a-server:// and a2a://¶
# Expose agents via A2A protocol (Google Agent-to-Agent)
trigger:
uri: "a2a-server://0.0.0.0:9000?name=my-agent&description=My+Agent"
# Call an external A2A agent
- target: "a2a://remote-host:9000/skill-name"
agui://¶
# Expose agent via AG-UI protocol (CopilotKit streaming)
trigger:
uri: "agui://0.0.0.0:4100/agent?name=my-chatbot"
a2ui://¶
# Expose agent via A2UI protocol (Google declarative UI)
trigger:
uri: "a2ui://0.0.0.0:4200/surface?name=my-ui"
7. Enterprise Integration Patterns¶
Content-Based Router (Choice)¶
- choice:
when:
- starlang: 'payload.type == "order"'
steps:
- target: direct://orders
- starlang: 'payload.type == "invoice"'
steps:
- target: direct://invoices
otherwise:
steps:
- target: direct://unknown
Message Filter¶
Splitter¶
# Split an array into individual messages
- split:
starlang: 'payload.items'
parallel: true
steps:
- target: direct://process-item
# With streaming (process items as they arrive)
- split:
starlang: 'payload.lines'
streaming: true
stop-on-exception: true
Properties set per split: HelixSplitIndex, HelixSplitSize, HelixSplitComplete.
Aggregator¶
- aggregate:
correlation: 'payload.customerId'
strategy: listStrategy
completion-size: 10
completion-timeout: "30s"
steps:
- target: direct://batch-process
Multicast¶
# Send to multiple endpoints in parallel
- multicast:
uris:
- direct://audit
- direct://analytics
- kafka://events
parallel: true
Recipient List¶
# Dynamic recipients from expression
- recipient-list:
starlang: 'payload.targets' # returns ["direct://a", "kafka://b"]
parallel: true
delimiter: ","
Routing Slip¶
# Sequential routing through dynamic list
- routing-slip:
starlang: 'headers["routingSlip"]'
delimiter: ","
Dynamic Router¶
# Iterative routing — called repeatedly until returns empty
- dynamic-router:
starlang: |
idx = properties.get("HelixDynamicRouterIndex", 0)
routes = ["direct://step1", "direct://step2", "direct://step3"]
routes[idx] if idx < sizeOf(routes) else ""
Load Balancer¶
- load-balancer:
strategy: roundRobin # roundRobin, random, weighted, failover, sticky
uris:
- http://server1:8080/api
- http://server2:8080/api
- http://server3:8080/api
Wire Tap¶
# Fire-and-forget copy to a side channel
- wiretap: kafka://audit-log
- target: direct://continue-main-flow
Idempotent Consumer¶
Normalizer¶
- normalize:
conditions:
- starlang: 'headers["Content-Type"] == "application/xml"'
steps:
- unmarshal: { format: xml }
- transform: { starlang: 'xml_to_canonical(payload)' }
- starlang: 'headers["Content-Type"] == "text/csv"'
steps:
- unmarshal: { format: csv }
- transform: { starlang: 'csv_to_canonical(payload)' }
On Completion¶
- on-completion:
on-complete:
- log: { message: "Success!" }
on-failure:
- target: kafka://error-notifications
steps:
- target: http://external-api/process
Step Group¶
- step:
id: "validation-phase"
steps:
- validate: 'payload.orderId != ""'
- validate: 'payload.amount > 0'
Claim Check (Store/Retrieve Large Payloads)¶
The Claim Check pattern replaces a message body with a lightweight key, storing the original payload in an external store. This allows large payloads to travel through the workflow as small references, reducing memory pressure and network bandwidth. The payload is retrieved later when needed.
Two operations are available: ClaimCheckIn stores the body and replaces it with a key, and ClaimCheckOut retrieves the stored body using the key.
YAML example:
workflows:
- id: claim-check-demo
trigger:
uri: direct://large-payload
steps:
- claim-check:
operation: push # store body, replace with key
key: 'headers["orderId"]'
- target: direct://lightweight-processing
- claim-check:
operation: pop # retrieve original body by key
key: 'headers["orderId"]'
- target: direct://final-processing
Go builder example:
import (
"github.com/arvenza/helix/pkg/processor"
"github.com/arvenza/helix/pkg/expression"
)
store := processor.NewInMemoryClaimCheckStore()
keyExpr := expression.NewStarlang("headers[\"orderId\"]")
// Store body, replace with key
claimIn := processor.NewClaimCheckIn(store, keyExpr)
// Retrieve body by key
claimOut := processor.NewClaimCheckOut(store, keyExpr)
// Retrieve without removing from store (peek)
claimGet := processor.NewClaimCheckGet(store, keyExpr, false /* remove */)
claimGet.Merge = true // set as header "HelixClaimCheckData" instead of replacing body
Key parameters:
| Parameter | Description |
|---|---|
operation |
push (store body, replace with key), pop (retrieve body by key), get (retrieve without removing), set (store with explicit key) |
key |
Starlang expression that evaluates to the claim check key |
Store implementations: InMemoryClaimCheckStore (thread-safe in-memory map), or implement the ClaimCheckStore interface for Redis/database backends.
Idempotent Consumer (Duplicate Detection) -- Expanded¶
The Idempotent Consumer pattern filters out duplicate messages using a unique message identifier. When a message arrives, the ID is checked against a repository. If the ID already exists, the message is silently dropped (or marked as duplicate). This is critical for exactly-once processing semantics.
Helix provides multiple repository backends: in-memory with TTL and bounded size, Redis-backed for distributed deduplication, and a context-aware variant.
YAML example:
workflows:
- id: dedup-orders
trigger:
uri: kafka://orders
steps:
- idempotent-consumer:
starlang: 'headers["messageId"]'
repository: redis # or "memory"
- target: direct://process-order
Go builder example:
import (
"github.com/arvenza/helix/pkg/processor"
"github.com/arvenza/helix/pkg/expression"
"time"
)
// In-memory repository with max 10000 keys and 1-hour TTL
repo := processor.NewInMemoryIdempotentRepository(10000, time.Hour)
keyExpr := expression.NewStarlang("headers[\"messageId\"]")
idem := processor.NewIdempotentConsumerProcessor(
repo,
keyExpr,
nextProcessor, // inner processor for non-duplicate messages
true, // skipDuplicate: true = silently drop, false = set property
)
Redis-backed repository for distributed deduplication:
import "github.com/redis/go-redis/v9"
redisClient := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
redisRepo := processor.NewRedisIdempotentRepository(redisClient, "dedup:", 24*time.Hour)
Key parameters:
| Parameter | Description |
|---|---|
starlang |
Expression that extracts a unique message ID from the exchange |
repository |
memory (in-memory with TTL) or redis (distributed via Redis SET NX with TTL) |
Repository implementations:
| Implementation | Description |
|---|---|
InMemoryIdempotentRepository |
Bounded size, TTL-based expiry, automatic eviction of oldest entries |
RedisIdempotentRepository |
Distributed, uses Redis SET NX with TTL for atomic dedup across cluster nodes |
MemoryIdempotentRepositoryV2 |
Context-aware variant using sync.Map with DistributedIdempotentRepository interface |
Properties set: HelixDuplicateMessage is set to true when a duplicate is detected and skipDuplicate is false.
Resequencer (Out-of-Order Messages)¶
The Resequencer pattern reorders messages that arrive out of sequence. It buffers incoming messages and releases them in the correct order based on a sequence number header. A configurable timeout ensures that missing sequence numbers are eventually skipped so buffered messages are not held indefinitely.
YAML example:
workflows:
- id: reorder-events
trigger:
uri: kafka://events
steps:
- resequence:
starlang: 'headers["HelixSequenceNumber"]'
batch-size: 100
timeout: "5s"
- target: direct://ordered-processing
Go builder example:
import (
"github.com/arvenza/helix/pkg/processor"
"time"
)
reseq := processor.NewResequencer(nextProcessor, 5*time.Second)
reseq.SequenceKey = "HelixSequenceNumber" // header key containing the sequence number
reseq.Capacity = 1000 // max buffered messages (0 = unlimited)
// Messages must have a sequence number header:
// ex.In().SetHeader("HelixSequenceNumber", 42)
// Clean shutdown drains buffered messages in order
defer reseq.Stop()
Key parameters:
| Parameter | Description |
|---|---|
starlang |
Expression for the sequence key (typically a header reference) |
batch-size |
Maximum number of messages to buffer (0 = unlimited) |
timeout |
Duration to wait for a missing sequence number before skipping it |
Behavior: Messages arriving in order are forwarded immediately. Out-of-order messages are buffered. When the next expected sequence number arrives, it and all consecutive buffered messages are released together. On timeout, the resequencer skips to the lowest buffered sequence number and releases the consecutive run. Implements StoppableProcessor for clean shutdown -- the background goroutine is stopped and all buffered messages are drained in order.
Saga (Distributed Transactions)¶
The Saga pattern orchestrates a sequence of steps as a long-running transaction. If any step fails, compensating actions are executed in reverse order for all previously completed steps, providing eventual consistency without distributed locks.
YAML example:
workflows:
- id: order-saga
trigger:
uri: direct://create-order
steps:
- saga:
steps:
- target: direct://reserve-inventory
- target: direct://charge-payment
- target: direct://send-confirmation
compensate:
- target: direct://release-inventory
- target: direct://refund-payment
- target: direct://cancel-confirmation
Go builder example:
import "github.com/arvenza/helix/pkg/processor"
saga := processor.NewSaga(
processor.SagaStep{
Name: "reserve-inventory",
Processor: reserveInventoryProc,
Compensator: releaseInventoryProc,
},
processor.SagaStep{
Name: "charge-payment",
Processor: chargePaymentProc,
Compensator: refundPaymentProc,
},
processor.SagaStep{
Name: "send-confirmation",
Processor: sendConfirmationProc,
Compensator: nil, // no compensator needed
},
)
Key parameters:
| Parameter | Description |
|---|---|
steps |
The forward processing steps, executed in order |
compensate |
Compensating steps, executed in reverse order on failure |
Properties set:
| Property | Description |
|---|---|
HelixSagaCurrentStep |
Name of the currently executing step (or name-compensate during rollback) |
HelixSagaCompleted |
true if saga completed successfully, false if compensated |
Behavior: Compensators run in reverse order of completed steps. Compensator errors are intentionally swallowed to ensure all compensators get a chance to execute. Context cancellation triggers compensation for all completed steps.
Sort (Message Sorting)¶
The Sort pattern sorts the elements of a collection body. The exchange body must be a slice/array. Elements are wrapped in temporary exchanges for the comparator function, then sorted using a stable sort algorithm.
YAML example:
workflows:
- id: sort-items
trigger:
uri: direct://unsorted
steps:
- sort:
starlang: 'payload.price' # sort by price field (ascending)
- target: direct://sorted-output
Go builder example:
import (
"github.com/arvenza/helix/pkg/processor"
"github.com/arvenza/helix/pkg/exchange"
)
// Default: sorts by string representation of body (ascending)
sorter := processor.NewSort(nextProcessor, nil)
// Custom comparator: sort by a numeric field
sorter = processor.NewSort(nextProcessor, func(a, b *exchange.Exchange) bool {
aMap := a.In().Body().(map[string]any)
bMap := b.In().Body().(map[string]any)
return aMap["price"].(float64) < bMap["price"].(float64)
})
Key parameters:
| Parameter | Description |
|---|---|
starlang |
Comparator expression for sorting (evaluates per element) |
Behavior: Requires the body to be a slice. Uses sort.SliceStable for deterministic ordering of equal elements. The sorted slice replaces the original body.
Rollback (Transaction Rollback)¶
The Rollback processor marks the exchange for rollback by setting the HelixRollbackOnly property to true. It can optionally set an exception on the exchange and return an error, or just mark the property for downstream processors to act upon.
YAML example:
workflows:
- id: validate-and-rollback
trigger:
uri: direct://input
steps:
- choice:
when:
- starlang: 'payload.amount < 0'
steps:
- log: { message: "Invalid amount, rolling back" }
- process:
ref: rollback
Go builder example:
import "github.com/arvenza/helix/pkg/processor"
// Rollback with error propagation
rollback := processor.NewRollback("insufficient funds")
// Rollback that only sets the property (no error returned)
markOnly := processor.NewRollback("flagged for review")
markOnly.MarkOnly = true
Key parameters:
| Parameter | Description |
|---|---|
message |
Custom rollback reason message (default: "rollback") |
mark-only |
If true, only sets the HelixRollbackOnly property without returning an error |
Properties set:
| Property | Description |
|---|---|
HelixRollbackOnly |
Always set to true |
HelixRollbackMessage |
The rollback reason string |
Stop (Explicit Workflow Stop)¶
The Stop processor immediately halts routing of the current exchange. It sets the HelixRouteStop property to true and returns nil (no error). Downstream processors in the pipeline will not execute. Use StopAwarePipeline to honor this property in custom pipelines.
YAML example:
workflows:
- id: conditional-stop
trigger:
uri: direct://input
steps:
- choice:
when:
- starlang: 'payload.type == "ignore"'
steps:
- log: { message: "Dropping message" }
- process:
ref: stop
otherwise:
steps:
- target: direct://continue
Go builder example:
import "github.com/arvenza/helix/pkg/processor"
stop := processor.NewStop()
// Use StopAwarePipeline to respect the stop signal
pipeline := processor.NewStopAwarePipeline(proc1, stop, proc2)
// proc2 will never execute because Stop was called before it
Properties set: HelixRouteStop -- always set to true.
Validate (Message Validation) -- Expanded¶
Helix provides two validation mechanisms: a predicate-based validator (used in YAML DSL) and a schema-based validator for JSON Schema-like validation rules.
Predicate Validation¶
Evaluates a Starlang predicate. If the predicate returns false, a ValidationError is returned.
YAML example:
workflows:
- id: validate-order
trigger:
uri: direct://orders
steps:
- validate: 'payload.orderId != "" and payload.amount > 0'
- validate: 'sizeOf(payload.items) > 0'
- target: direct://process-order
Go builder example:
builder := dsl.NewRoute().
ID("validate-order").
From("direct://orders").
Validate(`payload.orderId != "" and payload.amount > 0`).
Validate(`sizeOf(payload.items) > 0`).
To("direct://process-order")
Schema Validation¶
The SchemaValidator validates exchange bodies against JSON Schema-like rules supporting type checking, required fields, min/max constraints, and regex pattern matching.
Go builder example:
import "github.com/arvenza/helix/pkg/processor"
schema := map[string]any{
"type": "object",
"required": []any{"name", "age"},
"properties": map[string]any{
"name": map[string]any{
"type": "string",
"pattern": "^[A-Z]",
},
"age": map[string]any{
"type": "number",
"min": 0,
"max": 150,
},
},
}
validator := processor.NewSchemaValidator(schema, true /* strict mode */)
// In strict mode, unknown properties cause validation errors
Schema rules:
| Rule | Description |
|---|---|
type |
string, number, integer, boolean, array, object |
required |
List of required field names (for object types) |
min / max |
Numeric min/max bounds |
pattern |
Regex pattern for string validation |
properties |
Nested schema rules for object fields |
Errors: The SchemaValidator returns a SchemaValidationErrors aggregating all violations, each with a field path and message. Strict mode rejects unknown properties.
Sample (Message Sampling) -- Expanded¶
The Sample processor allows only a subset of messages through, useful for monitoring, debugging, or load shedding. It supports two modes: rate-based (probabilistic) and time-based (at most one message per period).
YAML example:
workflows:
- id: sample-logs
trigger:
uri: kafka://logs
steps:
- sample:
rate: 10 # 10% of messages pass through
- target: direct://log-analysis
- id: sample-periodic
trigger:
uri: kafka://metrics
steps:
- sample:
period: "5s" # at most 1 message every 5 seconds
- target: direct://dashboard
Go builder example:
import (
"github.com/arvenza/helix/pkg/processor"
"time"
)
// Rate-based: 10% of messages pass through
sampler := processor.NewSampleProcessor(0.10, 0)
// Time-based: at most 1 message per 5 seconds
sampler = processor.NewSampleProcessor(0, 5*time.Second)
Key parameters:
| Parameter | Description |
|---|---|
rate |
Percentage (1-100 in YAML, 0.0-1.0 in Go) of messages to pass through |
period |
Duration string; at most one message per period is allowed through. Takes precedence over rate when both are set. |
Properties set: HelixSampleSkipped is set to true on messages that are dropped.
Threads (Goroutine Pool)¶
The Threads processor executes an inner processor using a bounded goroutine pool (worker pool pattern). This decouples the caller from processing by submitting work to a pool of goroutines, providing backpressure when all workers are busy.
Go builder example:
import "github.com/arvenza/helix/pkg/processor"
// Pool of 20 goroutines, blocks when all workers are busy
threads := processor.NewThreads(innerProcessor, 20)
// Non-blocking mode: return error when pool is exhausted
threads.WaitForSlot = false
Key parameters:
| Parameter | Description |
|---|---|
pool-size |
Number of goroutines in the pool (default: 10) |
wait-for-slot |
If true (default), block until a slot is available. If false, return error when pool is exhausted. |
Properties set:
| Property | Description |
|---|---|
HelixThreadsPoolSize |
The configured pool size |
HelixThreadsRejected |
true if the exchange was rejected (non-blocking mode, pool full) |
Detour (Conditional Bypass)¶
The Detour pattern conditionally routes the exchange through an additional processor (the "detour") based on a predicate. This is useful for debug/audit scenarios where you want to conditionally inject processing without modifying the main route.
When the predicate matches and Intercept is false (default), the detour processor runs first, then the main processor. When Intercept is true, only the detour processor runs (main is bypassed).
Go builder example:
import (
"github.com/arvenza/helix/pkg/processor"
"github.com/arvenza/helix/pkg/expression"
)
// When debug header is set, route through debug processor before main
detour := processor.NewDetour(
expression.NewStarlang("headers[\"X-Debug\"]"), // predicate: header exists and is truthy
debugProcessor, // detour processor
mainProcessor, // main processor
)
// Intercept mode: detour replaces main processor entirely
detour.Intercept = true
Key parameters:
| Parameter | Description |
|---|---|
enabled |
Predicate that controls whether the detour is active |
intercept |
If true, the detour replaces the main processor (bypass). If false (default), detour runs before main. |
Properties set: HelixDetourEnabled -- true if the detour was activated for this exchange.
Normalize / Normalizer (Message Format Normalization) -- Expanded¶
The Normalizer pattern routes messages to different transformation processors based on their format, producing a common (canonical) output format. This is essential when receiving the same logical message in multiple formats (XML, CSV, JSON, etc.).
Helix provides two implementations:
Normalize (YAML DSL)¶
Conditions are evaluated in order; the first matching condition's processor steps are executed.
workflows:
- id: normalize-formats
trigger:
uri: kafka://mixed-formats
steps:
- normalize:
conditions:
- starlang: 'headers["Content-Type"] == "application/xml"'
steps:
- unmarshal: { format: xml }
- transform: { starlang: 'xml_to_canonical(payload)' }
- starlang: 'headers["Content-Type"] == "text/csv"'
steps:
- unmarshal: { format: csv }
- transform: { starlang: 'csv_to_canonical(payload)' }
- starlang: 'headers["Content-Type"] == "application/json"'
steps:
- transform: { starlang: 'json_to_canonical(payload)' }
- target: direct://canonical-processing
Go builder example:
builder := dsl.NewRoute().
ID("normalize-formats").
From("kafka://mixed-formats").
Normalize().
When(`headers["Content-Type"] == "application/xml"`).
Transform(`xml_to_canonical(payload)`).
When(`headers["Content-Type"] == "text/csv"`).
Transform(`csv_to_canonical(payload)`).
When(`headers["Content-Type"] == "application/json"`).
Transform(`json_to_canonical(payload)`).
End().
To("direct://canonical-processing")
Normalizer (Programmatic API)¶
The Normalizer struct provides a data-driven API where rules can be added dynamically at runtime. It also supports an optional fallback processor for unrecognized formats.
import "github.com/arvenza/helix/pkg/processor"
normalizer := processor.NewNormalizer(fallbackProcessor)
normalizer.
AddRule(xmlPredicate, xmlTransformProcessor).
AddRule(csvPredicate, csvTransformProcessor).
AddRule(jsonPredicate, jsonTransformProcessor)
Properties set: HelixNormalizerMatched -- index of the matched rule (-1 if no rule matched).
Bean (Method Invocation) -- Expanded¶
The Bean processor looks up a Go object from a registry by name and invokes a method on it using reflection. The exchange (or its body) is passed as an argument. This is useful for integrating existing Go services without wrapping them in the Processor interface.
YAML example:
workflows:
- id: bean-example
trigger:
uri: direct://orders
steps:
- bean:
ref: orderService
method: ProcessOrder
- target: direct://next
Go builder example:
// Register the bean in the registry
registry.Bind("orderService", &OrderService{})
// Use in a route
builder := dsl.NewRoute().
ID("bean-example").
From("direct://orders").
Bean("orderService", "ProcessOrder").
To("direct://next")
Key parameters:
| Parameter | Description |
|---|---|
ref |
Name of the bean in the registry |
method |
Method name to invoke on the bean |
Method signature support: The Bean processor automatically maps arguments by type:
| Argument type | Mapping |
|---|---|
context.Context |
Receives the current context |
*exchange.Exchange |
Receives the full exchange |
| Other types | Receives the exchange body (if assignable) |
If the last return value is error, it is checked and propagated. The first non-error return value replaces the exchange body. Panics in the reflected method call are recovered and returned as errors.
ServiceCall (Service Registry Lookup)¶
The ServiceCall pattern looks up a service (processor) from the registry by name and invokes it. This decouples workflow definitions from concrete service implementations, enabling late binding and dynamic service resolution.
Go builder example:
import "github.com/arvenza/helix/pkg/processor"
// Register services in the registry
registry.Bind("paymentService", paymentProcessor)
registry.Bind("inventoryService", inventoryProcessor)
// ServiceCall looks up and invokes by name
sc := processor.NewServiceCall(registry, "paymentService")
// Optional: add a filter that runs before the service call
sc.ServiceFilter = headerSetterProcessor
Key parameters:
| Parameter | Description |
|---|---|
service-name |
Name of the service to look up in the registry |
filter |
Optional pre-processing step before the service call (e.g., to set required headers) |
Properties set:
| Property | Description |
|---|---|
HelixServiceCallServiceName |
The service name used for lookup |
HelixServiceCallServiceFound |
true if the service was found in the registry |
Additional Advanced Patterns (Go API Only)¶
The following patterns are available through the Go API for advanced use cases:
ScatterGather¶
Sends a copy of the exchange to multiple processors in parallel (scatter) and aggregates all responses (gather). Combines the Multicast and Aggregate patterns with an optional timeout.
Properties: HelixScatterGatherSize, HelixScatterGatherTimeout.
ProcessManager¶
Stateful routing through named steps with transition functions. Useful for workflow orchestration where the next step depends on the result of the current step.
pm := processor.NewProcessManager("start")
pm.AddStep(&processor.ProcessStep{
Name: "start",
Processor: startProc,
Transition: func(ctx context.Context, ex *exchange.Exchange) (string, error) {
if ex.In().Header("approved") == true { return "ship", nil }
return "review", nil
},
})
pm.MaxSteps = 50 // safety limit to prevent infinite loops (default: 100)
Properties: HelixProcessManagerStep, HelixProcessManagerStepCount, HelixProcessManagerComplete.
MessageBridge¶
Bridges between two messaging systems with an optional transformation in between.
Properties: HelixBridgeSource, HelixBridgeTarget.
ChannelAdapter¶
Adapts a non-messaging data source (file, database, REST API) into the exchange system.
Properties: HelixChannelAdapterSource.
ContentEnricher / ContentFilter¶
In-process enrichment and filtering using Go callback functions, without involving an external endpoint.
enricher := processor.NewContentEnricher(lookupFunc, mergeStrategy)
filter := processor.NewContentFilter(stripSensitiveFields)
Convert Body (Type Conversion)¶
The ConvertBody processor converts the exchange body to a specified type. This is useful when a downstream processor or endpoint expects a specific type (e.g., converting a JSON string to a map, or a numeric string to an integer).
YAML example:
workflows:
- id: convert-example
trigger:
uri: kafka://events
steps:
# Parse JSON string body into a map
- convert-body:
type: json
# Or convert body to a string
- convert-body:
type: string
# Or convert to bytes
- convert-body:
type: "[]byte"
charset: utf-8
Go builder example:
import "github.com/arvenza/helix/pkg/processor"
// Convert body to map[string]any (JSON parse)
conv := processor.NewConvertBody("json", "")
// Convert body to integer
conv = processor.NewConvertBody("int", "")
// Convert body to float64
conv = processor.NewConvertBody("float64", "")
// Convert body to boolean
conv = processor.NewConvertBody("bool", "")
Supported types:
| Type | Go result type | Notes |
|---|---|---|
string |
string |
Uses fmt.Sprintf("%v", body) |
[]byte |
[]byte |
Byte representation of the body |
int |
int |
Parses from string, float64, int64, json.Number |
float64 |
float64 |
Parses from string, int, int64, float32, json.Number |
bool |
bool |
Parses from string ("true"/"false"), int (0/1), float64 |
json / map |
map[string]any |
Unmarshals JSON string/bytes, or marshals+unmarshals structs |
Set Property / Remove Property / Remove Properties¶
Exchange properties are metadata attached to the exchange (not the message). Unlike headers, properties are not forwarded to endpoints — they are local to the workflow for internal state tracking.
YAML example:
workflows:
- id: property-example
trigger:
uri: direct://input
steps:
# Set a property from a starlang expression
- set-property:
name: processingStage
starlang: '"validation"'
# Set a property to a constant
- set-property:
name: source
constant: "api-gateway"
# Remove a single property
- remove-property: tempData
# Remove all properties matching a pattern
- remove-properties:
pattern: "temp.*"
except:
- tempImportant
Go builder example:
import "github.com/arvenza/helix/pkg/processor"
// Set property from a dynamic expression
setProp := processor.NewSetProperty("stage", func(ctx context.Context, ex *exchange.Exchange) any {
return "enriched"
})
// Set property to a constant value
setConst := processor.NewSetPropertyConstant("source", "api-gateway")
// Remove a single property
rmProp := processor.NewRemoveProperty("tempData")
// Remove properties matching a glob pattern, except specified ones
rmProps := processor.NewRemoveProperties("temp.*", "tempImportant")
Key parameters:
| Step | Parameter | Description |
|---|---|---|
set-property |
name |
Property name to set |
set-property |
starlang |
Expression that evaluates to the property value |
set-property |
constant |
Static string value (alternative to starlang) |
remove-property |
name |
Property name to remove |
remove-properties |
pattern |
Glob pattern (e.g., "temp.*", "Helix*") |
remove-properties |
except |
List of property names to keep even if they match |
Set Exchange Pattern¶
Changes the exchange pattern between InOnly (fire-and-forget) and InOut (request-reply). This is useful when a workflow receives messages as InOnly but needs to call an endpoint that requires InOut semantics, or vice versa.
YAML example:
workflows:
- id: pattern-switch
trigger:
uri: kafka://events # InOnly by default
steps:
# Switch to InOut for request-reply call
- set-exchange-pattern: InOut
- target: http://api-server/process
# Switch back to InOnly
- set-exchange-pattern: InOnly
- target: kafka://results
Go builder example:
import (
"github.com/arvenza/helix/pkg/processor"
"github.com/arvenza/helix/pkg/exchange"
)
setInOut := processor.NewSetExchangePattern(exchange.InOut)
setInOnly := processor.NewSetExchangePattern(exchange.InOnly)
Key parameters:
| Parameter | Description |
|---|---|
pattern |
InOnly (fire-and-forget) or InOut (request-reply) |
Throw Exception¶
Explicitly throws an error from within a route. This is useful for validating conditions and raising typed errors that can be caught by do-try / do-catch blocks or the workflow-level error handler.
YAML example:
workflows:
- id: validation-route
trigger:
uri: direct://input
steps:
- choice:
when:
- starlang: 'payload.amount < 0'
steps:
- throw-exception:
message: "Amount must not be negative"
type: ValidationError
otherwise:
steps:
- target: direct://process
# With error handling
- id: handled-exception
trigger:
uri: direct://input
steps:
- do-try:
steps:
- throw-exception:
message: "Something went wrong"
type: BusinessError
do-catch:
- exceptions: ["BusinessError"]
steps:
- log: { message: "Caught business error" }
Go builder example:
import "github.com/arvenza/helix/pkg/processor"
// Throw a typed exception
throw := processor.NewThrowException("Amount must not be negative", "ValidationError")
// Throw without a type
throw = processor.NewThrowException("Processing failed", "")
Key parameters:
| Parameter | Description |
|---|---|
message |
The error message |
type |
Optional error type/category (prepended to the message as "type: message") |
To Dynamic (Runtime URI Resolution)¶
The ToDynamic processor evaluates a Starlang expression at runtime to determine the target URI, then sends the exchange to that endpoint. This is useful when the destination is not known at workflow definition time.
YAML example:
workflows:
- id: dynamic-routing
trigger:
uri: direct://input
steps:
# Route based on a header value
- to-dynamic:
starlang: '"direct://" ++ headers["targetService"]'
# Route based on body content
- to-dynamic:
starlang: |
region = payload.get("region", "us-east")
"http://" ++ region ++ "-api.example.com/process"
Go builder example:
import "github.com/arvenza/helix/pkg/processor"
dynamic := processor.NewToDynamic(
func(ctx context.Context, ex *exchange.Exchange) string {
target := ex.In().Header("targetService").(string)
return "direct://" + target
},
endpointResolver, // implements EndpointResolver interface
)
Key parameters:
| Parameter | Description |
|---|---|
starlang |
Expression that evaluates to a URI string at runtime |
Behavior: If the expression returns an empty string, an error is returned. The resolved URI is looked up via the same component resolution mechanism used by to.
Script (Inline Starlang Execution)¶
The Script processor executes an inline Starlang script or loads one from a file. It is equivalent to a process step with starlang but provides a dedicated step type for clarity, and for compatibility with other orchestration frameworks that distinguish scripts from processors.
YAML example:
workflows:
- id: script-example
trigger:
uri: timer://trigger?period=5s
steps:
# Inline script
- script:
starlang: |
payload["timestamp"] = now()
payload["processed"] = True
headers["X-Processed-By"] = "helix"
# Script from file
- script:
starlang-file: scripts/transform.star
- target: direct://output
Go builder example:
import "github.com/arvenza/helix/pkg/processor"
// Wrap any processor as a ScriptProcessor
scriptProc := processor.NewScript(starlangProcessor)
Key parameters:
| Parameter | Description |
|---|---|
starlang |
Inline Starlang script to execute |
starlang-file |
Path to a .star file to load and execute |
Behavior: The script has full access to the exchange — it can read/write body, headers, and properties. This is identical to process: { starlang: ... } but makes the intent explicit when used as a scripting step rather than a processing step.
8. Resilience Patterns¶
Retry with Exponential Backoff¶
- retry:
max-attempts: 5
delay: "500ms"
backoff-multiplier: 2.0 # 500ms, 1s, 2s, 4s, 8s
steps:
- target: http://flaky-service/api
Circuit Breaker¶
- circuit-breaker:
threshold: 5 # open after 5 failures
half-open-after: "30s" # try again after 30s
steps:
- target: http://external-service/api
fallback:
- transform:
constant: '{"status": "service unavailable"}'
States: Closed (normal) → Open (reject all) → Half-Open (test one request).
Throttle (Rate Limiting)¶
- throttle:
rate: 100 # max 100 messages
period: "1s" # per second
steps:
- target: http://rate-limited-api/v1/data
Timeout¶
- timeout:
duration: "5s"
steps:
- target: http://slow-service/api
fallback:
- transform:
constant: '{"error": "timeout"}'
Delay¶
Loop¶
- loop:
count: 3
steps:
- log: { message: "Iteration ${property.HelixLoopIndex}" }
- target: direct://retry-step
Sample (Percentage-Based)¶
Only process a percentage of messages:
Try / Catch / Finally (Error Handling)¶
Structured error handling within a route. Catch blocks match errors by substring patterns in the error message. Finally blocks always execute regardless of success or failure.
workflows:
- id: resilient-processing
trigger:
uri: direct://input
steps:
- do-try:
steps:
- target: http://unreliable-service/api
- target: direct://next-step
do-catch:
- exceptions: ["timeout", "connection refused"]
steps:
- log: { message: "Network error, using fallback" }
- transform:
constant: '{"status": "fallback"}'
- exceptions: ["ValidationError"]
steps:
- log: { message: "Validation failed" }
- throw-exception:
message: "Invalid data"
type: ValidationError
- steps: # catch-all (no exceptions filter)
- log: { message: "Unexpected error caught" }
do-finally:
- log: { message: "Processing complete" }
Key parameters:
| Parameter | Description |
|---|---|
steps |
The main processing steps to try |
do-catch |
List of catch blocks, each with optional exceptions patterns and steps |
do-catch[].exceptions |
List of error message substrings to match (empty = catch-all) |
do-finally |
Steps that always execute (success or failure) |
Properties set: HelixExceptionCaught — the original error is stored when a catch block matches.
9. Data Transformation (Starlang)¶
Starlang is the default expression language, used everywhere: transforms, predicates, set-header, filter, split, and more.
In Transforms¶
- transform:
starlang: |
# Access body
order = payload
# Build new object
{
"id": order.orderId,
"customer": order.customer.name,
"total": order.items | map(item -> item.price * item.qty) | sum,
"status": "received",
"timestamp": now()
}
In Predicates (Choice/Filter)¶
- choice:
when:
- starlang: 'payload.total > 1000 and payload.status == "pending"'
steps: [...]
- starlang: '"VIP" in payload.customer.tags'
steps: [...]
Accessing Exchange Data¶
| Expression | Accesses |
|---|---|
payload |
Message body |
payload.fieldName |
Body field (if body is a map) |
headers["Name"] |
Message header |
properties["key"] |
Exchange property |
Using Starlang Files¶
transforms/enrich-order.star:
order = payload
order["processed_at"] = now()
order["region"] = headers["X-Region"] or "default"
order
Supported Starlang Formats¶
Starlang supports 17 data formats for marshal/unmarshal: JSON, XML, CSV, YAML, TOML, INI, Properties, MessagePack, BSON, HCL, EDN, S-Expression, Transit, Bencode, Pickle, Protocol Buffers, and Avro.
Expression Language — Starlang¶
Starlang is the sole expression language in Helix. All expressions — predicates, transforms, set-header values, filters, splits — use Starlang.
Variables available in every expression:
| Variable | Source | Example |
|---|---|---|
payload |
Exchange message body | upper(payload) |
headers |
Message headers map | headers["Content-Type"] |
properties |
Exchange properties map | properties["retryCount"] |
Go API usage:
// Simple expression
expr := expression.NewStarlang("upper(payload)")
// Header comparison predicate
isJSON := expression.NewStarlang(`headers["Content-Type"] == "application/json"`)
// String concatenation
greeting := expression.NewStarlang(`"Hello " ++ payload`)
// Constant value (for literals)
fixed := expression.NewConstant("fixed-value")
YAML DSL — all steps use starlang: field:
steps:
- filter:
starlang: 'sizeOf(payload) > 100'
- choice:
when:
- starlang: 'headers["priority"] == "high"'
steps:
- target: direct://urgent
- transform:
starlang: 'upper(payload)'
- set-header:
name: X-Processed
starlang: '"true"'
Expression vs Predicate reference:
| Constructor | Returns | Used as Predicate |
|---|---|---|
NewStarlang(script) |
Evaluation result | Result is truthy |
NewStarlangFile(path) |
File evaluation result | Result is truthy |
NewConstant(value) |
Constant value | Value is truthy |
10. REST DSL and OpenAPI¶
YAML REST Definition¶
rest:
- path: /api/v1
tag: Orders
description: "Order management API"
consumes: application/json
produces: application/json
cors:
allowed-origins: ["http://localhost:3000"]
allowed-methods: ["GET", "POST", "PUT", "DELETE"]
allowed-headers: ["Content-Type", "Authorization"]
max-age: 3600
security:
- type: bearer
scheme: JWT
verbs:
- method: GET
path: /orders
description: "List all orders"
target: direct://listOrders
produces: application/json
params:
- name: page
type: query
data-type: integer
description: "Page number"
default: "1"
- name: limit
type: query
data-type: integer
description: "Items per page"
default: "20"
- method: GET
path: /orders/{id}
description: "Get order by ID"
target: direct://getOrder
produces: application/json
params:
- name: id
type: path
data-type: string
required: true
- method: POST
path: /orders
description: "Create an order"
target: direct://createOrder
consumes: application/json
request-type: CreateOrderRequest
response-type: Order
# Handler workflows for the REST endpoints above
workflows:
- id: list-orders
trigger:
uri: "direct://listOrders"
steps:
- transform:
constant: '[{"id":"O1","item":"Widget","qty":5},{"id":"O2","item":"Gadget","qty":2}]'
- set-header:
name: Content-Type
constant: "application/json"
- id: get-order
trigger:
uri: "direct://getOrder"
steps:
- transform:
constant: '{"id":"O1","item":"Widget","qty":5,"status":"shipped"}'
- set-header:
name: Content-Type
constant: "application/json"
- id: create-order
trigger:
uri: "direct://createOrder"
steps:
- set-header:
name: Content-Type
constant: "application/json"
- transform:
constant: '{"id":"O3","status":"created"}'
Important: REST
verbsdefine the HTTP API surface (paths, params, security). Each verb'sto:field routes to adirect://endpoint. You must also define handlerworkflows:that consume thosedirect://endpoints and implement the actual logic.
Auto-Generated OpenAPI¶
Helix automatically generates an OpenAPI 3.0.3 spec at /openapi.json for REST definitions. It includes paths, parameters, request/response schemas, security schemes, and tags.
Go REST Builder¶
rest := dsl.Rest("/api/v1").
Tag("Users").
Description("User management").
RestConsumes("application/json").
RestProduces("application/json").
EnableCORS(
[]string{"*"},
[]string{"GET", "POST", "PUT", "DELETE"},
[]string{"Content-Type", "Authorization"},
3600,
).
Security("bearer", "", "", "JWT").
Get("/users").
VerbDescription("List all users").
QueryParam("page", "integer", "Page number").
To("direct://listUsers").
Post("/users").
VerbDescription("Create user").
RequestType("CreateUserRequest").
ResponseType("User").
To("direct://createUser").
Get("/users/{id}").
PathParam("id", "string", "User ID").
To("direct://getUser").
Build()
11. AI Agents¶
Helix includes a full AI agent framework. Agents can be defined inline in workflow steps with full configuration of model, tools (including MCP servers), memory, and sub-agents. For complete details, see the AI Agents Guide.
Agent Types¶
| Type | Description | Best For |
|---|---|---|
react |
ReAct (Reasoning + Acting) loop | General-purpose, tool-using agents |
supervisor |
Routes to specialist sub-agents | Multi-agent coordination |
plan-execute |
Plans first, then executes each step | Complex multi-step tasks |
deep |
Recursive task decomposition | Analysis, research tasks |
sequential |
Runs sub-agents in order | Pipeline workflows |
parallel |
Runs sub-agents concurrently | Independent parallel tasks |
loop |
Iterative refinement until DONE | Quality improvement cycles |
Supported LLM Providers (16)¶
Helix supports 17 LLM providers for agent steps and LLM route components. Any provider can be used in model.provider fields throughout your YAML configuration.
| Provider | Name | Default Model | Auth |
|---|---|---|---|
| OpenAI | openai |
gpt-4o |
OPENAI_API_KEY |
| Anthropic | anthropic |
claude-sonnet-4-20250514 |
ANTHROPIC_API_KEY |
| Ollama | ollama |
llama3 |
None (local) |
| Qwen (DashScope) | qwen |
qwen-plus |
DASHSCOPE_API_KEY |
| Groq | groq |
llama-3.3-70b-versatile |
GROQ_API_KEY |
| Mistral | mistral |
mistral-large-latest |
MISTRAL_API_KEY |
| Together AI | together |
meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo |
TOGETHER_API_KEY |
| Fireworks AI | fireworks |
accounts/fireworks/models/llama-v3p1-70b-instruct |
FIREWORKS_API_KEY |
| DeepInfra | deepinfra |
meta-llama/Meta-Llama-3.1-70B-Instruct |
DEEPINFRA_API_KEY |
| xAI (Grok) | xai |
grok-2-latest |
XAI_API_KEY |
| vLLM | vllm |
default |
Optional |
| Azure OpenAI | azure |
gpt-4o (deployment) |
AZURE_API_KEY |
| AWS Bedrock | bedrock |
(specify model ARN) | ACCESS_KEY:SECRET_KEY |
| Vertex / Gemini | vertex, gemini |
gemini-2.0-flash |
GOOGLE_API_KEY |
| Cohere | cohere |
command-r-plus |
COHERE_API_KEY |
| HuggingFace | huggingface |
meta-llama/Meta-Llama-3.1-70B-Instruct |
HF_API_KEY |
| Replicate | replicate |
meta/meta-llama-3.1-405b-instruct |
REPLICATE_API_TOKEN |
For complete details, API endpoints, and YAML examples for every provider, see the LLM & Embedding Providers Reference.
Inline Agent Step (Recommended)¶
The agent step provides full control over model, tools, memory, and sub-agents:
workflows:
- id: support-api
trigger:
uri: http-server://0.0.0.0:8080/api/support
steps:
- agent:
id: support-agent
type: react
model:
provider: openai
model: gpt-4o
api_key: "${OPENAI_API_KEY}"
base_url: "" # custom endpoint (optional)
temperature: 0.7
max_tokens: 4096
top_p: 0.9 # nucleus sampling (optional)
response_format: text # text or json_object (optional)
system_prompt: |
You are a customer support agent. Use the available tools
to look up customer information and create tickets.
tools:
- name: search_kb
type: rag-query
description: "Search the knowledge base"
parameters:
type: object
properties:
query:
type: string
required: [query]
config:
knowledge_bases: ["support-docs"]
- name: create_ticket
type: connector
connector: "http://ticketing:8080/v1/tickets"
description: "Create a support ticket"
# Import tools from an external MCP server
- name: crm-tools
type: mcp
description: "Customer CRM tools"
mcp_server:
url: "http://crm-server:3000/mcp"
transport: http
headers:
Authorization: "Bearer ${CRM_API_KEY}"
tools: [lookup_customer, update_customer] # filter
memory:
enabled: true
type: sliding-window # see Memory Types below
window_size: 30
namespace: support
max_iterations: 10
- marshal:
format: json
Sub-Agents¶
Supervisor, sequential, and parallel agents support nested sub_agents:
- agent:
id: supervisor
type: supervisor
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${ANTHROPIC_API_KEY}"
system_prompt: "Route requests to the right specialist."
sub_agents:
- id: billing
type: react
model:
provider: openai
model: gpt-4o
api_key: "${OPENAI_API_KEY}"
system_prompt: "Handle billing inquiries."
max_iterations: 5
- id: technical
type: react
model:
provider: openai
model: gpt-4o
api_key: "${OPENAI_API_KEY}"
system_prompt: "Handle technical support."
tools:
- name: diagnostics
type: mcp
description: "Diagnostic tools"
mcp_server:
url: "http://diag:3001/mcp"
transport: http
max_iterations: 8
Top-Level Agent Definitions¶
Define reusable agents alongside workflows:
agents:
- id: summarizer
type: react
model:
provider: ollama
model: llama3.1
base_url: "http://localhost:11434/api"
system_prompt: "Summarize the given text concisely."
max_iterations: 3
workflows:
- id: summarize-api
trigger:
uri: http-server://0.0.0.0:8080/api/summarize
steps:
- target: agent://summarizer
Memory Types¶
All 8 memory backends are configurable via YAML:
YAML type |
Description | Key Fields |
|---|---|---|
conversation |
Full conversation history (default) | — |
sliding-window |
Last N messages | window_size |
summary |
Summarizes old messages via LLM | max_messages |
token-window |
Bounded by estimated token count | max_tokens |
long-term |
In-memory persistent store | namespace |
sql-long-term |
PostgreSQL-backed persistence | dsn, session_id |
vector |
Semantic search via embeddings | embed, top_k |
composite |
Combines multiple backends | backends: [...] |
Per-session isolation: Add session_id_header to any memory type to create isolated memory per user/session. Without it, all requests share one memory instance.
memory:
enabled: true
type: sliding-window
window_size: 30
session_id_header: X-Session-ID # each unique header value gets its own memory
Example composite memory with per-session isolation:
memory:
enabled: true
type: composite
session_id_header: X-Session-ID
backends:
- type: sliding-window
window_size: 20
- type: vector
top_k: 5
embed:
provider: openai
model: text-embedding-3-small
api_key: "${OPENAI_API_KEY}"
See the AI Agents Guide — Per-Session Memory Isolation for full details.
MCP Server Tools¶
Import tools from external MCP servers into any agent:
tools:
# HTTP transport
- name: external-tools
type: mcp
description: "Tools from external MCP server"
mcp_server:
url: "http://mcp-server:3000/mcp"
transport: http # http, sse, stdio
headers:
Authorization: "Bearer ${API_KEY}"
tools: [tool_a, tool_b] # optional filter (empty = all)
# Stdio transport (local process)
- name: local-tools
type: mcp
description: "Local MCP tools"
mcp_server:
transport: stdio
command: "npx"
args: ["-y", "@anthropic/code-mcp-server"]
12. RAG Pipelines¶
Helix provides a complete Retrieval-Augmented Generation pipeline with dedicated YAML steps.
Ingestion Pipeline¶
workflows:
- id: rag-ingest
trigger:
uri: file://docs/?include=*.pdf&recursive=true&poll=60s
steps:
# 1. Parse documents
- rag-ingest:
parser: pdf # pdf, markdown, text, html, docx, csv, json, xml, starlang
# 2. Chunk into segments
- rag-chunk:
strategy: recursive # recursive, sentence, fixed, semantic, parent-child, token
chunk_size: 512
overlap: 50
# 3. Generate embeddings
- rag-embed:
provider: openai # openai, ollama, voyage, cohere
model: text-embedding-3-small
# 4. Store in vector database
- target: qdrant://knowledge-base/upsert?url=http://qdrant:6333
Query Pipeline¶
workflows:
- id: rag-query
trigger:
uri: http-server://0.0.0.0:8080/api/ask
steps:
# 1. Embed the query
- rag-embed:
provider: openai
model: text-embedding-3-small
# 2. Retrieve relevant chunks
- enrich:
uri: qdrant://knowledge-base/search?url=http://qdrant:6333&topK=10
# 3. Rerank results
- rag-rerank:
strategy: mmr # mmr (Maximal Marginal Relevance)
lambda: 0.7 # diversity vs relevance tradeoff
top_k: 5
# 4. Augment prompt with context
- rag-augment:
template: |
Answer the question based on the provided context.
Context:
{{.Context}}
Question: {{.Query}}
Answer:
# 5. Generate answer
- target: "openai://chat?model=gpt-4o&apiKey=${OPENAI_API_KEY}"
- marshal:
format: json
Full Ingest Pipeline (Multi-Format)¶
workflows:
- id: full-ingest
trigger:
uri: file://./documents?include=*.{pdf,md,html,json,xml,star,csv,docx,txt}&recursive=true&poll=60s
steps:
# Parse all supported formats (auto-detected from file extension)
- rag-ingest:
reader: directory
recursive: true
# Token-based chunking for precise control over LLM context windows
- rag-chunk:
strategy: token
chunk_size: 256
overlap: 32
# Embed with Voyage (Anthropic's recommended embedding provider)
- rag-embed:
provider: voyage
model: voyage-3
api_key: "${VOYAGE_API_KEY}"
# Store in vector database
- target: qdrant://knowledge-base/upsert?url=http://qdrant:6333
- id: query-pipeline
trigger:
uri: http-server://0.0.0.0:8080/api/ask
steps:
- rag-embed:
provider: voyage
model: voyage-3
api_key: "${VOYAGE_API_KEY}"
- enrich:
uri: qdrant://knowledge-base/search?url=http://qdrant:6333&topK=10
- rag-rerank:
strategy: mmr
lambda: 0.7
top_k: 5
- rag-augment:
template: |
Answer based on context.
Context:
{{.Context}}
Question: {{.Query}}
- target: "openai://chat?model=gpt-4o&apiKey=${OPENAI_API_KEY}"
- marshal:
format: json
RAG Step Reference¶
Helix provides 13 RAG step types for building complete retrieval-augmented generation pipelines.
Core Pipeline Steps¶
| Step | Purpose | Key Parameters |
|---|---|---|
rag-ingest |
Parse documents | parser (pdf, markdown, text, html, docx, csv, json, xml, starlang) or reader (csv, markdown, xml, html, pdf, directory) with options |
rag-chunk |
Chunk text | strategy (fixed, sentence, recursive, parent-child, semantic, token, code, markdown, html, sliding_window, hierarchical), chunk_size, overlap + strategy-specific options |
rag-embed |
Generate embeddings | provider (10 providers), model, api_key, base_url |
rag-store |
Configure vector store | store (pgvector, redis, chromadb, memory, opensearch, milvus, pinecone, qdrant, elasticsearch, weaviate), connection, collection, dimensions |
rag-rerank |
Rerank results | strategy (mmr, cross-encoder), lambda, top_k |
rag-augment |
Build prompt | template with {{.Context}} and {{.Query}} |
Advanced Steps¶
| Step | Purpose | Key Parameters |
|---|---|---|
rag-retrieve |
Retrieve from vector store | strategy (simple, mmr, hybrid, ensemble, auto-merging, parent, fusion, bm25, router), top_k, alpha, lambda + strategy-specific options |
rag-postprocess |
Filter/transform results | steps[] with type (similarity-filter, top-k, keyword, metadata, dedup, pii-mask, recency, length, diversity, normalize, keyword-boost, cohere-rerank, jina-rerank, threshold-similarity-filter, multi-metadata-filter, long-context-reorder) |
rag-synthesize |
Generate response from results | mode (compact, refine, tree-summarize, simple-summarize, accumulate, compact-refine, context-only, generation-only), model, + mode-specific options (max_chunks, chunks_per_level, parallel, max_tokens) |
rag-extract |
Extract metadata from documents | extractors[] (title, summary, keywords, questions, entities, statistical-keywords, language), model (optional LLM config) |
rag-eval |
Evaluate RAG quality | metrics[] (faithfulness, answer_relevancy, context_relevancy, correctness, hit_rate, mrr, ndcg, map, semantic_similarity), model (optional LLM config) |
rag-cache |
Deduplicate ingestion | enabled, max_size, transform |
rag-graph-build |
Build knowledge graph | store (in-memory), model (optional LLM config for entity extraction) |
rag-graph-query |
Query knowledge graph | hops, top_k |
Chunking Strategies¶
# Code-aware chunking (respects function/class boundaries)
- rag-chunk:
strategy: code
language: python # go, python, javascript, java, rust, etc.
chunk_size: 1024
overlap: 100
# Markdown-aware chunking (preserves heading hierarchy)
- rag-chunk:
strategy: markdown
chunk_size: 512
retain_headers: true # prepend parent headings to each chunk
# HTML-aware chunking (respects DOM structure)
- rag-chunk:
strategy: html
chunk_size: 256
# Sliding window with configurable stride
- rag-chunk:
strategy: sliding_window
window_size: 512
stride: 256
unit: words # chars or words
# Hierarchical multi-level chunking
- rag-chunk:
strategy: hierarchical
levels: [2048, 512, 128]
overlap: 50
Vector Store Configuration¶
# PostgreSQL with pgvector
- rag-store:
store: pgvector
connection: "postgres://user:pass@localhost:5432/db"
collection: my-docs
dimensions: 1536
# Redis with RediSearch
- rag-store:
store: redis
connection: "localhost:6379"
collection: vector_idx
password: "${env:REDIS_PASSWORD}"
# ChromaDB
- rag-store:
store: chromadb
connection: "http://localhost:8000"
collection: my-collection
# OpenSearch with k-NN plugin
- rag-store:
store: opensearch
connection: "http://localhost:9200"
collection: docs-index
dimensions: 768
username: admin
password: "${env:OS_PASSWORD}"
# In-memory (for development/testing)
- rag-store:
store: memory
collection: test
Data Readers (rag-ingest)¶
The reader field provides advanced reader options beyond the basic parser:
# CSV with custom delimiter
- rag-ingest:
reader: csv
delimiter: ";"
has_header: false
# Markdown with heading-based splitting
- rag-ingest:
reader: markdown
split_by_heading: true
strip_formatting: true
# XML with repeating element extraction
- rag-ingest:
reader: xml
row_tag: item
# Directory reader (auto-detects file types)
- rag-ingest:
reader: directory
recursive: true
extensions: [".md", ".txt", ".pdf"]
Retrieval Strategies¶
# Dual hybrid: weighted combination of vector + keyword search
- rag-retrieve:
strategy: hybrid
vector_weight: 0.7
keyword_weight: 0.3
top_k: 10
# Auto-merging: merges child chunks back to parent when threshold met
- rag-retrieve:
strategy: auto-merging
merge_threshold: 0.5
top_k: 10
# Ensemble: combines multiple retrieval strategies with RRF
- rag-retrieve:
strategy: ensemble
top_k: 15
LLM-Powered Steps¶
Steps that use LLMs (rag-synthesize, rag-extract, rag-eval, rag-graph-build) accept an optional model block:
- rag-synthesize:
mode: refine
max_chunks: 10 # process at most 10 chunks
model:
provider: groq
model: llama-3.3-70b-versatile
api_key: "${GROQ_API_KEY}"
- rag-synthesize:
mode: tree-summarize
chunks_per_level: 4 # combine 4 chunks per tree level
model:
provider: anthropic
model: claude-sonnet-4-20250514
- rag-synthesize:
mode: accumulate
parallel: true # generate per-chunk answers in parallel
model:
provider: openai
model: gpt-4o
- rag-synthesize:
mode: compact
max_tokens: 8000 # max context window tokens
model:
provider: openai
model: gpt-4o-mini
- rag-extract:
extractors: [keywords, title, entities]
model:
provider: mistral
model: mistral-large-latest
api_key: "${MISTRAL_API_KEY}"
- rag-eval:
metrics: [faithfulness, answer_relevancy]
model:
provider: openai
model: gpt-4o
api_key: "${OPENAI_API_KEY}"
- rag-graph-build:
store: in-memory
model:
provider: gemini
model: gemini-2.0-flash
api_key: "${GOOGLE_API_KEY}"
Postprocessing Example¶
- rag-postprocess:
steps:
- type: similarity-filter
params:
min_score: "0.3"
- type: dedup
- type: cohere-rerank
params:
api_key: "${env:COHERE_API_KEY}"
top_n: "5"
- type: jina-rerank
params:
api_key: "${env:JINA_API_KEY}"
top_n: "3"
- type: multi-metadata-filter
mode: all
filters:
- key: source
operator: eq
value: docs
- key: language
operator: contains
value: en
- type: long-context-reorder
- type: top-k
params:
k: "5"
All 10 embedding providers support api_key and base_url fields. All 17 LLM providers can be used in model blocks. See the LLM & Embedding Providers Reference for full details.
13. Connectors (440+)¶
Beyond the 31 built-in components, Helix provides 440+ schema-driven connectors across 33 categories.
Using Connectors in YAML¶
connectors:
my-s3:
type: s3
config:
bucket: my-bucket
region: us-east-1
access_key: "${AWS_ACCESS_KEY}"
secret_key: "${AWS_SECRET_KEY}"
lifecycle:
reconnect: true
health_interval: "30s"
max_retries: 5
retry_delay: "5s"
my-slack:
type: slack
config:
token: "${SLACK_TOKEN}"
channel: "#alerts"
workflows:
- id: process-uploads
trigger:
uri: connector://my-s3
steps:
- transform:
starlang: 'process_upload(payload)'
- target: connector://my-slack
Connector Categories¶
| Category | Count | Examples |
|---|---|---|
| AI | 14 | OpenAI, Anthropic, Cohere, HuggingFace, Groq, DeepSeek, Mistral |
| AWS | 46 | S3, DynamoDB, Lambda, SQS, SNS, Kinesis, Bedrock, SageMaker |
| Azure | 21 | Blob Storage, Cosmos DB, Service Bus, Azure AI, Key Vault |
| 34 | BigQuery, Cloud Storage, Pub/Sub, Vertex AI, Workspace | |
| Database | 23 | PostgreSQL, MySQL, MongoDB, Cassandra, Neo4j, Snowflake, ClickHouse |
| Messaging | 15 | Kafka, RabbitMQ, ActiveMQ, Pulsar, MQTT, ZeroMQ, IBM MQ |
| Communication | 25 | Slack, Teams, Discord, SendGrid, Twilio, Telegram |
| Microsoft | 22 | Teams, Outlook, OneDrive, Dynamics, Power Platform |
| Kubernetes | 20 | K8s API, Helm, Operators, Deployments, Services |
| File | 23 | S3, GCS, Azure Blob, FTP, SFTP, HDFS |
| Data Format | 17 | JSON, XML, CSV, Avro, Protobuf, Parquet |
| DevOps | 12 | Terraform, Ansible, Jenkins, GitLab CI |
| eCommerce | 13 | Shopify, Stripe, WooCommerce, PayPal |
| Monitoring | 10 | Datadog, New Relic, Grafana, Prometheus |
| Other | 175+ | Blockchain, IoT, EDI, ERP, CRM, Financial, and more |
All connectors are schema-driven with config validation, automatic defaults, and lifecycle management (reconnect, health checks).
14. Security¶
Workflow-Level Authentication (YAML DSL)¶
Helix supports declarative authentication on any route. The auth: block is added at the workflow level and runs before any steps — unauthenticated requests are rejected immediately.
API Key Authentication¶
workflows:
- id: protected-api
trigger:
uri: http-server://0.0.0.0:8080/api
auth:
type: apikey
apikey:
header: X-API-Key # default header name
keys:
- "${env:API_KEY_1}" # resolve from env var
- "${vault:secret/api/key}" # resolve from Vault
steps:
- target: direct://process
JWT Authentication¶
workflows:
- id: jwt-protected
trigger:
uri: http-server://0.0.0.0:8080/api
auth:
type: jwt
jwt:
secret: "${env:JWT_SECRET}" # HMAC shared secret
# public-key: /certs/public.pem # or RSA/ECDSA public key
issuer: my-auth-server # required issuer claim
audience: helix-api # required audience claim
claims: # additional required claims
role: admin
steps:
- log: { message: "Authenticated user: ${header.auth.subject}" }
- target: direct://process
HTTP Basic Authentication¶
workflows:
- id: basic-auth-route
trigger:
uri: http-server://0.0.0.0:8080/admin
auth:
type: basic
basic:
users:
admin: "$2a$10$..." # bcrypt hash of password
readonly: "$2a$10$..."
steps:
- target: direct://admin-panel
OAuth2 Token Introspection¶
workflows:
- id: oauth2-protected
trigger:
uri: http-server://0.0.0.0:8080/api
auth:
type: oauth2
oauth2:
token-url: https://auth.example.com/oauth/introspect
client-id: helix-app
client-secret: "${vault:secret/oauth/client-secret}"
scopes:
- read
- write
steps:
- target: direct://process
Auth types summary:
| Type | Mechanism | Key Config |
|---|---|---|
apikey |
Constant-time header comparison | header, keys[] |
jwt |
HMAC or RSA signature verification, claims validation | secret or public-key, issuer, audience, claims |
basic |
HTTP Basic with bcrypt password hashing | users map (username -> bcrypt hash) |
oauth2 |
RFC 7662 token introspection | token-url, client-id, client-secret, scopes[] |
Headers set on successful auth: auth.subject, auth.issuer, and claim-specific headers (e.g., auth.role) are added to the exchange for downstream use.
Secret management: Use ${env:VAR}, ${vault:path/key}, or ${aws-sm:secret-name} to avoid hardcoding secrets in YAML files. See Configuration Reference for details.
JWT Authentication (Go API)¶
import "github.com/arvenza/helix/pkg/security"
// Create an auth processor for JWT (stdlib crypto — no external dependencies)
proc := security.NewAuthProcessor(&security.AuthConfig{
Type: "jwt",
JWT: &security.JWTConfig{
Secret: "my-secret",
Issuer: "helix",
Audience: "my-api",
},
})
// proc is an engine.Processor — use it in a workflow pipeline
Basic Auth¶
proc := security.NewAuthProcessor(&security.AuthConfig{
Type: "basic",
Basic: &security.BasicAuthConfig{
Users: map[string]string{
"admin": "$2a$10$...", // bcrypt hashed password
},
},
})
OAuth2 Token Introspection¶
// RFC 7662 compliant
proc := security.NewAuthProcessor(&security.AuthConfig{
Type: "oauth2",
OAuth2: &security.OAuth2Config{
TokenURL: "https://auth.example.com/oauth/introspect",
ClientID: "helix-app",
ClientSecret: "secret",
Scopes: []string{"read", "write"},
},
})
TLS Configuration¶
tlsConfig, err := security.NewTLSConfig(security.NetworkTLSConfig{
Enabled: true,
CertFile: "/certs/server.crt",
KeyFile: "/certs/server.key",
CAFile: "/certs/ca.crt",
MinVersion: tls.VersionTLS12,
})
Multi-Tenancy¶
helix:
tenant:
enabled: true
header_name: "X-Tenant-ID"
required: true
allowed_tenants:
- acme-corp
- globex
When enabled, the TenantInterceptor extracts the tenant ID from every request header, validates it against the allowlist, and sets helix.tenant as an exchange property.
15. Cluster and High Availability¶
Architecture¶
┌─────────────┐ UDP Gossip ┌─────────────┐ UDP Gossip ┌─────────────┐
│ Node A │◄───────────────►│ Node B │◄───────────────►│ Node C │
│ (Leader) │ │ (Follower) │ │ (Follower) │
└─────────────┘ └─────────────┘ └─────────────┘
Configuration¶
helix:
cluster:
enabled: true
node_id: "helix-1" # unique per node (defaults to hostname)
bind_addr: "0.0.0.0"
bind_port: 7946
seeds:
- "helix-2:7946"
- "helix-3:7946"
Features¶
| Feature | How It Works |
|---|---|
| Membership | UDP gossip protocol — nodes ping each other every 1s, declared dead after 30s |
| Leader Election | Deterministic — sorted alive member IDs, first one wins. No voting needed |
| Workflow Distribution | Consistent hashing — each singleton workflow assigned to one node |
| Distributed Locks | TTL-based locks with re-entrant support |
Singleton Routes¶
Routes marked as singleton: true (or using cron:// / timer:// schemes) run on exactly one cluster node:
workflows:
- id: daily-report
singleton: true
trigger:
uri: cron://report?schedule=0 9 * * *
steps:
- target: direct://generate-report
When a node joins or leaves: 1. Leader re-election occurs 2. Workflow distribution is rebalanced 3. Singleton workflows are started on their new owner and suspended on the old owner
Duplicate Prevention by Trigger Type¶
| Trigger | Primary Mechanism | Backup |
|---|---|---|
| Kafka | Consumer groups (partition assignment) | Idempotent consumer |
| Cron/Timer | Singleton workflow (one node only) | Distributed lock |
| NATS | Queue groups | Idempotent consumer |
| AMQP | Single-consumer queue + ack | Idempotent consumer |
| HTTP | Load balancer routes to one pod | Idempotent consumer |
Idempotent Consumer (Cross-Instance Dedup)¶
- idempotent-consumer:
starlang: 'headers["kafka.KEY"]'
repository: redis # redis (distributed) or memory (local)
Uses Redis SETNX with configurable TTL for atomic check-and-set across all instances.
Kubernetes Deployment¶
apiVersion: apps/v1
kind: Deployment
metadata:
name: helix
spec:
replicas: 3
template:
spec:
containers:
- name: helix
image: helix:latest
ports:
- containerPort: 8080 # workflows
- containerPort: 8081 # health
- containerPort: 9090 # management
- containerPort: 7946 # gossip (UDP)
protocol: UDP
readinessProbe:
httpGet:
path: /ready
port: 8081
livenessProbe:
httpGet:
path: /health
port: 8081
env:
- name: HELIX_CLUSTER_ENABLED
value: "true"
- name: HELIX_CLUSTER_NODE_ID
valueFrom:
fieldRef:
fieldPath: metadata.name
---
apiVersion: v1
kind: Service
metadata:
name: helix-gossip
spec:
clusterIP: None # headless for DNS peer discovery
selector: { app: helix }
ports:
- port: 7946
protocol: UDP
16. Configuration Reference¶
helix.yaml¶
helix:
name: my-app
# Management REST API
management:
enabled: true
port: 9090 # GET /api/routes, POST /api/workflows/{id}/stop, etc.
# Health checks
health:
enabled: true
port: 8081 # GET /health, GET /ready
# Metrics
metrics:
prometheus:
enabled: true
port: 9100 # GET /metrics (Prometheus format)
# Distributed tracing
tracing:
opentelemetry:
enabled: true
endpoint: "http://jaeger:4318"
# Graceful shutdown
shutdown:
drain_timeout: "30s" # wait for in-flight exchanges
# Structured logging
logging:
level: info # debug, info, warn, error
format: json # text, json
components: # per-component log levels
kafka: debug
http: warn
# Multi-tenancy
tenant:
enabled: true
header_name: "X-Tenant-ID"
required: true
allowed_tenants: [acme, globex]
# Audit trail
audit:
enabled: true
store: memory # memory (ring buffer) or log (slog)
max_entries: 10000
body_preview: 256 # bytes to capture
# Cluster coordination
cluster:
enabled: true
node_id: "helix-1"
bind_addr: "0.0.0.0"
bind_port: 7946
seeds: ["helix-2:7946", "helix-3:7946"]
# MCP server
mcp:
servers:
- enabled: true
transport: http # http, sse, stdio
addr: "0.0.0.0:3000"
name: helix-mcp
version: "1.0.0"
workflows: [] # empty = expose all routes
Environment Variable Resolution¶
All ${VAR} and ${VAR:-default} patterns in YAML are resolved from environment variables:
trigger:
uri: "kafka://orders?brokers=${KAFKA_BROKERS:-localhost:9092}"
steps:
- target: "openai://chat?apiKey=${OPENAI_API_KEY}"
17. Observability¶
Management API (port 9090)¶
| Endpoint | Method | Description |
|---|---|---|
/api/info |
GET | Engine info (name, version, uptime) |
/api/workflows |
GET | List all routes with state |
/api/workflows/{id} |
GET | Route detail |
/api/workflows/{id}/start |
POST | Start a stopped workflow |
/api/workflows/{id}/stop |
POST | Stop a running workflow |
/api/workflows/{id}/suspend |
POST | Suspend a workflow |
/api/workflows/{id}/resume |
POST | Resume a suspended workflow |
/api/metrics |
GET | Workflow-level metrics (exchanges, errors, latency) |
/api/tracer |
GET | Backlog tracer output |
/api/config |
GET | Current configuration |
/api/agents |
GET | Registered AI agents |
/api/audit |
GET | Audit trail entries |
/api/health |
GET | Health check results |
Health Endpoints (port 8081)¶
| Endpoint | Purpose |
|---|---|
/health |
Liveness — returns engine state, inflight count, cluster info |
/ready |
Readiness — returns UP only when engine is fully started |
Prometheus Metrics (port 9100)¶
Automatically tracks per-workflow: exchange count, error count, processing latency histograms.
Distributed Tracing¶
When enabled, every exchange is traced through the pipeline with span context propagation. Compatible with Jaeger, Zipkin, and any OpenTelemetry collector.
Audit Trail¶
When enabled, every exchange is recorded with: timestamp, workflow ID, exchange ID, body preview, headers, processing duration, and status (success/failure).
Query via: GET /api/audit?routeId=order-processing&limit=100
18. Performance Tuning¶
Exchange Pool¶
Helix uses lock-free atomics and object pooling for the Exchange hot path:
- Body reads: lock-free via
atomic.Value - Headers/Properties:
sync.Map(no mutex contention) - Object pooling:
AcquireExchange()/ReleaseExchange()reuses Exchange objects
SEDA Backpressure¶
For async workflows, tune SEDA for your workload:
| Policy | Behavior |
|---|---|
fail |
Return error immediately when queue is full |
block |
Block the producer until space is available |
drop |
Silently drop the message |
newest |
Drop the oldest message to make room |
Kafka Consumer Groups¶
Scale horizontally by running multiple Helix instances with the same group ID. Kafka automatically distributes partitions across consumers.
Connection Pooling¶
All database and messaging components use connection pooling: - PostgreSQL: pgxpool (configurable pool size) - MongoDB: mongo-driver internal pool - Redis: go-redis connection pool - Kafka: kafka-go Writer/Reader pools
Graceful Shutdown¶
On SIGINT/SIGTERM:
1. Stop accepting new exchanges
2. Wait up to drain_timeout for in-flight exchanges to complete
3. Stop all consumers
4. Stop all components
19. Cookbook¶
Recipe 1: Kafka to PostgreSQL with Error Handling¶
Process orders from Kafka, route by priority, store in PostgreSQL. Failures go to a dead-letter topic.
workflows:
- id: order-pipeline
trigger:
uri: kafka://raw-orders?brokers=kafka:9092&group=order-pipeline
steps:
- unmarshal:
format: json
- transform:
starlang: |
{
"id": payload.orderId,
"customer": payload.customer.name,
"total": payload.items | map(i -> i.price * i.qty) | sum,
"priority": "high" if payload.total > 1000 else "normal",
"created_at": now()
}
- choice:
when:
- starlang: 'payload.priority == "high"'
steps:
- target: direct://fast-lane
otherwise:
steps:
- target: seda://normal-queue?concurrency=5
- target: "postgres://orders/insert?dsn=postgres://user:pass@db:5432/orders"
error-handler:
dead-letter-channel: kafka://order-dlq?brokers=kafka:9092
max-redeliveries: 3
redelivery-delay: "2s"
backoff-multiplier: 2.0
Recipe 2: REST API with CRUD Operations¶
rest:
- path: /api/v1
tag: Products
produces: application/json
consumes: application/json
security:
- type: bearer
scheme: JWT
verbs:
- method: get
path: /products
target: direct://list-products
- method: get
path: /products/{id}
target: direct://get-product
- method: post
path: /products
target: direct://create-product
- method: delete
path: /products/{id}
target: direct://delete-product
workflows:
- id: list-products
trigger: { uri: direct://list-products }
steps:
- set-header:
name: postgres.sql
constant: "SELECT * FROM products ORDER BY created_at DESC LIMIT 50"
- target: "postgres://products/query?dsn=${DATABASE_URL}"
- marshal: { format: json }
- id: create-product
trigger: { uri: direct://create-product }
steps:
- unmarshal: { format: json }
- validate: '"name" in payload and payload.name != ""'
- target: "postgres://products/insert?dsn=${DATABASE_URL}"
- marshal: { format: json }
Recipe 3: AI Support Agent with RAG¶
Complete customer support with knowledge base search, ticket creation, and conversation memory.
workflows:
# Ingest support docs
- id: ingest-docs
trigger:
uri: file://support-docs/?include=*.md&recursive=true&poll=300s
steps:
- rag-ingest: { parser: markdown }
- rag-chunk: { strategy: sentence, chunk_size: 256, overlap: 30 }
- rag-embed: { provider: openai, model: text-embedding-3-small }
- target: "qdrant://support-kb/upsert?url=http://qdrant:6333"
# Support API
- id: support-api
trigger:
uri: http-server://0.0.0.0:8080/api/support
steps:
- agent:
id: support-bot
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${ANTHROPIC_API_KEY}"
system_prompt: |
You are a customer support agent for Acme Corp.
Always search the knowledge base before answering.
Create tickets for issues you cannot resolve.
tools:
- name: search_kb
type: rag-query
config:
vector_store: "qdrant://support-kb?url=http://qdrant:6333"
embedding_model: text-embedding-3-small
- name: create_ticket
type: connector
connector: "http://ticketing-service:8080/v1"
memory:
enabled: true
store: memory
namespace: support
max_iterations: 8
- marshal: { format: json }
Recipe 4: Event-Driven Microservices with SEDA¶
Decouple ingestion from processing with async queues.
workflows:
# Fast HTTP ingest
- id: api-ingest
trigger:
uri: http-server://0.0.0.0:8080/api/events
steps:
- unmarshal: { format: json }
- set-header:
name: receivedAt
starlang: 'now()'
- target: seda://event-queue?bufferSize=50000&overflowPolicy=block
- set-body:
constant: '{"status": "accepted"}'
# Parallel processing
- id: event-processor
trigger:
uri: seda://event-queue?concurrency=20
steps:
- choice:
when:
- starlang: 'payload.type == "click"'
steps:
- target: kafka://clicks
- starlang: 'payload.type == "purchase"'
steps:
- target: kafka://purchases
- wiretap: "slack://webhook?url=${SLACK_WEBHOOK}"
otherwise:
steps:
- target: kafka://other-events
Recipe 5: Scatter-Gather with Aggregation¶
Call three services in parallel, aggregate results.
workflows:
- id: price-check
trigger:
uri: http-server://0.0.0.0:8080/api/best-price
steps:
- multicast:
uris:
- direct://supplier-a
- direct://supplier-b
- direct://supplier-c
parallel: true
strategy: bestPriceStrategy
- id: supplier-a
trigger: { uri: direct://supplier-a }
steps:
- circuit-breaker:
threshold: 3
half-open-after: "10s"
steps:
- target: "http://supplier-a.example.com/api/price?timeout=5s"
fallback:
- set-body: { constant: '{"price": 999999, "source": "supplier-a", "error": true}' }
- id: supplier-b
trigger: { uri: direct://supplier-b }
steps:
- target: "http://supplier-b.example.com/api/price?timeout=5s"
- id: supplier-c
trigger: { uri: direct://supplier-c }
steps:
- target: "http://supplier-c.example.com/api/price?timeout=5s"
Recipe 6: File Processing with S3¶
Watch for uploads, process, and archive.
workflows:
- id: s3-file-processor
trigger:
uri: "s3://uploads?region=us-east-1&pollInterval=30s&accessKey=${AWS_KEY}&secretKey=${AWS_SECRET}"
steps:
- log: { message: "Processing file: ${header.HelixS3Key}" }
- transform:
starlang: |
lines = payload.split("\n")
records = []
for line in lines[1:]:
fields = line.split(",")
if sizeOf(fields) >= 3:
records.append({
"name": fields[0],
"email": fields[1],
"amount": float(fields[2])
})
records
- split:
starlang: 'payload'
parallel: true
steps:
- target: "postgres://customers/insert?dsn=${DATABASE_URL}"
- set-body:
starlang: '{"processed": sizeOf(payload), "file": headers["HelixS3Key"]}'
- target: "s3://processed?region=us-east-1&accessKey=${AWS_KEY}&secretKey=${AWS_SECRET}"
Recipe 7: Rate-Limited External API Integration¶
workflows:
- id: api-proxy
trigger:
uri: http-server://0.0.0.0:8080/api/proxy
steps:
- throttle:
rate: 50
period: "1s"
steps:
- retry:
max-attempts: 3
delay: "500ms"
backoff-multiplier: 2.0
steps:
- target: "http://external-api.example.com/v2/data?timeout=10s"
- marshal: { format: json }
Recipe 8: Multi-Protocol Bridge¶
Bridge messages between Kafka, AMQP, and NATS.
workflows:
# Kafka → AMQP
- id: kafka-to-amqp
trigger:
uri: kafka://events?brokers=kafka:9092&group=bridge
steps:
- set-header:
name: amqp.ROUTING_KEY
starlang: 'payload.type if "type" in payload else "default"'
- target: "amqp://events?url=amqp://rabbit:5672/&exchange=events&durable=true"
# AMQP → NATS
- id: amqp-to-nats
trigger:
uri: "amqp://notifications?url=amqp://rabbit:5672/&durable=true"
steps:
- target: "nats://notifications.new?servers=nats://nats:4222"
# NATS → Redis Pub/Sub
- id: nats-to-redis
trigger:
uri: "nats://alerts.>?servers=nats://nats:4222&queue=alert-bridge"
steps:
- target: "redis://alerts?addr=redis:6379"
Recipe 9: Scheduled Report with Email¶
workflows:
- id: daily-report
singleton: true # only run on one cluster node
trigger:
uri: "cron://daily?schedule=0 9 * * 1-5"
steps:
- set-header:
name: postgres.sql
constant: |
SELECT status, COUNT(*) as count, SUM(amount) as total
FROM orders
WHERE created_at >= NOW() - INTERVAL '24 hours'
GROUP BY status
- target: "postgres://orders/query?dsn=${DATABASE_URL}"
- transform:
starlang: |
rows = payload
lines = ["Daily Order Report", "=" * 40, ""]
total_count = 0
total_amount = 0.0
for row in rows:
lines.append("{}: {} orders, ${:.2f}".format(
row.status, row.count, row.total))
total_count += row.count
total_amount += row.total
lines.append("")
lines.append("Total: {} orders, ${:.2f}".format(total_count, total_amount))
"\n".join(lines)
- set-header: { name: To, constant: "team@example.com" }
- set-header: { name: Subject, constant: "Daily Order Report" }
- target: "smtp://mail.example.com:587?from=reports@example.com&username=${SMTP_USER}&password=${SMTP_PASS}&starttls=true"
Recipe 10: Docker Container Orchestration¶
workflows:
- id: deploy-container
trigger:
uri: http-server://0.0.0.0:8080/api/deploy
steps:
- unmarshal: { format: json }
# Pull the image
- set-body:
starlang: '{"fromImage": payload.image, "tag": payload.tag or "latest"}'
- target: "docker://images/pull"
- log: { message: "Image pulled" }
# Create the container
- set-body:
starlang: |
{
"Image": payload.image ++ ":" ++ (payload.tag or "latest"),
"Env": [k ++ "=" ++ v for k, v in payload.get("env", {}).items()],
"ExposedPorts": {p ++ "/tcp": {} for p in payload.get("ports", [])}
}
- target: "docker://containers/create"
# Start it
- target: "docker://containers/start"
- set-body:
constant: '{"status": "deployed"}'
- marshal: { format: json }
Recipe 11: MCP Tool Server¶
Expose Helix workflows as tools for LLM clients (Claude Desktop, etc.).
helix:
mcp:
servers:
- enabled: true
transport: http
addr: "0.0.0.0:3000"
name: my-tools
workflows:
- id: search-products
description: "Search the product catalog by keyword"
trigger: { uri: direct://search-products }
steps:
- set-header:
name: postgres.sql
starlang: |
"SELECT * FROM products WHERE name ILIKE '%" ++ payload.keyword ++ "%' LIMIT 10"
- target: "postgres://products/query?dsn=${DATABASE_URL}"
- marshal: { format: json }
- id: get-order-status
description: "Look up the status of an order by ID"
trigger: { uri: direct://get-order-status }
steps:
- set-header:
name: postgres.sql
starlang: |
"SELECT * FROM orders WHERE id = '" ++ payload.orderId ++ "'"
- target: "postgres://orders/query?dsn=${DATABASE_URL}"
- marshal: { format: json }
LLM clients discover these as MCP tools named search-products and get-order-status with the descriptions provided.
Recipe 12: A2A Agent-to-Agent Communication¶
Expose an AI agent via A2A and call external A2A agents from workflows.
helix:
a2a:
servers:
- enabled: true
addr: "0.0.0.0:9000"
name: "support-agent"
description: "Customer Support AI"
version: "1.0.0"
workflows:
# Agent exposed via A2A (auto-discovered)
- id: support-handler
trigger: { uri: direct://support-handler }
steps:
- agent:
id: support
type: react
model:
provider: openai
model: gpt-4o
api_key: "${OPENAI_API_KEY}"
system_prompt: "You are a customer support agent."
# Call an external A2A agent from a route
- id: delegate-to-expert
trigger:
uri: http-server://0.0.0.0:8080/api/ask-expert
steps:
- target: "a2a://expert-agent.internal:9000/research"
- marshal: { format: json }
The A2A server auto-discovers agents and invocable workflows, exposing them as skills with an Agent Card at /.well-known/agent-card.json.
Recipe 13: AG-UI Streaming Chat Frontend¶
Expose an agent via AG-UI for real-time streaming chat with CopilotKit or custom frontends.
helix:
agui:
servers:
- enabled: true
addr: "0.0.0.0:4100"
path: "/agent"
name: "chat-assistant"
workflows:
- id: chat-handler
trigger: { uri: direct://chat-handler }
steps:
- agent:
id: assistant
type: react
model:
provider: openai
model: gpt-4o
api_key: "${OPENAI_API_KEY}"
system_prompt: "You are a helpful assistant."
memory:
enabled: true
type: sliding-window
window_size: 20
Frontend sends POST /agent with RunAgentInput JSON and receives SSE events:
RUN_STARTED -> TEXT_MESSAGE_START -> TEXT_MESSAGE_CONTENT* -> TEXT_MESSAGE_END -> RUN_FINISHED.
Recipe 14: A2UI Rich Form Agent¶
Expose an agent via A2UI to generate rich declarative UIs rendered by native frontend widgets.
helix:
a2ui:
servers:
- enabled: true
addr: "0.0.0.0:4200"
path: "/surface"
name: "form-agent"
workflows:
- id: form-handler
trigger: { uri: direct://form-handler }
steps:
- agent:
id: form-builder
type: react
model:
provider: openai
model: gpt-4o
api_key: "${OPENAI_API_KEY}"
system_prompt: |
Generate A2UI surfaces with forms and cards.
Return *a2ui.Surface objects with appropriate components.
Frontend connects GET /surface?message=<prompt> and receives JSONL A2UI messages:
beginRendering -> updateComponents -> dataModelUpdate.
User interactions sent back via POST /surface with userAction events.
Recipe 15: Multi-Protocol Agent Gateway¶
Expose a single agent via all protocols simultaneously.
helix:
name: multi-protocol-agent
mcp:
servers:
- enabled: true
transport: http
addr: "0.0.0.0:3000"
name: "agent-tools"
a2a:
servers:
- enabled: true
addr: "0.0.0.0:9000"
name: "agent"
description: "Multi-protocol AI Agent"
agui:
servers:
- enabled: true
addr: "0.0.0.0:4100"
path: "/agent"
name: "agent-chat"
a2ui:
servers:
- enabled: true
addr: "0.0.0.0:4200"
path: "/surface"
name: "agent-ui"
workflows:
- id: smart-agent
description: "AI agent accessible via MCP, A2A, AG-UI, and A2UI"
trigger: { uri: direct://smart-agent }
steps:
- agent:
id: smart
type: react
model:
provider: openai
model: gpt-4o
api_key: "${OPENAI_API_KEY}"
system_prompt: "You are a versatile AI assistant."
tools:
- name: search_kb
type: rag-query
description: "Search the knowledge base"
memory:
enabled: true
type: sliding-window
window_size: 30
This single configuration exposes the agent via: - MCP (port 3000) — for Claude Desktop and LLM tool calling - A2A (port 9000) — for other AI agents to call - AG-UI (port 4100) — for chat frontend streaming - A2UI (port 4200) — for rich declarative UI frontends
Recipe 16: Idempotent Kafka Consumer (Cluster-Safe)¶
workflows:
- id: payment-processor
trigger:
uri: kafka://payments?brokers=kafka:9092&group=payment-service
steps:
# Deduplicate across all instances via Redis
- idempotent-consumer:
starlang: 'headers["kafka.KEY"]'
repository: redis
- unmarshal: { format: json }
- validate: 'payload.amount > 0 and "currency" in payload'
# Process with retry
- retry:
max-attempts: 3
delay: "1s"
steps:
- target: "http://payment-gateway:8080/v1/charge?timeout=15s"
# Record result
- target: "postgres://payment_log/insert?dsn=${DATABASE_URL}"
- target: kafka://payment-results?brokers=kafka:9092
Recipe 17: Vision Agent — Image Analysis¶
Analyze images using multimodal LLM providers. Supports OpenAI GPT-4o, Anthropic Claude, Google Gemini, and Ollama vision models.
workflows:
- id: image-analyzer
trigger:
uri: http-server://0.0.0.0:8080/api/analyze-image
steps:
- agent:
id: vision
type: react
model:
provider: openai
model: gpt-4o
api_key: "${OPENAI_API_KEY}"
system_prompt: |
You are an expert image analyst. Describe images in detail,
identify objects, text, colors, composition, and context.
Send multimodal requests with image URLs or base64 data:
# Image URL
curl -X POST http://localhost:8080/api/analyze-image \
-H "Content-Type: application/json" \
-d '{"text": "What is in this image?", "images": ["https://example.com/photo.jpg"]}'
# Base64 data
curl -X POST http://localhost:8080/api/analyze-image \
-H "Content-Type: application/json" \
-d '{"text": "Describe this", "images": [{"mime_type": "image/png", "data": "iVBOR..."}]}'
Recipe 18: Multi-Image Comparison Agent¶
Compare multiple images in a single request.
workflows:
- id: image-compare
trigger:
uri: http-server://0.0.0.0:8080/api/compare
steps:
- agent:
id: comparator
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${ANTHROPIC_API_KEY}"
system_prompt: |
Compare the provided images. Identify similarities and differences
in content, style, composition, and visual elements.
Each provider formats multi-image content natively:
- OpenAI: Array of image_url content parts
- Anthropic: Array of image source blocks with base64 and media_type
- Gemini: Array of inlineData parts with mimeType
- Ollama: Native images array on the message
Recipe 19: Audio Processing Agent¶
Process audio input for transcription, analysis, or conversation.
workflows:
- id: audio-processor
trigger:
uri: http-server://0.0.0.0:8080/api/process-audio
steps:
- agent:
id: audio-agent
type: react
model:
provider: openai
model: gpt-4o-audio-preview
api_key: "${OPENAI_API_KEY}"
system_prompt: |
You process audio input. Transcribe speech, identify speakers,
detect languages, and summarize audio content.
Audio is sent as input_audio blocks with format (wav, mp3, etc.) for OpenAI.
Recipe 20: Supervisor Multi-Agent Team¶
Coordinate specialized agents with a supervisor that delegates and synthesizes results.
workflows:
- id: research-team
trigger:
uri: http-server://0.0.0.0:8080/api/research
steps:
- agent:
type: supervisor
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${ANTHROPIC_API_KEY}"
system_prompt: |
You coordinate a research team. Delegate tasks to the right specialist,
review their output, and synthesize a final report.
sub_agents:
- id: researcher
type: react
model:
provider: groq
model: llama-3.3-70b-versatile
api_key: "${GROQ_API_KEY}"
system_prompt: "You research topics thoroughly using web search."
tools:
- name: web_search
type: connector
connector: "https://api.search.example.com"
- id: analyst
type: react
model:
provider: openai
model: gpt-4o
api_key: "${OPENAI_API_KEY}"
system_prompt: "You analyze data and produce charts."
- id: writer
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${ANTHROPIC_API_KEY}"
system_prompt: "You write clear, concise reports from research findings."
The supervisor decides which sub-agent to call, passes context, and combines their outputs.
Recipe 21: Plan-Execute Agent¶
Break complex tasks into steps, execute each, then synthesize.
workflows:
- id: complex-task
trigger:
uri: http-server://0.0.0.0:8080/api/plan
steps:
- agent:
type: plan-execute
model:
provider: openai
model: gpt-4o
api_key: "${OPENAI_API_KEY}"
system_prompt: |
You are a project planner. Break tasks into concrete steps,
execute each step, and compile the results.
tools:
- name: search_web
type: connector
connector: "https://search.api.example.com"
- name: run_code
type: connector
connector: "http://sandbox:8080/execute"
max_iterations: 15
Recipe 22: Deep Research Agent¶
Recursive self-refinement agent that dives deep into topics.
workflows:
- id: deep-research
trigger:
uri: http-server://0.0.0.0:8080/api/deep-research
steps:
- agent:
type: deep
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${ANTHROPIC_API_KEY}"
system_prompt: |
You are a deep researcher. Explore topics thoroughly,
following leads and cross-referencing sources.
tools:
- name: search
type: rag-query
config:
vector_store: "elasticsearch://research-kb?url=http://es:9200"
embedding_model: text-embedding-3-small
max_iterations: 25
Recipe 23: Agent Flows — Sequential, Parallel, Conditional, Loop, Pipeline¶
Orchestrate multiple agents in complex workflows using standard Helix steps.
workflows:
# Sequential: agents run one after another (output of each feeds the next)
- id: content-pipeline
trigger:
uri: http-server://0.0.0.0:8080/api/content
steps:
- agent:
id: researcher
type: react
model: { provider: groq, model: llama-3.3-70b-versatile }
system_prompt: "Research the given topic."
- agent:
id: writer
type: react
model: { provider: openai, model: gpt-4o }
system_prompt: "Write an article from the research."
- agent:
id: editor
type: react
model: { provider: anthropic, model: claude-sonnet-4-20250514 }
system_prompt: "Edit for clarity and grammar."
# Parallel: agents run concurrently via multicast, results combined
- id: multi-analysis
trigger:
uri: http-server://0.0.0.0:8080/api/analyze
steps:
- multicast:
strategy: collect
steps:
- agent:
id: sentiment
type: react
model: { provider: groq, model: llama-3.3-70b-versatile }
system_prompt: "Analyze sentiment."
- agent:
id: entities
type: react
model: { provider: groq, model: llama-3.3-70b-versatile }
system_prompt: "Extract named entities."
- agent:
id: summary
type: react
model: { provider: groq, model: llama-3.3-70b-versatile }
system_prompt: "Summarize the text."
# Conditional: route to different agents based on input
- id: smart-router
trigger:
uri: http-server://0.0.0.0:8080/api/route
steps:
- choice:
when:
- starlang: '"code" in payload.lower()'
steps:
- agent:
id: coder
type: react
model: { provider: openai, model: gpt-4o }
system_prompt: "You are a coding assistant."
- starlang: '"equation" in payload or "+" in payload'
steps:
- agent:
id: mathematician
type: react
model: { provider: openai, model: gpt-4o }
system_prompt: "You solve math problems."
otherwise:
steps:
- agent:
id: assistant
type: react
model: { provider: groq, model: llama-3.3-70b-versatile }
system_prompt: "You are a general assistant."
# Loop: repeat agent until condition met
- id: iterative-refine
trigger:
uri: http-server://0.0.0.0:8080/api/refine
steps:
- loop:
count: 5
steps:
- agent:
id: refiner
type: react
model: { provider: openai, model: gpt-4o }
system_prompt: |
Improve the text. When it meets quality standards, include "APPROVED" in your response.
- filter:
starlang: '"APPROVED" not in payload'
# Pipeline: chain agents with transformation between steps
- id: data-pipeline
trigger:
uri: http-server://0.0.0.0:8080/api/pipeline
steps:
- agent:
id: extractor
type: react
model: { provider: groq, model: llama-3.3-70b-versatile }
system_prompt: "Extract structured data as JSON."
- transform:
starlang: 'read(payload, "application/json")'
- agent:
id: enricher
type: react
model: { provider: openai, model: gpt-4o }
system_prompt: "Enrich the data with additional context."
Recipe 24: Agent Memory Systems¶
Configure different memory backends for agent conversations.
workflows:
# Sliding window — keep last N messages
- id: chat-sliding
trigger: { uri: http-server://0.0.0.0:8080/api/chat }
steps:
- agent:
type: react
model: { provider: openai, model: gpt-4o }
system_prompt: "You are a helpful assistant."
memory:
type: sliding-window
window_size: 20
# Summary — summarize old messages to save tokens
- id: chat-summary
trigger: { uri: http-server://0.0.0.0:8081/api/chat }
steps:
- agent:
type: react
model: { provider: openai, model: gpt-4o }
system_prompt: "You are a helpful assistant."
memory:
type: summary
model: { provider: groq, model: llama-3.3-70b-versatile }
max_messages: 50
# Token window — keep messages within token budget
- id: chat-token
trigger: { uri: http-server://0.0.0.0:8082/api/chat }
steps:
- agent:
type: react
model: { provider: openai, model: gpt-4o }
system_prompt: "You are a helpful assistant."
memory:
type: token-window
max_tokens: 4096
# Long-term — persist conversations across restarts
- id: chat-persistent
trigger: { uri: http-server://0.0.0.0:8083/api/chat }
steps:
- agent:
type: react
model: { provider: openai, model: gpt-4o }
system_prompt: "You are a helpful assistant."
memory:
type: long-term
store: sql
dsn: "postgres://user:pass@db:5432/memory"
# Vector memory — semantic search over past conversations
- id: chat-vector
trigger: { uri: http-server://0.0.0.0:8084/api/chat }
steps:
- agent:
type: react
model: { provider: openai, model: gpt-4o }
system_prompt: "You are a helpful assistant."
memory:
type: vector
embed: { provider: openai, model: text-embedding-3-small }
store: "qdrant://conversations?url=http://qdrant:6333"
top_k: 5
# Composite — combine multiple memory strategies
- id: chat-composite
trigger: { uri: http-server://0.0.0.0:8085/api/chat }
steps:
- agent:
type: react
model: { provider: openai, model: gpt-4o }
system_prompt: "You are a helpful assistant."
memory:
type: composite
backends:
- type: sliding-window
window_size: 10
- type: summary
model: { provider: groq, model: llama-3.3-70b-versatile }
- type: long-term
store: sql
dsn: "postgres://user:pass@db:5432/memory"
Recipe 25: All 17 LLM Providers¶
Every supported LLM provider with configuration.
workflows:
# OpenAI (GPT-4o, GPT-4, GPT-3.5)
- id: openai-agent
trigger: { uri: direct://openai }
steps:
- agent:
type: react
model:
provider: openai
model: gpt-4o
api_key: "${OPENAI_API_KEY}"
# Anthropic (Claude 4.5 Sonnet, Claude Opus, Haiku)
- id: anthropic-agent
trigger: { uri: direct://anthropic }
steps:
- agent:
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${ANTHROPIC_API_KEY}"
# Ollama (local models — Llama, Mistral, etc.)
- id: ollama-agent
trigger: { uri: direct://ollama }
steps:
- agent:
type: react
model:
provider: ollama
model: llama3.3
base_url: "http://localhost:11434"
# Groq (ultra-fast inference)
- id: groq-agent
trigger: { uri: direct://groq }
steps:
- agent:
type: react
model:
provider: groq
model: llama-3.3-70b-versatile
api_key: "${GROQ_API_KEY}"
# Mistral
- id: mistral-agent
trigger: { uri: direct://mistral }
steps:
- agent:
type: react
model:
provider: mistral
model: mistral-large-latest
api_key: "${MISTRAL_API_KEY}"
# Azure OpenAI
- id: azure-agent
trigger: { uri: direct://azure }
steps:
- agent:
type: react
model:
provider: azure
model: gpt-4o
api_key: "${AZURE_OPENAI_KEY}"
base_url: "https://myresource.openai.azure.com/openai/deployments/gpt-4o?api-version=2024-02-15-preview"
# AWS Bedrock
- id: bedrock-agent
trigger: { uri: direct://bedrock }
steps:
- agent:
type: react
model:
provider: bedrock
model: anthropic.claude-3-5-sonnet-20241022-v2:0
base_url: "https://bedrock-runtime.us-east-1.amazonaws.com"
# Google Vertex AI / Gemini
- id: vertex-agent
trigger: { uri: direct://vertex }
steps:
- agent:
type: react
model:
provider: vertex
model: gemini-2.0-flash
api_key: "${GOOGLE_API_KEY}"
# Cohere
- id: cohere-agent
trigger: { uri: direct://cohere }
steps:
- agent:
type: react
model:
provider: cohere
model: command-r-plus
api_key: "${COHERE_API_KEY}"
# Together AI
- id: together-agent
trigger: { uri: direct://together }
steps:
- agent:
type: react
model:
provider: together
model: meta-llama/Llama-3.3-70B-Instruct-Turbo
api_key: "${TOGETHER_API_KEY}"
# Fireworks AI
- id: fireworks-agent
trigger: { uri: direct://fireworks }
steps:
- agent:
type: react
model:
provider: fireworks
model: accounts/fireworks/models/llama-v3p3-70b-instruct
api_key: "${FIREWORKS_API_KEY}"
# DeepInfra
- id: deepinfra-agent
trigger: { uri: direct://deepinfra }
steps:
- agent:
type: react
model:
provider: deepinfra
model: meta-llama/Llama-3.3-70B-Instruct
api_key: "${DEEPINFRA_API_KEY}"
# xAI (Grok)
- id: xai-agent
trigger: { uri: direct://xai }
steps:
- agent:
type: react
model:
provider: xai
model: grok-2
api_key: "${XAI_API_KEY}"
# HuggingFace Inference
- id: huggingface-agent
trigger: { uri: direct://huggingface }
steps:
- agent:
type: react
model:
provider: huggingface
model: meta-llama/Llama-3.3-70B-Instruct
api_key: "${HF_API_KEY}"
# Replicate
- id: replicate-agent
trigger: { uri: direct://replicate }
steps:
- agent:
type: react
model:
provider: replicate
model: meta/llama-3.3-70b-instruct
api_key: "${REPLICATE_API_KEY}"
# vLLM (self-hosted)
- id: vllm-agent
trigger: { uri: direct://vllm }
steps:
- agent:
type: react
model:
provider: vllm
model: meta-llama/Llama-3.3-70B-Instruct
base_url: "http://vllm-server:8000"
Recipe 26: RAG with Semantic Chunking and Hybrid Retrieval¶
Advanced RAG pipeline using embedding-based chunking, pgvector, and hybrid search.
workflows:
# Ingest with semantic chunking
- id: rag-ingest-semantic
trigger:
uri: http-server://0.0.0.0:8080/api/ingest
steps:
- rag-ingest:
reader: directory
path: ./documents
recursive: true
- rag-chunk:
strategy: semantic
threshold: 0.5
chunk_size: 2048
- rag-embed:
provider: openai
model: text-embedding-3-small
- rag-store:
store: pgvector
connection: "postgres://user:pass@db:5432/rag"
collection: documents
dimensions: 1536
# Query with hybrid retrieval
- id: rag-query-hybrid
trigger:
uri: http-server://0.0.0.0:8080/api/ask
steps:
- rag-retrieve:
strategy: hybrid
vector_weight: 0.7
keyword_weight: 0.3
top_k: 20
store: pgvector
connection: "postgres://user:pass@db:5432/rag"
- rag-postprocess:
steps:
- type: cohere-rerank
params:
api_key: "${COHERE_API_KEY}"
model: rerank-english-v3.0
top_n: "5"
- type: similarity-filter
params:
min_score: "0.3"
- rag-synthesize:
mode: compact
model:
provider: openai
model: gpt-4o
Recipe 27: RAG with Code Chunking¶
Specialized chunking for source code repositories.
workflows:
- id: code-ingest
trigger:
uri: http-server://0.0.0.0:8080/api/ingest-code
steps:
- rag-ingest:
reader: directory
path: ./src
recursive: true
- rag-chunk:
strategy: code
language: go # go, python, javascript, java, rust, generic
chunk_size: 4096
- rag-embed:
provider: openai
model: text-embedding-3-small
- rag-store:
store: elasticsearch
connection: "http://es:9200"
collection: code-embeddings
Supported languages: go (func/type/const/var blocks), python (def/class/if blocks), javascript (function/class/const/import), java (class/method/interface), rust (fn/struct/impl/enum), generic (brace-based).
Recipe 28: RAG with Hierarchical Chunking and Auto-Merging Retrieval¶
Parent-child chunk relationships for context-aware retrieval.
workflows:
- id: hierarchical-ingest
trigger:
uri: http-server://0.0.0.0:8080/api/ingest
steps:
- rag-ingest:
reader: text
- rag-chunk:
strategy: hierarchical
levels:
- 2048 # parent chunks
- 512 # child chunks
- 128 # leaf chunks
- rag-embed:
provider: openai
model: text-embedding-3-small
- rag-store:
store: qdrant
url: "http://qdrant:6333"
collection: hierarchical-docs
- id: hierarchical-query
trigger:
uri: http-server://0.0.0.0:8080/api/ask
steps:
- rag-retrieve:
strategy: auto-merging
merge_threshold: 0.6
top_k: 10
url: "http://qdrant:6333"
collection: hierarchical-docs
- rag-synthesize:
mode: refine
model:
provider: openai
model: gpt-4o
Auto-merging groups child chunks by parent_id metadata. When enough children from the same parent are retrieved, they merge into the parent chunk for better context.
Recipe 29: RAG with All 10 Data Readers¶
Ingest diverse document formats.
workflows:
# Text files
- id: ingest-text
trigger: { uri: direct://ingest-text }
steps:
- rag-ingest: { reader: text }
- rag-chunk: { strategy: sentence, chunk_size: 512 }
- rag-embed: { provider: openai, model: text-embedding-3-small }
- rag-store: { store: memory }
# JSON documents
- id: ingest-json
trigger: { uri: direct://ingest-json }
steps:
- rag-ingest: { reader: json }
- rag-chunk: { strategy: recursive, chunk_size: 1024 }
- rag-embed: { provider: openai, model: text-embedding-3-small }
- rag-store: { store: memory }
# PDF with image extraction
- id: ingest-pdf
trigger: { uri: direct://ingest-pdf }
steps:
- rag-ingest:
reader: pdf
- rag-chunk: { strategy: sentence, chunk_size: 512 }
- rag-embed: { provider: openai, model: text-embedding-3-small }
- rag-store: { store: pgvector, connection: "${DATABASE_URL}" }
# HTML with tag stripping
- id: ingest-html
trigger: { uri: direct://ingest-html }
steps:
- rag-ingest: { reader: html }
- rag-chunk: { strategy: html }
- rag-embed: { provider: openai, model: text-embedding-3-small }
- rag-store: { store: memory }
# CSV with header mapping
- id: ingest-csv
trigger: { uri: direct://ingest-csv }
steps:
- rag-ingest:
reader: csv
delimiter: ","
has_header: true
- rag-embed: { provider: openai, model: text-embedding-3-small }
- rag-store: { store: memory }
# Markdown with heading-based splitting
- id: ingest-markdown
trigger: { uri: direct://ingest-markdown }
steps:
- rag-ingest:
reader: markdown
split_by_heading: true
- rag-embed: { provider: openai, model: text-embedding-3-small }
- rag-store: { store: memory }
# XML
- id: ingest-xml
trigger: { uri: direct://ingest-xml }
steps:
- rag-ingest:
reader: xml
row_tag: item
text_tags: [title, description, content]
- rag-chunk: { strategy: sentence, chunk_size: 512 }
- rag-embed: { provider: openai, model: text-embedding-3-small }
- rag-store: { store: memory }
# DOCX
- id: ingest-docx
trigger: { uri: direct://ingest-docx }
steps:
- rag-ingest: { reader: docx }
- rag-chunk: { strategy: sentence, chunk_size: 512 }
- rag-embed: { provider: openai, model: text-embedding-3-small }
- rag-store: { store: memory }
# Starlang transform reader
- id: ingest-starlang
trigger: { uri: direct://ingest-starlang }
steps:
- rag-ingest:
reader: starlang
script: |
records = read(payload, "application/json")
[r["title"] ++ ": " ++ r["content"] for r in records]
- rag-embed: { provider: openai, model: text-embedding-3-small }
- rag-store: { store: memory }
# Directory auto-detect (walks directory, picks reader by extension)
- id: ingest-directory
trigger:
uri: cron://hourly?schedule=0 * * * *
steps:
- rag-ingest:
reader: directory
path: ./knowledge-base
recursive: true
- rag-chunk: { strategy: recursive, chunk_size: 1024, overlap: 100 }
- rag-embed: { provider: openai, model: text-embedding-3-small }
- rag-store: { store: pgvector, connection: "${DATABASE_URL}" }
Recipe 30: All 10 Chunking Strategies¶
workflows:
# Fixed-size character chunks
- id: chunk-fixed
steps:
- rag-chunk: { strategy: fixed, chunk_size: 1000, overlap: 100 }
# Token-based chunks (for token budget control)
- id: chunk-token
steps:
- rag-chunk: { strategy: token, chunk_size: 256, overlap: 20 }
# Sentence-aware chunks (paragraph-grouped)
- id: chunk-sentence
steps:
- rag-chunk: { strategy: sentence, chunk_size: 512, overlap: 30 }
# Recursive separator hierarchy (\n\n → \n → . → " " → "")
- id: chunk-recursive
steps:
- rag-chunk: { strategy: recursive, chunk_size: 1024, overlap: 100 }
# Semantic — split on embedding similarity breakpoints
- id: chunk-semantic
steps:
- rag-chunk:
strategy: semantic
threshold: 0.5
min_size: 100
chunk_size: 2048
# Code — language-aware function/class splitting
- id: chunk-code
steps:
- rag-chunk: { strategy: code, language: python, chunk_size: 4096 }
# Markdown — heading hierarchy
- id: chunk-markdown
steps:
- rag-chunk: { strategy: markdown, retain_headers: true }
# HTML — block-level tag boundaries
- id: chunk-html
steps:
- rag-chunk: { strategy: html, chunk_size: 1024 }
# Sliding window — overlapping windows
- id: chunk-sliding
steps:
- rag-chunk:
strategy: sliding_window
window_size: 512
stride: 256
unit: words # chars or words
# Hierarchical — parent-child for auto-merging
- id: chunk-hierarchical
steps:
- rag-chunk:
strategy: hierarchical
levels: [2048, 512, 128]
Recipe 31: All 10 Vector Store Backends¶
workflows:
# Milvus
- id: store-milvus
steps:
- rag-store:
store: milvus
url: "http://milvus:19530"
collection: documents
# Pinecone
- id: store-pinecone
steps:
- rag-store:
store: pinecone
api_key: "${PINECONE_API_KEY}"
index: documents
environment: us-east-1
# Qdrant
- id: store-qdrant
steps:
- rag-store:
store: qdrant
url: "http://qdrant:6333"
collection: documents
# Elasticsearch
- id: store-elasticsearch
steps:
- rag-store:
store: elasticsearch
connection: "http://es:9200"
collection: documents
# Weaviate
- id: store-weaviate
steps:
- rag-store:
store: weaviate
url: "http://weaviate:8080"
class: Document
# pgvector (PostgreSQL)
- id: store-pgvector
steps:
- rag-store:
store: pgvector
connection: "postgres://user:pass@db:5432/rag"
table: embeddings
dimensions: 1536
# Redis (RediSearch)
- id: store-redis
steps:
- rag-store:
store: redis
addr: "redis:6379"
index: documents
prefix: "doc:"
# ChromaDB
- id: store-chromadb
steps:
- rag-store:
store: chromadb
url: "http://chroma:8000"
collection: documents
# OpenSearch (k-NN plugin)
- id: store-opensearch
steps:
- rag-store:
store: opensearch
url: "https://opensearch:9200"
index: documents
username: admin
password: "${OPENSEARCH_PASS}"
# In-Memory (development/testing)
- id: store-memory
steps:
- rag-store:
store: memory
Recipe 32: All 9 Retrieval Strategies¶
workflows:
# Simple — direct vector similarity
- id: retrieve-simple
steps:
- rag-retrieve: { strategy: simple, top_k: 10 }
# BM25 — keyword/term frequency
- id: retrieve-bm25
steps:
- rag-retrieve: { strategy: bm25, top_k: 10 }
# Fusion — Reciprocal Rank Fusion across multiple retrievers
- id: retrieve-fusion
steps:
- rag-retrieve: { strategy: fusion, top_k: 10 }
# Hybrid — weighted vector + keyword combination
- id: retrieve-hybrid
steps:
- rag-retrieve:
strategy: hybrid
top_k: 20
vector_weight: 0.7
keyword_weight: 0.3
# MMR — Maximal Marginal Relevance (diversity)
- id: retrieve-mmr
steps:
- rag-retrieve:
strategy: mmr
top_k: 10
lambda: 0.7 # 1.0 = pure relevance, 0.0 = pure diversity
# Ensemble — weighted RRF across multiple retrievers
- id: retrieve-ensemble
steps:
- rag-retrieve:
strategy: ensemble
top_k: 10
# Auto-Merging — groups child chunks by parent
- id: retrieve-auto-merging
steps:
- rag-retrieve:
strategy: auto-merging
top_k: 10
threshold: 0.6
# Router — query-based retriever selection
- id: retrieve-router
steps:
- rag-retrieve:
strategy: router
top_k: 10
# Parent — retrieve parent document of matching chunks
- id: retrieve-parent
steps:
- rag-retrieve: { strategy: parent, top_k: 5 }
Recipe 33: RAG Postprocessing — Rerankers, Filters, Reorder¶
All 16 postprocessors.
workflows:
- id: full-postprocess
trigger:
uri: http-server://0.0.0.0:8080/api/ask
steps:
- rag-retrieve: { strategy: hybrid, top_k: 50 }
- rag-postprocess:
steps:
# API-based rerankers
- type: cohere-rerank
params:
api_key: "${COHERE_API_KEY}"
model: rerank-english-v3.0
top_n: "20"
- type: jina-rerank
params:
api_key: "${JINA_API_KEY}"
model: jina-reranker-v2-base-multilingual
top_n: "15"
# Score-based filters
- type: similarity-filter
params:
min_score: "0.3"
# Metadata filters (eq, neq, gt, lt, contains, exists)
- type: multi-metadata-filter
mode: all # all or any
filters:
- key: source
operator: eq
value: documentation
- key: updated_at
operator: gt
value: "2025-01-01"
- key: category
operator: contains
value: guide
# PII masking
- type: pii-mask
# Diversity filter (limit results per source document)
- type: diversity
params:
max_per_document: "3"
# Recency boost
- type: recency
params:
timestamp_key: created_at
boost_factor: "0.95"
# Long-context reorder ("Lost in the Middle")
- type: long-context-reorder
# Top-K limit
- type: top-k
params:
k: "5"
- rag-synthesize:
mode: compact
model: { provider: openai, model: gpt-4o }
Recipe 34: All 5 RAG Synthesis Modes¶
workflows:
# Default — single prompt with all context
- id: synth-default
steps:
- rag-synthesize:
mode: default
model: { provider: openai, model: gpt-4o }
# Compact — stuff as much context as fits, refine overflow
- id: synth-compact
steps:
- rag-synthesize:
mode: compact
max_chunks: 10
model: { provider: openai, model: gpt-4o }
# Refine — iterative refinement, one chunk at a time
- id: synth-refine
steps:
- rag-synthesize:
mode: refine
model: { provider: openai, model: gpt-4o }
# Tree Summarize — recursive hierarchical summarization
- id: synth-tree
steps:
- rag-synthesize:
mode: tree_summarize
chunks_per_level: 5
model: { provider: openai, model: gpt-4o }
# Accumulate — parallel per-chunk answers, then combine
- id: synth-accumulate
steps:
- rag-synthesize:
mode: accumulate
model: { provider: openai, model: gpt-4o }
Recipe 35: RAG with 10 Embedding Providers¶
workflows:
# OpenAI Embeddings
- id: embed-openai
steps:
- rag-embed:
provider: openai
model: text-embedding-3-small # or text-embedding-3-large
api_key: "${OPENAI_API_KEY}"
# Ollama (local)
- id: embed-ollama
steps:
- rag-embed:
provider: ollama
model: nomic-embed-text
base_url: "http://localhost:11434"
# Mistral
- id: embed-mistral
steps:
- rag-embed:
provider: mistral
model: mistral-embed
api_key: "${MISTRAL_API_KEY}"
# Vertex / Gemini
- id: embed-vertex
steps:
- rag-embed:
provider: vertex
model: text-embedding-004
api_key: "${GOOGLE_API_KEY}"
# AWS Bedrock
- id: embed-bedrock
steps:
- rag-embed:
provider: bedrock
model: amazon.titan-embed-text-v2:0
base_url: "https://bedrock-runtime.us-east-1.amazonaws.com"
# Jina
- id: embed-jina
steps:
- rag-embed:
provider: jina
model: jina-embeddings-v3
api_key: "${JINA_API_KEY}"
# HuggingFace Inference
- id: embed-huggingface
steps:
- rag-embed:
provider: huggingface
model: BAAI/bge-large-en-v1.5
api_key: "${HF_API_KEY}"
# Nomic
- id: embed-nomic
steps:
- rag-embed:
provider: nomic
model: nomic-embed-text-v1.5
api_key: "${NOMIC_API_KEY}"
# Cohere
- id: embed-cohere
steps:
- rag-embed:
provider: cohere
model: embed-english-v3.0
api_key: "${COHERE_API_KEY}"
# Voyage
- id: embed-voyage
steps:
- rag-embed:
provider: voyage
model: voyage-3
api_key: "${VOYAGE_API_KEY}"
Recipe 36: Multimodal RAG with CLIP Embeddings¶
Cross-modal image+text search using CLIP embeddings.
workflows:
# Ingest images with descriptions
- id: image-ingest
trigger:
uri: http-server://0.0.0.0:8080/api/ingest-images
steps:
- rag-ingest:
reader: image
vision_model:
provider: openai
model: gpt-4o
api_key: "${OPENAI_API_KEY}"
- rag-embed:
provider: clip
model: ViT-B/32
- rag-store:
store: qdrant
url: "http://qdrant:6333"
collection: multimodal
# Ingest audio with transcription
- id: audio-ingest
trigger:
uri: http-server://0.0.0.0:8080/api/ingest-audio
steps:
- rag-ingest:
reader: audio
transcription_model:
provider: openai
model: whisper-1
api_key: "${OPENAI_API_KEY}"
- rag-chunk: { strategy: sentence, chunk_size: 512 }
- rag-embed:
provider: openai
model: text-embedding-3-small
- rag-store:
store: qdrant
url: "http://qdrant:6333"
collection: audio-transcripts
# Cross-modal search (text query finds images)
- id: multimodal-search
trigger:
uri: http-server://0.0.0.0:8080/api/search
steps:
- rag-retrieve:
strategy: simple
top_k: 10
store: qdrant
url: "http://qdrant:6333"
collection: multimodal
- rag-synthesize:
mode: default
model: { provider: openai, model: gpt-4o }
Recipe 37: Saga Pattern — Distributed Transaction¶
Orchestrate a multi-step transaction with compensation on failure.
workflows:
- id: book-trip
trigger:
uri: http-server://0.0.0.0:8080/api/book-trip
steps:
- unmarshal: { format: json }
- saga:
steps:
- target: direct://book-flight
- target: direct://book-hotel
- target: direct://book-car
compensate:
- target: direct://cancel-car
- target: direct://cancel-hotel
- target: direct://cancel-flight
- id: book-flight
trigger: { uri: direct://book-flight }
steps:
- target: "http://flight-service:8080/book?timeout=10s"
- id: cancel-flight
trigger: { uri: direct://cancel-flight }
steps:
- target: "http://flight-service:8080/cancel?timeout=10s"
- id: book-hotel
trigger: { uri: direct://book-hotel }
steps:
- target: "http://hotel-service:8080/book?timeout=10s"
- id: cancel-hotel
trigger: { uri: direct://cancel-hotel }
steps:
- target: "http://hotel-service:8080/cancel?timeout=10s"
- id: book-car
trigger: { uri: direct://book-car }
steps:
- target: "http://car-service:8080/book?timeout=10s"
- id: cancel-car
trigger: { uri: direct://cancel-car }
steps:
- target: "http://car-service:8080/cancel?timeout=10s"
If book-hotel fails, the saga automatically calls cancel-flight to compensate.
Recipe 38: Resequencer — Order Restoration¶
Reorder out-of-sequence messages by sequence number.
workflows:
- id: ordered-processing
trigger:
uri: kafka://events?brokers=kafka:9092&group=resequencer
steps:
- resequence:
starlang: 'int(headers["sequenceNumber"])'
batch-size: 100
timeout: "5s"
allow-duplicates: false
reverse: false
steps:
- target: direct://process-ordered
- id: process-ordered
trigger: { uri: direct://process-ordered }
steps:
- log: { message: "Processing seq ${header.sequenceNumber}" }
- target: "postgres://events/insert?dsn=${DATABASE_URL}"
Recipe 39: Content Normalizer¶
Normalize messages from different sources into a uniform format.
workflows:
- id: normalize-orders
trigger:
uri: http-server://0.0.0.0:8080/api/orders
steps:
- normalize:
conditions:
- starlang: '"customerName" in payload'
steps:
- transform:
starlang: |
{"customer_name": payload.customerName, "total": payload.orderTotal,
"currency": "USD", "status": "pending"}
- starlang: '"custName" in payload'
steps:
- transform:
starlang: |
{"customer_name": payload.custName, "total": payload.order_amount,
"currency": "USD", "status": "pending"}
- target: kafka://normalized-orders?brokers=kafka:9092
Recipe 40: Aggregate with Timeout¶
Collect messages in batches before processing.
workflows:
- id: batch-processor
trigger:
uri: kafka://sensor-readings?brokers=kafka:9092&group=aggregator
steps:
- aggregate:
correlation: 'headers["sensorId"]'
strategy: collect
completion-size: 100
completion-timeout: "30s"
steps:
- transform:
starlang: |
readings = payload
{
"sensor_id": readings[0].sensorId,
"count": sizeOf(readings),
"avg_value": sum(r.value for r in readings) / sizeOf(readings),
"max_value": max(r.value for r in readings),
"min_value": min(r.value for r in readings),
"window_start": readings[0].timestamp,
"window_end": readings[-1].timestamp
}
- target: "postgres://sensor_stats/insert?dsn=${DATABASE_URL}"
Recipe 41: Load Balancer¶
Distribute work across multiple downstream services.
workflows:
- id: load-balanced
trigger:
uri: http-server://0.0.0.0:8080/api/process
steps:
- load-balancer:
strategy: roundRobin # roundRobin, random, weighted, sticky
uris:
- "http://worker-1:8080/process"
- "http://worker-2:8080/process"
- "http://worker-3:8080/process"
Recipe 42: Claim Check Pattern¶
Store large payloads externally, pass a claim ticket through the route.
workflows:
- id: large-file-pipeline
trigger:
uri: http-server://0.0.0.0:8080/api/upload
steps:
# Check in: store body in S3, replace with claim ticket
- claim-check:
operation: push
store: "s3://claim-store?region=us-east-1"
# Workflow only carries the small claim ticket
- target: kafka://file-processing?brokers=kafka:9092
- log: { message: "Claim ticket: ${body}" }
- id: file-processor
trigger:
uri: kafka://file-processing?brokers=kafka:9092&group=processor
steps:
# Check out: retrieve full payload from S3
- claim-check:
operation: pop
store: "s3://claim-store?region=us-east-1"
# Now body has the original large payload
- target: direct://process-file
Recipe 43: Recipient List — Dynamic Routing¶
Route to dynamically determined destinations.
workflows:
- id: dynamic-fanout
trigger:
uri: kafka://events?brokers=kafka:9092&group=fanout
steps:
- unmarshal: { format: json }
- recipient-list:
starlang: |
destinations = []
if payload.get("notify_email"):
destinations.append("smtp://mail:587")
if payload.get("notify_slack"):
destinations.append("slack://webhook?url=" ++ payload["slack_url"])
if payload.get("store"):
destinations.append("postgres://events/insert?dsn=${DATABASE_URL}")
destinations
parallel: true
Recipe 44: Wire-Tap — Non-Intrusive Monitoring¶
Send a copy of exchanges to a monitoring endpoint without affecting the main flow.
workflows:
- id: monitored-pipeline
trigger:
uri: http-server://0.0.0.0:8080/api/orders
steps:
- wiretap: "kafka://order-audit?brokers=kafka:9092"
- unmarshal: { format: json }
- wiretap: "elasticsearch://order-events/create?url=http://es:9200"
- target: direct://process-order
Wire-tap is asynchronous — it does not block the main route.
Recipe 45: Poll Enrich — Merge External Data¶
Enrich an exchange by polling an external source mid-route.
workflows:
- id: enrich-order
trigger:
uri: kafka://orders?brokers=kafka:9092&group=enricher
steps:
- unmarshal: { format: json }
# Fetch customer details from database
- poll-enrich:
uri: "postgres://customers/query?dsn=${DATABASE_URL}"
strategy: merge
starlang: |
original = payload
enriched = enrichment.payload
original["customer_details"] = enriched
original
# Fetch inventory status
- poll-enrich:
uri: "redis://inventory?addr=redis:6379"
strategy: merge
starlang: |
payload["in_stock"] = enrichment.payload != None
payload
- target: kafka://enriched-orders?brokers=kafka:9092
Recipe 46: Splitter with Streaming¶
Split large payloads and process items in parallel.
workflows:
- id: order-splitter
trigger:
uri: http-server://0.0.0.0:8080/api/bulk-orders
steps:
- unmarshal: { format: json }
- split:
starlang: 'payload["orders"]'
parallel: true
streaming: true
steps:
- transform:
starlang: |
{
"order_id": payload.id,
"total": payload.items | map(i -> i.price * i.qty) | sum,
"item_count": sizeOf(payload.items)
}
- choice:
when:
- starlang: 'payload.total > 500'
steps:
- target: kafka://high-value-orders?brokers=kafka:9092
otherwise:
steps:
- target: kafka://standard-orders?brokers=kafka:9092
Recipe 47: gRPC Bidirectional Integration¶
Bridge HTTP to gRPC and vice versa.
workflows:
# HTTP → gRPC
- id: http-to-grpc
trigger:
uri: http-server://0.0.0.0:8080/api/users
steps:
- unmarshal: { format: json }
- marshal: { format: protobuf }
- target: "grpc://user-service:9090/UserService/GetUser"
- unmarshal: { format: protobuf }
- marshal: { format: json }
# gRPC → Kafka
- id: grpc-to-kafka
trigger:
uri: "grpc-server://0.0.0.0:9090/EventService/StreamEvents"
steps:
- unmarshal: { format: protobuf }
- marshal: { format: json }
- target: kafka://grpc-events?brokers=kafka:9092
Recipe 48: WebSocket Real-Time Processing¶
workflows:
- id: websocket-handler
trigger:
uri: "websocket-server://0.0.0.0:8080/ws/chat"
steps:
- unmarshal: { format: json }
- agent:
type: react
model:
provider: groq
model: llama-3.3-70b-versatile
api_key: "${GROQ_API_KEY}"
system_prompt: "You are a real-time chat assistant."
- marshal: { format: json }
Recipe 49: Data Format Conversions with Starlang¶
Convert between 17 data formats using Starlang.
workflows:
# JSON → YAML
- id: json-to-yaml
trigger: { uri: http-server://0.0.0.0:8080/api/convert/yaml }
steps:
- unmarshal: { format: json }
- marshal: { format: yaml }
# CSV → JSON
- id: csv-to-json
trigger: { uri: http-server://0.0.0.0:8080/api/convert/json }
steps:
- unmarshal: { format: csv }
- marshal: { format: json }
# XML → JSON with Starlang transform
- id: xml-transform
trigger: { uri: http-server://0.0.0.0:8080/api/transform }
steps:
- transform:
starlang: |
data = read(payload, "application/xml")
records = data["root"]["items"]["item"]
result = []
for item in records:
result.append({
"name": item["@name"],
"value": float(item["#text"]),
"category": item.get("@category", "unknown")
})
write(result, "application/json")
# Custom Starlang data pipeline
- id: starlang-pipeline
trigger: { uri: http-server://0.0.0.0:8080/api/pipeline }
steps:
- transform:
starlang: |
data = read(payload, "application/json")
# Filter, transform, aggregate
active = [r for r in data if r["status"] == "active"]
enriched = [{
**r,
"full_name": r["first_name"] ++ " " ++ r["last_name"],
"score": r["points"] * 1.5 if r["premium"] else r["points"]
} for r in active]
# Group by category
groups = {}
for r in enriched:
cat = r.get("category", "other")
if cat not in groups:
groups[cat] = []
groups[cat].append(r)
write({
"total": sizeOf(enriched),
"groups": groups,
"top_scorer": max(enriched, key=lambda r: r["score"])
}, "application/json")
Supported formats: json, yaml, xml, csv, toml, ini, properties, msgpack, base64, hex, url, html, markdown, protobuf, avro, parquet, text.
Recipe 50: Connector SDK — Custom API Integration¶
Use the connector:// meta-component for any of 444 built-in connectors.
workflows:
# Slack connector
- id: slack-notify
trigger: { uri: direct://notify }
steps:
- target: "connector://slack?operation=send&channel=${SLACK_CHANNEL}&token=${SLACK_TOKEN}"
# Shopify connector
- id: shopify-orders
trigger:
uri: "cron://hourly?schedule=0 * * * *"
steps:
- target: "connector://shopify?operation=list-orders&shop=${SHOPIFY_SHOP}&token=${SHOPIFY_TOKEN}"
- split: { starlang: 'payload["orders"]' }
- target: "postgres://shopify_orders/insert?dsn=${DATABASE_URL}"
# AWS S3 connector
- id: s3-via-connector
trigger: { uri: direct://s3-upload }
steps:
- target: "connector://aws-s3?operation=put-object&bucket=my-bucket®ion=us-east-1"
# Kubernetes connector
- id: k8s-pods
trigger: { uri: direct://list-pods }
steps:
- target: "connector://kubernetes-pod?operation=list&namespace=production"
- marshal: { format: json }
Categories: ai(14), aws(46), azure(21), blockchain(2), caching(6), cdc(6), communication(25), database(27), devops(12), ecommerce(13), enterprise(13), file(23), google(34), kubernetes(20), messaging(19), monitoring(10), and more.
Recipe 51: Security — JWT, OAuth2, API Key, Rate Limiting¶
workflows:
# JWT-protected route
- id: jwt-api
trigger:
uri: http-server://0.0.0.0:8080/api/protected
auth:
- type: bearer
scheme: JWT
jwks_url: "https://auth.example.com/.well-known/jwks.json"
issuer: "https://auth.example.com"
audience: "my-api"
steps:
- log: { message: "User: ${header.jwt.sub}" }
- target: direct://handler
# OAuth2 token validation
- id: oauth2-api
trigger:
uri: http-server://0.0.0.0:8080/api/oauth
auth:
- type: oauth2
introspection_url: "https://auth.example.com/oauth/introspect"
client_id: "${OAUTH_CLIENT_ID}"
client_secret: "${OAUTH_CLIENT_SECRET}"
steps:
- target: direct://handler
# API Key auth
- id: apikey-api
trigger:
uri: http-server://0.0.0.0:8080/api/simple
auth:
- type: api-key
header: X-API-Key
keys:
- "${API_KEY_1}"
- "${API_KEY_2}"
steps:
- target: direct://handler
# Rate limiting + IP allowlist
- id: rate-limited
trigger:
uri: http-server://0.0.0.0:8080/api/rate-limited
auth:
- type: rate-limit
rate: 100
period: "1m"
per: ip
- type: ip-allowlist
allow:
- "10.0.0.0/8"
- "192.168.0.0/16"
steps:
- target: direct://handler
Recipe 52: Multi-Tenancy¶
Isolate workflows and data by tenant.
helix:
tenant:
enabled: true
header: X-Tenant-ID
required: true
allowlist:
- acme-corp
- globex-inc
- initech
workflows:
- id: tenant-api
trigger:
uri: http-server://0.0.0.0:8080/api/data
steps:
- log: { message: "Tenant: ${property.helix.tenant}" }
- set-header:
name: postgres.sql
starlang: |
"SELECT * FROM data WHERE tenant_id = '" ++ properties["helix.tenant"] ++ "'"
- target: "postgres://data/query?dsn=${DATABASE_URL}"
- marshal: { format: json }
Recipe 53: Audit Logging¶
Track all exchanges for compliance.
helix:
audit:
enabled: true
store: slog
max_entries: 10000
include_body: true
include_headers: true
workflows:
- id: audited-api
trigger:
uri: http-server://0.0.0.0:8080/api/sensitive
steps:
- target: direct://process
Audit entries are accessible via GET /api/audit and logged to structured slog output.
Recipe 54: Graceful Shutdown with Drain¶
On SIGTERM/SIGINT:
1. Stop accepting new messages
2. Wait up to drain_timeout for in-flight exchanges to complete
3. Stop all consumers, then components
Recipe 55: Cluster Mode with Leader Election¶
Run distributed with singleton workflow gating.
helix:
cluster:
enabled: true
backend: redis
redis:
addr: "redis:6379"
node_id: "${HOSTNAME}"
workflows:
# Singleton — only runs on leader
- id: scheduled-cleanup
singleton: true
trigger:
uri: "cron://cleanup?schedule=0 2 * * *"
steps:
- set-header:
name: postgres.sql
constant: "DELETE FROM expired_sessions WHERE expires_at < NOW()"
- target: "postgres://sessions/execute?dsn=${DATABASE_URL}"
Cluster features: gossip-based membership, leader election, distributed locks, singleton workflow gating.
Recipe 56: Observability — Prometheus + OpenTelemetry¶
helix:
metrics:
enabled: true
prometheus:
enabled: true
addr: "0.0.0.0:9090"
path: /metrics
tracing:
enabled: true
exporter: otlp
endpoint: "http://jaeger:4318/v1/traces"
sampling_rate: 0.1
sla:
workflows:
order-pipeline:
max_latency: 500ms
max_error_rate: 0.01
payment-processor:
max_latency: 2s
max_error_rate: 0.001
Exposed metrics: helix_exchange_total, helix_exchange_duration_seconds, helix_exchange_inflight, helix_sla_violations_total, plus per-workflow p50/p95/p99 latency percentiles.
Recipe 57: Dead Letter Channel with Monitoring¶
workflows:
- id: critical-pipeline
trigger:
uri: kafka://critical-events?brokers=kafka:9092&group=critical
steps:
- target: "http://processing-service:8080/process?timeout=10s"
error-handler:
dead-letter-channel: kafka://critical-dlq?brokers=kafka:9092
max-redeliveries: 5
redelivery-delay: "1s"
backoff-multiplier: 2.0
retry-on:
- "java.net.ConnectException"
- "io/timeout"
on-redelivery:
- log: { message: "Retry ${header.HelixRedeliveryCounter} for ${header.HelixOriginalMessageId}" }
on-failure:
- wiretap: "slack://webhook?url=${SLACK_WEBHOOK}"
- log: { message: "Message sent to DLQ after 5 retries" }
# DLQ processor — manual review
- id: dlq-processor
trigger:
uri: kafka://critical-dlq?brokers=kafka:9092&group=dlq-review
steps:
- set-header: { name: Subject, constant: "DLQ Alert: Failed Message" }
- target: "smtp://mail:587?from=alerts@example.com&to=ops@example.com"
Recipe 58: MongoDB Change Stream to Elasticsearch¶
Real-time data sync using change data capture.
workflows:
- id: mongo-to-es
trigger:
uri: "mongodb://products?uri=mongodb://mongo:27017/catalog&operation=watch"
steps:
- transform:
starlang: |
change = payload
doc = change.get("fullDocument", {})
{
"id": str(doc["_id"]),
"name": doc["name"],
"price": doc["price"],
"updated_at": change["clusterTime"]
}
- choice:
when:
- starlang: 'payload.get("operationType") == "delete"'
steps:
- target: "elasticsearch://products/delete?url=http://es:9200"
otherwise:
steps:
- target: "elasticsearch://products/upsert?url=http://es:9200"
Recipe 59: FTP File Watcher to Processing Pipeline¶
workflows:
- id: ftp-ingest
trigger:
uri: "ftp://uploads?host=ftp.example.com&port=21&username=${FTP_USER}&password=${FTP_PASS}&poll=60s&include=*.csv"
steps:
- log: { message: "New file: ${header.HelixFileName}" }
- unmarshal: { format: csv }
- split:
starlang: 'payload'
steps:
- validate: 'sizeOf(payload) >= 3'
- transform:
starlang: |
{
"name": payload[0],
"email": payload[1],
"amount": float(payload[2]),
"source_file": headers["HelixFileName"]
}
- target: "postgres://imports/insert?dsn=${DATABASE_URL}"
- target: "ftp://processed?host=ftp.example.com&port=21&username=${FTP_USER}&password=${FTP_PASS}"
Recipe 60: Elasticsearch Full-Text Search API¶
workflows:
- id: search-api
trigger:
uri: http-server://0.0.0.0:8080/api/search
steps:
- unmarshal: { format: json }
- set-body:
starlang: |
{
"query": {
"bool": {
"must": [
{"multi_match": {
"query": payload["q"],
"fields": ["title^3", "content", "tags^2"]
}}
],
"filter": [
{"term": {"status": "published"}}
]
}
},
"size": payload.get("limit", 20),
"from": payload.get("offset", 0),
"highlight": {"fields": {"content": {}}}
}
- target: "elasticsearch://articles/search?url=http://es:9200"
- marshal: { format: json }
Recipe 61: Timer-Driven Periodic Tasks¶
workflows:
# Simple heartbeat every 10 seconds
- id: heartbeat
trigger:
uri: "timer://heartbeat?period=10s"
steps:
- set-body:
starlang: |
{
"service": "order-api",
"timestamp": now(),
"status": "alive"
}
- target: "redis://heartbeat?addr=redis:6379&operation=set&ttl=30s"
# Delayed start (wait 30s before first tick)
- id: delayed-init
trigger:
uri: "timer://init?period=60s&delay=30s"
steps:
- target: "http://config-service:8080/refresh?timeout=5s"
# Fire once on startup
- id: startup-task
trigger:
uri: "timer://startup?repeatCount=1"
steps:
- log: { message: "Application started — running initialization" }
- target: direct://initialize
Recipe 62: Redis Advanced Operations¶
workflows:
# Key-value operations
- id: redis-cache
trigger:
uri: http-server://0.0.0.0:8080/api/cache
steps:
- unmarshal: { format: json }
- choice:
when:
- starlang: 'payload.operation == "get"'
steps:
- set-header: { name: redis.key, starlang: 'payload.key' }
- target: "redis://cache?addr=redis:6379&operation=get"
- starlang: 'payload.operation == "set"'
steps:
- set-header: { name: redis.key, starlang: 'payload.key' }
- set-header: { name: redis.ttl, starlang: 'payload.ttl or "3600"' }
- set-body: { starlang: 'payload.value' }
- target: "redis://cache?addr=redis:6379&operation=set"
- starlang: 'payload.operation == "del"'
steps:
- set-header: { name: redis.key, starlang: 'payload.key' }
- target: "redis://cache?addr=redis:6379&operation=del"
# List operations (queue pattern)
- id: redis-queue
trigger:
uri: http-server://0.0.0.0:8080/api/queue
steps:
- unmarshal: { format: json }
- target: "redis://task-queue?addr=redis:6379&operation=lpush"
- id: redis-worker
trigger:
uri: "redis://task-queue?addr=redis:6379&operation=brpop&timeout=5"
steps:
- log: { message: "Processing task from queue" }
- target: direct://process-task
# Hash operations
- id: redis-hash
trigger: { uri: direct://user-session }
steps:
- set-header: { name: redis.key, starlang: '"session:" ++ payload.session_id' }
- target: "redis://sessions?addr=redis:6379&operation=hset"
# Pub/Sub consumer
- id: redis-subscriber
trigger:
uri: "redis://notifications?addr=redis:6379&operation=subscribe"
steps:
- log: { message: "Received: ${body}" }
- target: "websocket://0.0.0.0:8080/ws/notifications"
Recipe 63: MongoDB CRUD and Aggregation¶
workflows:
# Insert
- id: mongo-insert
trigger:
uri: http-server://0.0.0.0:8080/api/users
method: POST
steps:
- unmarshal: { format: json }
- target: "mongodb://users?uri=mongodb://mongo:27017/myapp&operation=insert"
# Find with query
- id: mongo-find
trigger:
uri: http-server://0.0.0.0:8080/api/users
method: GET
steps:
- set-body:
starlang: |
query = {}
if "name" in headers:
query["name"] = {"$regex": headers["name"], "$options": "i"}
if "status" in headers:
query["status"] = headers["status"]
query
- target: "mongodb://users?uri=mongodb://mongo:27017/myapp&operation=find"
- marshal: { format: json }
# Aggregation pipeline
- id: mongo-aggregate
trigger:
uri: http-server://0.0.0.0:8080/api/stats
steps:
- set-body:
starlang: |
[
{"$match": {"status": "active"}},
{"$group": {
"_id": "$department",
"count": {"$sum": 1},
"avg_salary": {"$avg": "$salary"}
}},
{"$sort": {"count": -1}}
]
- target: "mongodb://employees?uri=mongodb://mongo:27017/hr&operation=aggregate"
- marshal: { format: json }
# Change stream (CDC)
- id: mongo-cdc
trigger:
uri: "mongodb://orders?uri=mongodb://mongo:27017/shop&operation=watch"
steps:
- transform:
starlang: |
{
"operation": payload.operationType,
"document_id": str(payload.documentKey["_id"]),
"timestamp": str(payload.clusterTime)
}
- target: kafka://mongo-changes?brokers=kafka:9092
Recipe 64: NATS Request-Reply and Queue Groups¶
workflows:
# Publish
- id: nats-publish
trigger:
uri: http-server://0.0.0.0:8080/api/publish
steps:
- target: "nats://events.new?servers=nats://nats:4222"
# Subscribe with queue group (load balanced)
- id: nats-queue-sub
trigger:
uri: "nats://events.>?servers=nats://nats:4222&queue=worker-group"
steps:
- log: { message: "Subject: ${header.nats.subject}" }
- target: direct://process-event
# Request-reply pattern
- id: nats-request
trigger:
uri: http-server://0.0.0.0:8080/api/ask
steps:
- target: "nats://service.query?servers=nats://nats:4222&requestReply=true&timeout=5s"
- marshal: { format: json }
# Reply service
- id: nats-responder
trigger:
uri: "nats://service.query?servers=nats://nats:4222"
steps:
- unmarshal: { format: json }
- transform:
starlang: '{"answer": "processed: " ++ str(payload)}'
- marshal: { format: json }
Recipe 65: AMQP Advanced Patterns¶
workflows:
# Topic exchange with routing keys
- id: amqp-topic-publisher
trigger:
uri: http-server://0.0.0.0:8080/api/events
steps:
- unmarshal: { format: json }
- set-header:
name: amqp.ROUTING_KEY
starlang: 'payload.category ++ "." ++ payload.severity'
- target: "amqp://events?url=amqp://rabbit:5672/&exchange=events&exchangeType=topic&durable=true"
# Subscribe to routing pattern
- id: amqp-critical-consumer
trigger:
uri: "amqp://critical-alerts?url=amqp://rabbit:5672/&exchange=events&exchangeType=topic&routingKey=*.critical&durable=true"
steps:
- target: "slack://webhook?url=${SLACK_WEBHOOK}"
# Dead-letter exchange
- id: amqp-with-dlx
trigger:
uri: "amqp://tasks?url=amqp://rabbit:5672/&durable=true&autoAck=false&deadLetterExchange=dlx&deadLetterRoutingKey=failed"
steps:
- target: "http://worker:8080/process?timeout=30s"
# Fanout exchange (broadcast)
- id: amqp-fanout
trigger: { uri: direct://broadcast }
steps:
- target: "amqp://all-services?url=amqp://rabbit:5672/&exchange=broadcast&exchangeType=fanout"
Recipe 66: SQS with Batch Processing¶
workflows:
# Send to SQS
- id: sqs-producer
trigger:
uri: http-server://0.0.0.0:8080/api/tasks
steps:
- unmarshal: { format: json }
- set-header:
name: sqs.messageGroupId
starlang: 'payload.get("group", "default")'
- target: "sqs://task-queue?region=us-east-1&accessKey=${AWS_KEY}&secretKey=${AWS_SECRET}"
# Consume from SQS with batch
- id: sqs-consumer
trigger:
uri: "sqs://task-queue?region=us-east-1&accessKey=${AWS_KEY}&secretKey=${AWS_SECRET}&maxMessages=10&waitTime=20"
steps:
- unmarshal: { format: json }
- target: direct://process-task
Recipe 67: Timeout with Fallback¶
workflows:
- id: timeout-with-fallback
trigger:
uri: http-server://0.0.0.0:8080/api/query
steps:
- timeout:
duration: "5s"
steps:
- target: "http://slow-service:8080/compute?timeout=30s"
fallback:
- log: { message: "Service timed out, returning cached result" }
- set-header: { name: redis.key, starlang: '"cache:" ++ payload.query_id' }
- target: "redis://cache?addr=redis:6379&operation=get"
- choice:
when:
- starlang: 'payload != None'
steps:
- set-header: { name: X-Cache, constant: "HIT" }
otherwise:
steps:
- set-body: { constant: '{"error": "timeout", "message": "Service unavailable"}' }
- set-header: { name: X-Cache, constant: "MISS" }
Recipe 68: Circuit Breaker with Health Monitoring¶
workflows:
- id: resilient-api
trigger:
uri: http-server://0.0.0.0:8080/api/external
steps:
- circuit-breaker:
threshold: 5 # failures before opening
half-open-after: "30s" # time before trying again
steps:
- retry:
max-attempts: 2
delay: "200ms"
steps:
- target: "http://external-api:8080/v1/data?timeout=5s"
fallback:
- log: { message: "Circuit open — returning degraded response" }
- set-body:
starlang: |
{
"status": "degraded",
"message": "External service unavailable",
"cached_at": now(),
"data": []
}
- set-header: { name: X-Degraded, constant: "true" }
Recipe 69: DoTry / DoCatch Error Handling¶
workflows:
- id: try-catch-pipeline
trigger:
uri: http-server://0.0.0.0:8080/api/process
steps:
- do-try:
steps:
- unmarshal: { format: json }
- validate: '"id" in payload and "data" in payload'
- target: "http://processing-service:8080/v1/process?timeout=10s"
do-catch:
- exceptions: ["validation"]
steps:
- set-body: { constant: '{"error": "validation_failed", "message": "Invalid input"}' }
- set-header: { name: HelixHttpResponseCode, constant: "400" }
- exceptions: ["timeout"]
steps:
- set-body: { constant: '{"error": "timeout", "message": "Processing timed out"}' }
- set-header: { name: HelixHttpResponseCode, constant: "504" }
- exceptions: ["*"]
steps:
- log: { message: "Unexpected error: ${exception.message}" }
- set-body:
starlang: |
{"error": "internal", "message": "An unexpected error occurred"}
- set-header: { name: HelixHttpResponseCode, constant: "500" }
do-finally:
- log: { message: "Request completed for ${header.X-Request-ID}" }
Recipe 70: On-Completion Callback¶
Execute steps after the exchange completes (success or failure).
workflows:
- id: tracked-pipeline
trigger:
uri: kafka://tasks?brokers=kafka:9092&group=tracked
steps:
- on-completion:
steps:
- transform:
starlang: |
{
"route_id": "tracked-pipeline",
"exchange_id": headers.get("HelixExchangeId", "unknown"),
"status": "failed" if properties.get("HelixExceptionCaught", "") != "" else "success",
"duration_ms": headers.get("HelixProcessingTime", 0),
"timestamp": now()
}
- target: "elasticsearch://exchange-log/create?url=http://es:9200"
- unmarshal: { format: json }
- target: "http://processor:8080/handle?timeout=10s"
Recipe 71: Filter — Selective Processing¶
workflows:
- id: filter-pipeline
trigger:
uri: kafka://raw-events?brokers=kafka:9092&group=filter
steps:
- unmarshal: { format: json }
# Only process events matching the condition
- filter:
starlang: 'payload.get("type") in ["purchase", "signup", "upgrade"]'
steps:
- target: kafka://important-events?brokers=kafka:9092
# Negative filter — skip known bots
- filter:
starlang: 'payload.get("user_agent", "") not in ["bot", "crawler", "spider"]'
steps:
- target: "postgres://analytics/insert?dsn=${DATABASE_URL}"
Recipe 72: Dynamic Router — Routing Slip¶
Route an exchange through a dynamic sequence of endpoints.
workflows:
- id: dynamic-pipeline
trigger:
uri: http-server://0.0.0.0:8080/api/process
steps:
- unmarshal: { format: json }
# Set routing slip based on document type
- set-header:
name: routing-slip
starlang: |
steps = ["direct://validate"]
if payload.get("type") == "invoice":
steps.append("direct://enrich-invoice")
steps.append("direct://calculate-tax")
elif payload.get("type") == "receipt":
steps.append("direct://parse-receipt")
steps.append("direct://store")
steps.append("direct://notify")
",".join(steps)
- routing-slip:
starlang: 'headers["routing-slip"]'
delimiter: ","
Recipe 73: Multicast with Custom Aggregation¶
Send to multiple destinations, aggregate with custom strategy.
workflows:
- id: multi-enrich
trigger:
uri: http-server://0.0.0.0:8080/api/customer-360
steps:
- unmarshal: { format: json }
- multicast:
parallel: true
timeout: "10s"
stop-on-exception: false
uris:
- direct://get-profile
- direct://get-orders
- direct://get-support-tickets
- direct://get-billing
strategy: merge-all
- marshal: { format: json }
- id: get-profile
trigger: { uri: direct://get-profile }
steps:
- target: "postgres://customers/query?dsn=${DATABASE_URL}"
- id: get-orders
trigger: { uri: direct://get-orders }
steps:
- target: "mongodb://orders?uri=mongodb://mongo:27017/shop&operation=find"
- id: get-support-tickets
trigger: { uri: direct://get-support-tickets }
steps:
- target: "elasticsearch://tickets/search?url=http://es:9200"
- id: get-billing
trigger: { uri: direct://get-billing }
steps:
- target: "http://billing-service:8080/v1/account?timeout=5s"
Recipe 74: Validate with Complex Rules¶
workflows:
- id: validated-api
trigger:
uri: http-server://0.0.0.0:8080/api/register
steps:
- unmarshal: { format: json }
# Multi-rule validation
- validate: '"email" in payload and "@" in payload.email'
- validate: '"password" in payload and sizeOf(payload.password) >= 8'
- validate: '"age" in payload and payload.age >= 18 and payload.age <= 120'
- validate: |
"username" in payload and
sizeOf(payload.username) >= 3 and
sizeOf(payload.username) <= 30 and
payload.username.isalnum()
# Conditional validation
- choice:
when:
- starlang: 'payload.get("type") == "business"'
steps:
- validate: '"company_name" in payload and payload.company_name != ""'
- validate: '"tax_id" in payload and sizeOf(payload.tax_id) >= 9'
- target: direct://create-user
Recipe 75: Header Manipulation Patterns¶
workflows:
- id: header-ops
trigger:
uri: http-server://0.0.0.0:8080/api/data
steps:
# Set headers
- set-header: { name: X-Request-ID, starlang: 'uuid()' }
- set-header: { name: X-Timestamp, starlang: 'now()' }
- set-header: { name: Content-Type, constant: "application/json" }
# Copy header
- set-header:
name: X-Original-Host
starlang: 'headers.get("Host", "unknown")'
# Remove sensitive headers before forwarding
- remove-headers:
pattern: "Authorization*"
- remove-headers:
pattern: "Cookie*"
- remove-headers:
pattern: "X-Internal-*"
# Set property (route-scoped, not forwarded)
- set-property:
name: startTime
starlang: 'now()'
- target: "http://backend:8080/api/data?timeout=10s"
# Add response headers
- set-header:
name: X-Processing-Time
starlang: 'str(now() - properties["startTime"]) ++ "ms"'
Recipe 76: Delay and Scheduled Delivery¶
workflows:
# Fixed delay
- id: delayed-processing
trigger:
uri: kafka://events?brokers=kafka:9092&group=delayed
steps:
- delay:
duration: "5s"
- target: direct://process
# Dynamic delay based on priority
- id: priority-delay
trigger:
uri: kafka://tasks?brokers=kafka:9092&group=priority
steps:
- unmarshal: { format: json }
- delay:
starlang: |
priority = payload.get("priority", "normal")
{"high": "0s", "normal": "5s", "low": "30s"}.get(priority, "10s")
- target: direct://execute-task
# Throttle with rejection
- id: throttled-api
trigger:
uri: http-server://0.0.0.0:8080/api/limited
steps:
- throttle:
rate: 10
period: "1s"
- target: direct://handler
Recipe 77: Content Enricher — HTTP Enrichment¶
workflows:
- id: enrich-order
trigger:
uri: kafka://orders?brokers=kafka:9092&group=enricher
steps:
- unmarshal: { format: json }
# Enrich with customer data
- set-header:
name: customerId
starlang: 'payload.customer_id'
- enrich:
uri: "http://customer-service:8080/v1/customers/${header.customerId}?timeout=5s"
strategy: merge
starlang: |
original = payload
customer = payload
original["customer_name"] = customer.get("name", "Unknown")
original["customer_email"] = customer.get("email", "")
original["customer_tier"] = customer.get("tier", "standard")
original
# Enrich with product details
- split:
starlang: 'payload["items"]'
steps:
- enrich:
uri: "http://product-service:8080/v1/products/${body.product_id}?timeout=5s"
strategy: merge
starlang: |
item = payload
product = payload
item["product_name"] = product["name"]
item["category"] = product["category"]
item
- target: kafka://enriched-orders?brokers=kafka:9092
Recipe 78: Multi-Provider LLM Fallback¶
Failover between LLM providers for high availability.
workflows:
- id: resilient-ai
trigger:
uri: http-server://0.0.0.0:8080/api/chat
steps:
- do-try:
steps:
# Primary: fast provider
- agent:
type: react
model:
provider: groq
model: llama-3.3-70b-versatile
api_key: "${GROQ_API_KEY}"
system_prompt: "You are a helpful assistant."
do-catch:
- exceptions: ["*"]
steps:
- log: { message: "Groq failed, falling back to OpenAI" }
- do-try:
steps:
# Secondary: reliable provider
- agent:
type: react
model:
provider: openai
model: gpt-4o
api_key: "${OPENAI_API_KEY}"
system_prompt: "You are a helpful assistant."
do-catch:
- exceptions: ["*"]
steps:
- log: { message: "OpenAI failed, falling back to Ollama" }
# Tertiary: local model
- agent:
type: react
model:
provider: ollama
model: llama3.3
base_url: "http://localhost:11434"
system_prompt: "You are a helpful assistant."
Recipe 79: RAG Evaluation Pipeline¶
Evaluate RAG quality with built-in metrics.
workflows:
- id: rag-eval
trigger:
uri: http-server://0.0.0.0:8080/api/evaluate
steps:
- unmarshal: { format: json }
# Run RAG query
- rag-retrieve:
strategy: hybrid
top_k: 10
store: pgvector
connection: "${DATABASE_URL}"
- rag-postprocess:
steps:
- type: cohere-rerank
params:
api_key: "${COHERE_API_KEY}"
top_n: "5"
- rag-synthesize:
mode: compact
model: { provider: openai, model: gpt-4o }
# Evaluate quality
- rag-eval:
metrics:
- faithfulness # is answer grounded in context?
- answer_relevancy # does answer address the question?
- context_relevancy # is retrieved context relevant?
- hit_rate # were relevant docs retrieved?
- mrr # Mean Reciprocal Rank
- ndcg # Normalized Discounted Cumulative Gain
model: { provider: openai, model: gpt-4o }
ground_truth_field: expected_answer
- marshal: { format: json }
Recipe 80: Knowledge Graph RAG¶
Entity extraction and multi-hop graph traversal for complex queries.
workflows:
# Build knowledge graph from documents
- id: kg-ingest
trigger:
uri: http-server://0.0.0.0:8080/api/kg/ingest
steps:
- rag-ingest:
reader: directory
path: ./knowledge
recursive: true
- rag-chunk: { strategy: sentence, chunk_size: 512 }
# Extract entities and relationships, store in knowledge graph
- rag-graph-build:
model: { provider: openai, model: gpt-4o }
store: memory # or neo4j, or embedded
- rag-embed: { provider: openai, model: text-embedding-3-small }
- rag-store: { store: pgvector, connection: "${DATABASE_URL}" }
# Query with graph traversal
- id: kg-query
trigger:
uri: http-server://0.0.0.0:8080/api/kg/ask
steps:
# Vector retrieval for initial context
- rag-retrieve: { strategy: simple, top_k: 5 }
# Multi-hop graph traversal for connected context
- rag-graph-query:
hops: 3
top_k: 20
- rag-synthesize:
mode: refine
model: { provider: openai, model: gpt-4o }
Recipe 81: Cron with Multiple Schedules¶
workflows:
# Every 5 minutes
- id: health-check
trigger:
uri: "cron://health?schedule=*/5 * * * *"
steps:
- target: "http://api:8080/health?timeout=5s"
- choice:
when:
- starlang: 'headers.get("HelixHttpResponseCode") != "200"'
steps:
- target: "slack://webhook?url=${SLACK_WEBHOOK}"
# Weekdays at 6 AM
- id: morning-digest
trigger:
uri: "cron://digest?schedule=0 6 * * 1-5"
steps:
- target: direct://build-digest
- target: "smtp://mail:587?from=digest@example.com&to=team@example.com"
# First day of month
- id: monthly-report
trigger:
uri: "cron://monthly?schedule=0 9 1 * *"
singleton: true
steps:
- target: direct://generate-monthly-report
# Every 30 seconds (sub-minute via timer)
- id: fast-poll
trigger:
uri: "timer://poll?period=30s"
steps:
- target: "http://external-api:8080/changes?timeout=5s"
- filter:
starlang: 'payload != "[]"'
steps:
- target: kafka://changes?brokers=kafka:9092
Recipe 82: Health Check Endpoint¶
helix:
health:
enabled: true
addr: "0.0.0.0:8081"
path: /health
liveness: /healthz
readiness: /readyz
workflows:
- id: api-route
trigger:
uri: http-server://0.0.0.0:8080/api/data
steps:
- target: direct://handler
GET /health returns engine state, inflight count, route status, component health.
GET /healthz for Kubernetes liveness probe. GET /readyz for readiness probe.
Recipe 83: Webhook Receiver with Signature Verification¶
workflows:
- id: github-webhook
trigger:
uri: http-server://0.0.0.0:8080/webhooks/github
steps:
# Verify GitHub webhook signature
- transform:
starlang: |
signature = headers["X-Webhook-Signature"]
expected = hmac_sha256(payload, headers.get("X-Webhook-Secret", ""))
signature == expected
- unmarshal: { format: json }
- choice:
when:
- starlang: 'headers["X-GitHub-Event"] == "push"'
steps:
- target: direct://handle-push
- starlang: 'headers["X-GitHub-Event"] == "pull_request"'
steps:
- target: direct://handle-pr
- starlang: 'headers["X-GitHub-Event"] == "issues"'
steps:
- target: direct://handle-issue
- id: handle-push
trigger: { uri: direct://handle-push }
steps:
- transform:
starlang: |
commits = payload["commits"]
branch = payload["ref"].split("/")[-1]
{
"text": "*Push to {}*: {} commits by {}".format(
branch, sizeOf(commits), payload["pusher"]["name"]
)
}
- target: "slack://webhook?url=${SLACK_WEBHOOK}"
Recipe 84: ETL Pipeline — Extract, Transform, Load¶
workflows:
# Extract from multiple sources
- id: etl-extract
trigger:
uri: "cron://etl?schedule=0 2 * * *"
singleton: true
steps:
- multicast:
parallel: true
uris:
- direct://extract-postgres
- direct://extract-mongo
- direct://extract-api
strategy: collect-all
# Transform
- transform:
starlang: |
all_records = []
for source in payload:
for record in source:
all_records.append({
"id": record.get("id") or str(record.get("_id")),
"name": record.get("name", ""),
"email": record.get("email", ""),
"source": record.get("_source", "unknown"),
"normalized_at": now()
})
# Deduplicate by email
seen = set()
unique = []
for r in all_records:
if r["email"] not in seen:
seen.add(r["email"])
unique.append(r)
unique
# Load
- split:
starlang: 'payload'
parallel: true
steps:
- target: "postgres://unified_customers/upsert?dsn=${DW_DATABASE_URL}"
- log: { message: "ETL complete: ${body} records loaded" }
- id: extract-postgres
trigger: { uri: direct://extract-postgres }
steps:
- set-header:
name: postgres.sql
constant: "SELECT id, name, email, 'postgres' as _source FROM customers WHERE updated_at > NOW() - INTERVAL '1 day'"
- target: "postgres://customers/query?dsn=${SOURCE_DB_URL}"
- id: extract-mongo
trigger: { uri: direct://extract-mongo }
steps:
- set-body:
starlang: '{"updated_at": {"$gte": {"$date": now() - 86400}}}'
- target: "mongodb://users?uri=mongodb://mongo:27017/app&operation=find"
- id: extract-api
trigger: { uri: direct://extract-api }
steps:
- target: "http://crm-api:8080/v1/contacts?updated_since=yesterday&timeout=30s"
- unmarshal: { format: json }
Recipe 85: Kubernetes Operator Pattern¶
Monitor and manage Kubernetes resources.
workflows:
# Watch pod events
- id: k8s-pod-watcher
trigger:
uri: "connector://kubernetes-pod?operation=watch&namespace=production"
steps:
- unmarshal: { format: json }
- choice:
when:
- starlang: 'payload.status.phase == "Failed"'
steps:
- transform:
starlang: |
{
"text": ":red_circle: Pod {} failed in namespace {}".format(
payload.metadata.name, payload.metadata.namespace
),
"reason": payload.status.get("reason", "Unknown")
}
- target: "slack://webhook?url=${SLACK_WEBHOOK}"
- starlang: 'payload.status.phase == "Running" and payload.metadata.labels.get("app") == "critical"'
steps:
- target: "elasticsearch://k8s-events/create?url=http://es:9200"
# Scale deployment based on queue depth
- id: auto-scaler
trigger:
uri: "timer://scale-check?period=30s"
steps:
- target: "redis://task-queue?addr=redis:6379&operation=llen"
- transform:
starlang: |
queue_depth = int(payload)
replicas = max(1, min(10, queue_depth // 100))
{"replicas": replicas}
- target: "connector://kubernetes-deployment?operation=scale&namespace=production&name=worker"
Recipe 86: Event Sourcing Pattern¶
workflows:
# Command handler — validate and emit events
- id: command-handler
trigger:
uri: http-server://0.0.0.0:8080/api/commands
steps:
- unmarshal: { format: json }
- validate: '"type" in payload and "aggregate_id" in payload'
# Store event in event store (append-only)
- transform:
starlang: |
{
"event_id": uuid(),
"aggregate_id": payload.aggregate_id,
"event_type": payload.type,
"data": payload.data,
"timestamp": now(),
"version": payload.get("expected_version", 0) + 1
}
- target: "postgres://events/insert?dsn=${DATABASE_URL}"
# Publish event for projections
- target: kafka://domain-events?brokers=kafka:9092
- set-body: { starlang: '{"status": "accepted", "event_id": payload.event_id}' }
- marshal: { format: json }
# Projection — build read model from events
- id: order-projection
trigger:
uri: kafka://domain-events?brokers=kafka:9092&group=order-projection
steps:
- unmarshal: { format: json }
- filter:
starlang: 'payload.event_type.startswith("order.")'
- choice:
when:
- starlang: 'payload.event_type == "order.created"'
steps:
- target: "postgres://order_view/insert?dsn=${DATABASE_URL}"
- starlang: 'payload.event_type == "order.updated"'
steps:
- target: "postgres://order_view/update?dsn=${DATABASE_URL}"
- starlang: 'payload.event_type == "order.cancelled"'
steps:
- set-header:
name: postgres.sql
starlang: |
"UPDATE order_view SET status='cancelled' WHERE id='" ++ payload.aggregate_id ++ "'"
- target: "postgres://order_view/execute?dsn=${DATABASE_URL}"
Recipe 87: API Gateway Pattern¶
workflows:
# Gateway with auth, rate limiting, routing
- id: api-gateway
trigger:
uri: http-server://0.0.0.0:8080/api/v1
auth:
- type: bearer
scheme: JWT
jwks_url: "https://auth.example.com/.well-known/jwks.json"
steps:
- throttle:
rate: 100
period: "1m"
per: header:X-User-ID
- log: { message: "${header.method} ${header.path} by ${header.jwt.sub}" }
- choice:
when:
- starlang: 'headers["path"].startswith("/api/v1/users")'
steps:
- remove-headers: { pattern: "Authorization*" }
- target: "http://user-service:8080${header.path}?timeout=10s"
- starlang: 'headers["path"].startswith("/api/v1/orders")'
steps:
- remove-headers: { pattern: "Authorization*" }
- target: "http://order-service:8080${header.path}?timeout=10s"
- starlang: 'headers["path"].startswith("/api/v1/products")'
steps:
- remove-headers: { pattern: "Authorization*" }
- target: "http://product-service:8080${header.path}?timeout=10s"
otherwise:
steps:
- set-body: { constant: '{"error": "not_found", "message": "Unknown endpoint"}' }
- set-header: { name: HelixHttpResponseCode, constant: "404" }
Recipe 88: CQRS — Command Query Responsibility Segregation¶
workflows:
# Command side — writes to primary database + emits events
- id: create-order
trigger:
uri: http-server://0.0.0.0:8080/api/orders
method: POST
steps:
- unmarshal: { format: json }
- validate: '"items" in payload and sizeOf(payload.items) > 0'
- transform:
starlang: |
{
**payload,
"id": uuid(),
"status": "pending",
"created_at": now()
}
- target: "postgres://orders/insert?dsn=${WRITE_DB_URL}"
- target: kafka://order-events?brokers=kafka:9092
- marshal: { format: json }
# Query side — reads from optimized read replica
- id: list-orders
trigger:
uri: http-server://0.0.0.0:8080/api/orders
method: GET
steps:
- set-header:
name: postgres.sql
starlang: |
status = headers.get("status", "")
limit = headers.get("limit", "20")
sql = "SELECT * FROM order_view"
if status:
sql = sql ++ " WHERE status = '" ++ status ++ "'"
sql = sql ++ " ORDER BY created_at DESC LIMIT " ++ limit
sql
- target: "postgres://order_view/query?dsn=${READ_DB_URL}"
- marshal: { format: json }
# Materializer — update read model from events
- id: order-materializer
trigger:
uri: kafka://order-events?brokers=kafka:9092&group=materializer
steps:
- unmarshal: { format: json }
- target: "postgres://order_view/upsert?dsn=${READ_DB_URL}"
- target: "elasticsearch://orders/upsert?url=http://es:9200"
- target: "redis://order-cache?addr=redis:6379&operation=set&ttl=3600"
Recipe 89: Outbox Pattern — Reliable Event Publishing¶
workflows:
# Business operation + outbox write in one transaction
- id: place-order
trigger:
uri: http-server://0.0.0.0:8080/api/orders
steps:
- unmarshal: { format: json }
- transform:
starlang: |
order = {**payload, "id": uuid(), "status": "created"}
# Both the order and outbox event will be in the same transaction
{
"order": order,
"outbox_event": {
"id": uuid(),
"aggregate_type": "order",
"aggregate_id": order["id"],
"event_type": "order_created",
"payload": write(order, "application/json"),
"created_at": now()
}
}
- target: "postgres://order_with_outbox/insert?dsn=${DATABASE_URL}"
- marshal: { format: json }
# Outbox poller — reliably publish events from outbox table
- id: outbox-poller
singleton: true
trigger:
uri: "timer://outbox?period=1s"
steps:
- set-header:
name: postgres.sql
constant: |
SELECT * FROM outbox_events
WHERE published = false
ORDER BY created_at
LIMIT 100
- target: "postgres://outbox_events/query?dsn=${DATABASE_URL}"
- split:
starlang: 'payload'
steps:
- target: kafka://domain-events?brokers=kafka:9092
- set-header:
name: postgres.sql
starlang: '"UPDATE outbox_events SET published = true WHERE id = ''" ++ payload.id ++ "''"'
- target: "postgres://outbox_events/execute?dsn=${DATABASE_URL}"
Recipe 90: Bulkhead Pattern — Isolation¶
Isolate route processing pools to prevent one slow consumer from starving others.
workflows:
# Critical path — dedicated high-priority pool
- id: critical-api
trigger:
uri: http-server://0.0.0.0:8080/api/critical
steps:
- target: seda://critical-pool?bufferSize=100&concurrency=10&overflowPolicy=fail
- set-body: { constant: '{"status": "accepted"}' }
- id: critical-worker
trigger:
uri: seda://critical-pool?concurrency=10
steps:
- target: "http://payment-service:8080/charge?timeout=10s"
# Non-critical — separate pool, can be throttled
- id: analytics-api
trigger:
uri: http-server://0.0.0.0:8080/api/analytics
steps:
- target: seda://analytics-pool?bufferSize=10000&concurrency=3&overflowPolicy=drop
- set-body: { constant: '{"status": "accepted"}' }
- id: analytics-worker
trigger:
uri: seda://analytics-pool?concurrency=3
steps:
- throttle: { rate: 20, period: "1s" }
- target: "http://analytics-service:8080/track?timeout=30s"
Recipe 91: Stream Large File Through Pipeline¶
Process a multi-gigabyte file without loading it into memory.
workflows:
- id: large-file-pipeline
trigger:
uri: "file://incoming/?include=*.csv&streaming=true&poll=60s"
steps:
- log: { message: "Streaming file: ${header.HelixFileName} (${header.HelixFileSize} bytes)" }
# StreamBody flows through — constant memory
- target: "file://processed/?streaming=true"
- log: { message: "Written ${header.HelixBytesWritten} bytes" }
Recipe 92: S3 Multipart Upload of Large Files¶
Upload multi-gigabyte files to S3 with concurrent parts.
workflows:
- id: large-s3-upload
trigger:
uri: http-server://0.0.0.0:8080/api/upload
steps:
# Body arrives as StreamBody from HTTP request
- log: { message: "Uploading ${header.Content-Length} bytes" }
# Multipart upload: 20MB parts, 10 concurrent uploads
- target: "s3://data-lake?region=us-east-1&multipart=true&partSize=20MB&concurrency=10"
- set-body:
constant: '{"status": "uploaded"}'
Recipe 93: Streaming HTTP Download and Forward¶
Download a large file and forward it without buffering.
workflows:
- id: download-and-store
trigger:
uri: "cron://hourly?schedule=0 * * * *"
steps:
# Download as StreamBody (no buffering)
- target: "http://data-provider.example.com/export/daily.csv?streaming=true&timeout=300s"
- log: { message: "Downloaded stream, forwarding to S3" }
# Stream directly to S3 (constant memory for any file size)
- target: "s3://imports/daily.csv?region=us-east-1"
Recipe 94: Stream-Process Large JSON Array¶
Process a 10GB JSON array of records without loading it all into memory.
workflows:
- id: bulk-json-processor
trigger:
uri: http-server://0.0.0.0:8080/api/bulk-import
steps:
- unmarshal: { format: json }
# Split the JSON array; each element processed individually
- split:
starlang: 'payload'
streaming: true
steps:
- validate: '"id" in payload and "email" in payload'
- transform:
starlang: |
{
**payload,
"imported_at": now(),
"source": "bulk-import"
}
- target: "postgres://users/upsert?dsn=${DATABASE_URL}"
- set-body:
constant: '{"status": "complete"}'
Recipe 95: Large File Copy Between Storage Systems¶
Stream a file from FTP to S3 with constant memory.
workflows:
- id: ftp-to-s3-stream
trigger:
uri: "ftp://exports?host=ftp.example.com&port=21&username=${FTP_USER}&password=${FTP_PASS}&streaming=true&poll=300s"
steps:
- log: { message: "Streaming ${header.HelixFileName} from FTP to S3" }
# File streams directly: FTP reader → S3 writer
# No intermediate buffering regardless of file size
- target: "s3://archive/${header.HelixFileName}?region=us-east-1&multipart=auto"
- transform:
starlang: |
{
"file": headers["HelixFileName"],
"size": headers["HelixFileSize"],
"status": "archived"
}
- target: kafka://file-events?brokers=kafka:9092
20. Large File & Streaming Support¶
Helix supports processing large files (multi-gigabyte) with constant memory through its StreamBody abstraction. Instead of loading entire payloads into memory, data flows through workflows as io.Reader streams.
StreamBody¶
StreamBody wraps any io.Reader with metadata (size, MIME type, filename) and flows through exchanges like any other body type.
# Components automatically detect StreamBody and use streaming I/O:
# - File component: reads/writes via io.Copy (no os.ReadFile)
# - HTTP component: chunked transfer encoding for unknown-size bodies
# - S3 component: multipart upload for files > 100MB
# - JSON processor: element-by-element array processing
Key properties:
| Property | Description |
|---|---|
Size() |
Total bytes (-1 if unknown, e.g., pipes) |
MediaType() |
MIME type (application/json, image/png, etc.) |
Name() |
Optional filename |
IsConsumed() |
Whether the stream has been fully read |
IsSeekable() |
Whether the stream can be reset/rewound |
Exchange helpers:
| Function | Description |
|---|---|
SetStreamBody(ex, sb) |
Set StreamBody as exchange body with headers |
GetStreamBody(ex) |
Extract StreamBody (auto-wraps []byte/string/io.Reader) |
IsStreamBody(ex) |
Check if exchange body is a StreamBody |
Streaming File I/O¶
Read and write large files without buffering the entire content:
workflows:
# Read large files as streams
- id: process-large-file
trigger:
uri: "file://uploads/?include=*.csv&streaming=true"
steps:
# Body is a StreamBody — processed in constant memory
- split:
starlang: 'payload'
streaming: true
steps:
- target: "postgres://records/insert?dsn=${DATABASE_URL}"
# Write large output as stream
- id: generate-report
trigger:
uri: "cron://daily?schedule=0 2 * * *"
steps:
- target: "postgres://orders/query?dsn=${DATABASE_URL}"
- target: "file://reports/?streaming=true"
StreamingFileReader opens files as StreamBody (wraps os.Open, not os.ReadFile):
- Sets headers: HelixFileName, HelixFilePath, HelixFileDir, HelixFileExt, HelixFileSize, HelixFileModified
- File handle is closed when the StreamBody is closed
StreamingFileProducer writes using io.Copy:
- For StreamBody: streams directly from reader (constant memory)
- For io.Reader: same streaming behavior
- For []byte/string: standard write (fallback)
- Sets headers: HelixBytesWritten, HelixFilePath
S3 Multipart Upload¶
Automatically uses multipart upload for large objects:
workflows:
- id: upload-large-file
trigger:
uri: http-server://0.0.0.0:8080/api/upload
steps:
# Files > 100MB automatically use multipart upload
- target: "s3://my-bucket?region=us-east-1&multipart=true&partSize=20MB&concurrency=10"
| Parameter | Default | Description |
|---|---|---|
multipart |
auto |
auto (size-based), true (always), false (never) |
partSize |
10MB |
Size of each upload part (minimum 5MB) |
concurrency |
5 |
Number of concurrent part uploads |
threshold |
100MB |
Size above which multipart is triggered |
How it works:
1. InitiateMultipartUpload — gets upload ID
2. Upload parts concurrently (bounded by concurrency setting)
3. CompleteMultipartUpload on success
4. AbortMultipartUpload on any failure (automatic cleanup)
Streaming HTTP¶
Send and receive large HTTP payloads without buffering:
workflows:
# Download large file and forward
- id: proxy-large-file
trigger:
uri: http-server://0.0.0.0:8080/api/proxy
steps:
# Response is a StreamBody — not buffered in memory
- target: "http://file-server:8080/large-file.zip?streaming=true"
# Forward stream directly to S3
- target: "s3://archive-bucket?region=us-east-1"
StreamingHTTPProducer: - StreamBody/io.Reader body → chunked transfer encoding (no Content-Length needed) - Known-size body → standard Content-Length header - Responses returned as StreamBody for downstream streaming - Headers copied from exchange to HTTP request
DownloadToStream: Convenience function for HTTP GET → StreamBody.
Streaming JSON Processing¶
Process large JSON arrays element-by-element without loading the entire array:
workflows:
- id: process-large-json
trigger:
uri: http-server://0.0.0.0:8080/api/bulk
steps:
- unmarshal: { format: json }
# Split the JSON array and process each element individually
- split:
starlang: 'payload'
streaming: true
steps:
- transform:
starlang: |
{
"id": payload.id,
"processed": true,
"timestamp": now()
}
- target: kafka://processed-items?brokers=kafka:9092
Uses the split step with streaming: true:
- Unmarshals the JSON array first
- Split iterates elements one at a time
- Creates child exchange per element
- Propagates parent headers to child exchanges
- Supports context cancellation mid-stream
ChunkedReader¶
Process any stream in fixed-size chunks:
# Conceptual — used internally by components
# ChunkSize default: 64KB
# Callback receives (chunk []byte, offset int64)
Useful for: - Checksum computation on large files - Progress tracking during transfers - Chunk-based encryption/compression
LimitedStreamBody¶
Protect against oversized payloads:
workflows:
- id: limited-upload
trigger:
# Use max-body-size parameter to limit incoming stream
uri: http-server://0.0.0.0:8080/api/upload?max-body-size=500MB
steps:
- target: "s3://uploads?region=us-east-1"
Best Practices¶
- Use streaming for files > 10MB — prevents memory pressure
- Close StreamBody when done — especially for file-backed streams (releases file handles)
- Check
IsSeekable()before callingReset()— pipes and network streams cannot be rewound - Use
TeeReaderfor audit/logging — reads stream without consuming it - Set
LimitedStreamBodyon untrusted input — prevents memory exhaustion - S3 multipart for > 100MB — automatic with
multipart=auto(default) - Streaming JSON for arrays > 1000 elements — avoids materializing entire array
Appendix: Workflow Templates¶
Helix supports parameterized route templates — reusable route snippets.
// Define a template
tmpl := template.New("polling-api",
template.Param("url", "API endpoint URL"),
template.Param("interval", "Poll interval", template.Default("30s")),
template.Param("destination", "Output endpoint"),
)
tmpl.Workflow(func(params map[string]string) *workflow.Definition {
return dsl.NewRoute().
From(fmt.Sprintf("timer://poll?period=%s", params["interval"])).
To(fmt.Sprintf("http://%s", params["url"])).
To(params["destination"]).
Build()
})
// Use the template
route := tmpl.Create(map[string]string{
"url": "api.example.com/data",
"interval": "60s",
"destination": "kafka://api-data",
})
YAML Template-Instantiated Routes¶
In YAML, template-instantiated workflows are expressed as regular routes with the template parameters already resolved. You can simulate template patterns by using headers to carry template metadata:
workflows:
# Workflow that represents an instantiated template
- id: template-http-gateway
trigger:
uri: "direct://template-input"
steps:
- set-header:
name: X-Template
constant: "yaml-http-gateway"
- set-header:
name: X-Template-Port
constant: "8080"
- transform:
constant: "processed-by-template"
- target: "log://template-output"
# Template-instantiated route with Starlang transform
- id: template-with-logging
trigger:
uri: "direct://template-log-input"
steps:
- log:
message: "Template route processing request"
level: "info"
- transform:
starlang: |
%sl 2.0
output text/plain
input payload
---
"template:" ++ payload
- target: "log://template-log-output"
# Chained template workflows
- id: template-chain
trigger:
uri: "direct://template-chain-input"
steps:
- set-header:
name: X-Template-Chain
constant: "step-1"
- target: "direct://template-input"
21. OCR & Document Intelligence¶
Helix integrates with 8 OCR backends to extract text from scanned PDFs, images, and complex layouts. Each backend runs as an external service — Helix calls it via HTTP (no cgo).
Recipe 96: OCR with Docling (PDF to Markdown)¶
workflows:
- id: docling-ocr
trigger:
uri: http-server://0.0.0.0:8080/api/ocr
steps:
- rag-ingest:
parser: ocr
ocr:
backend: docling
endpoint: "http://localhost:5000"
output_format: markdown
- rag-chunk:
strategy: markdown
- rag-embed:
provider: openai
model: text-embedding-3-small
- rag-store:
store: pgvector
collection: scanned_docs
Recipe 97: OCR with MinerU (Academic Papers)¶
workflows:
- id: mineru-ocr
trigger:
uri: http-server://0.0.0.0:8080/api/ingest-paper
steps:
- rag-ingest:
parser: ocr
ocr:
backend: mineru
endpoint: "http://localhost:8888"
language: en
- rag-chunk:
strategy: semantic
threshold: 0.5
- rag-embed:
provider: jina
model: jina-embeddings-v3
- rag-store:
store: qdrant
collection: papers
Recipe 98: OCR with Marker (Fast PDF Conversion)¶
workflows:
- id: marker-ocr
trigger:
uri: http-server://0.0.0.0:8080/api/convert
steps:
- rag-ingest:
parser: ocr
ocr:
backend: marker
endpoint: "http://localhost:8001"
Recipe 99: OCR with Tesseract (Classic OCR)¶
workflows:
- id: tesseract-ocr
trigger:
uri: http-server://0.0.0.0:8080/api/scan
steps:
- rag-ingest:
parser: ocr
ocr:
backend: tesseract
endpoint: "http://localhost:8884"
language: eng
Recipe 100: OCR with Unstructured.io (Element Classification)¶
workflows:
- id: unstructured-ocr
trigger:
uri: http-server://0.0.0.0:8080/api/parse
steps:
- rag-ingest:
parser: ocr
ocr:
backend: unstructured
endpoint: "https://api.unstructured.io"
api_key: "${env:UNSTRUCTURED_API_KEY}"
Recipe 101: OCR with Azure Document Intelligence (Tables & Forms)¶
workflows:
- id: azure-ocr
trigger:
uri: http-server://0.0.0.0:8080/api/analyze
steps:
- rag-ingest:
parser: ocr
ocr:
backend: azure-doc-intelligence
endpoint: "${env:AZURE_ENDPOINT}"
api_key: "${env:AZURE_DOC_KEY}"
model: prebuilt-layout
- rag-chunk:
strategy: markdown
- rag-embed:
provider: openai
model: text-embedding-3-small
- rag-store:
store: elasticsearch
collection: invoices
Recipe 102: OCR with Vision LLM (No External Service)¶
workflows:
- id: vision-ocr
trigger:
uri: http-server://0.0.0.0:8080/api/vision-ocr
steps:
- rag-ingest:
parser: ocr
ocr:
backend: vision-llm
model:
provider: openai
model: gpt-4o
output_format: markdown
- rag-chunk:
strategy: markdown
- rag-embed:
provider: openai
model: text-embedding-3-small
- rag-store:
store: pgvector
collection: vision_docs
Recipe 103: OCR with Surya (Multilingual)¶
workflows:
- id: surya-ocr
trigger:
uri: http-server://0.0.0.0:8080/api/multilingual-ocr
steps:
- rag-ingest:
parser: ocr
ocr:
backend: surya
endpoint: "http://localhost:8885"
language: ja
Recipe 104: Table Extraction from Scanned Documents¶
workflows:
- id: table-extract
trigger:
uri: http-server://0.0.0.0:8080/api/extract-tables
steps:
- rag-ingest:
parser: ocr
ocr:
backend: azure-doc-intelligence
endpoint: "${env:AZURE_ENDPOINT}"
api_key: "${env:AZURE_DOC_KEY}"
extract_tables: true
- transform:
starlang: |
tables = payload
result = [{"table": t, "index": i} for i, t in enumerate(tables)]
payload = write(result, "application/json")
- target: kafka://extracted-tables?brokers=localhost:9092
22. Query Transformation¶
Advanced retrieval strategies that rewrite or expand queries before searching.
Recipe 105: HyDE — Hypothetical Document Embeddings¶
workflows:
- id: hyde-rag
trigger:
uri: http-server://0.0.0.0:8080/api/ask
steps:
- rag-retrieve:
strategy: hyde
top_k: 10
hyde:
model:
provider: openai
model: gpt-4o
embedding:
provider: openai
model: text-embedding-3-small
- rag-postprocess:
steps:
- type: cohere-rerank
params:
api_key: "${env:COHERE_API_KEY}"
top_n: "5"
- rag-synthesize:
mode: compact
model:
provider: openai
model: gpt-4o
Recipe 106: MultiQuery Retrieval (Multiple Phrasings)¶
workflows:
- id: multi-query-rag
trigger:
uri: http-server://0.0.0.0:8080/api/ask
steps:
- rag-retrieve:
strategy: multi_query
top_k: 10
multi_query:
model:
provider: openai
model: gpt-4o
num_queries: 3
embedding:
provider: openai
model: text-embedding-3-small
- rag-synthesize:
mode: refine
model:
provider: anthropic
model: claude-sonnet-4-20250514
Recipe 107: StepBack Prompting (Broader Retrieval)¶
workflows:
- id: stepback-rag
trigger:
uri: http-server://0.0.0.0:8080/api/ask
steps:
- rag-retrieve:
strategy: step_back
top_k: 10
step_back:
model:
provider: openai
model: gpt-4o
embedding:
provider: openai
model: text-embedding-3-small
- rag-synthesize:
mode: compact
model:
provider: groq
model: llama-3.3-70b-versatile
Recipe 108: SubQuestion Decomposition¶
workflows:
- id: subquestion-rag
trigger:
uri: http-server://0.0.0.0:8080/api/research
steps:
- rag-retrieve:
strategy: sub_question
top_k: 10
sub_question:
model:
provider: openai
model: gpt-4o
embedding:
provider: openai
model: text-embedding-3-small
- rag-synthesize:
mode: tree_summarize
model:
provider: openai
model: gpt-4o
Recipe 109: OCR + HyDE Full Pipeline¶
Combines OCR ingestion with HyDE retrieval for scanned document Q&A:
workflows:
- id: ocr-hyde-pipeline
trigger:
uri: http-server://0.0.0.0:8080/api/scan-and-ask
steps:
- rag-ingest:
parser: ocr
ocr:
backend: docling
endpoint: "http://localhost:5000"
- rag-chunk:
strategy: semantic
threshold: 0.5
- rag-embed:
provider: openai
model: text-embedding-3-small
- rag-store:
store: pgvector
collection: scanned
- rag-retrieve:
strategy: hyde
top_k: 10
hyde:
model:
provider: openai
model: gpt-4o
embedding:
provider: openai
model: text-embedding-3-small
- rag-postprocess:
steps:
- type: cohere-rerank
params:
api_key: "${env:COHERE_API_KEY}"
top_n: "5"
- rag-synthesize:
mode: refine
model:
provider: anthropic
model: claude-sonnet-4-20250514
Recipe 110: Streaming OCR for Large Scanned PDFs¶
workflows:
- id: large-pdf-ocr
trigger:
uri: http-server://0.0.0.0:8080/api/ingest-large
steps:
- rag-ingest:
parser: streaming-ocr
ocr:
backend: docling
endpoint: "http://localhost:5000"
streaming:
max_memory_bytes: 52428800 # 50MB threshold
page_split: true
concurrency: 4
max_file_size: 1073741824 # 1GB limit
- rag-chunk:
strategy: markdown
- rag-embed:
provider: openai
model: text-embedding-3-small
- rag-store:
store: pgvector
collection: large_scanned
Recipe 111: Parallel Page OCR with Progress Tracking¶
workflows:
- id: page-by-page-ocr
trigger:
uri: http-server://0.0.0.0:8080/api/ocr-pages
steps:
- rag-ingest:
parser: streaming-ocr
ocr:
backend: azure-doc-intelligence
endpoint: "${env:AZURE_ENDPOINT}"
api_key: "${env:AZURE_DOC_KEY}"
streaming:
page_split: true
concurrency: 8
- rag-chunk:
strategy: recursive
chunk_size: 1000
overlap: 100
- rag-embed:
provider: jina
model: jina-embeddings-v3
- rag-store:
store: milvus
collection: ocr_pages
Recipe 112: Streaming OCR with File Size Limit¶
workflows:
- id: size-limited-ocr
trigger:
uri: http-server://0.0.0.0:8080/api/safe-ocr
steps:
- rag-ingest:
parser: streaming-ocr
ocr:
backend: marker
endpoint: "http://localhost:8001"
streaming:
max_file_size: 104857600 # 100MB limit
max_memory_bytes: 10485760 # 10MB memory threshold
Helix v0.2.0 — 48 components (incl. MCP, A2A, AG-UI, A2UI), 444 connectors, 60+ orchestration patterns, 7 built-in agent types (+ RegisterAgentType for custom), 5 agent flows (+ parallel_error_policy fail_fast / collect), 8 memory backends + named middleware, 17 LLM providers (incl. Qwen with thinking + Qwen-Omni audio I/O), reasoning_effort / reasoning_content / FinishReason across providers, 10 embedding providers, 10 vector stores, 10 chunking strategies, 10+ data readers (incl. OCR), 13 retrieval strategies (incl. HyDE, MultiQuery, StepBack, SubQuestion), 16 postprocessors, 5 synthesis modes, full multimodal support, knowledge graph RAG, RAG evaluation, large file streaming, 8 OCR backends, streaming OCR for large files. Runtime governance: hash-chained audit, scoped kill switches, 4-tier privilege rings, Starlang-predicate policy engine, MCP security scanner, prompt-defense classifier.