Workflow Quickstart
Create multi-step workflows with YAML or Python.
Quick Start
# workflows/my_workflow.yaml
workflows:
my_task:
nodes:
- id: step1
type: agent
role: researcher
goal: "Research the topic"
next: [step2]
- id: step2
type: compute
handler: summarize
next: []
victor workflow render workflows/my_workflow.yaml
victor workflow execute my_workflow
Node Types
| Type |
Purpose |
LLM |
Example |
agent |
LLM-powered tasks |
✅ |
Research, analysis |
compute |
Function calls |
❌ |
Statistics, transforms |
condition |
Branching logic |
❌ |
Quality checks |
parallel |
Concurrent execution |
✅ |
Multi-source |
hitl |
Human approval |
✅ |
Code reviews |
Workflow Structure
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Start │────▶│ Node 1 │────▶│ Node 2 │
└─────────────┘ └─────────────┘ └─────────────┘
│
┌───────┴───────┐
▼ ▼
┌───────────┐ ┌───────────┐
│ Condition │ │ Parallel │
└───────────┘ └───────────┘
│ │
▼ ▼
┌────────────────────────┐
│ End Node │
└────────────────────────┘
Node Fields
| Field |
Required |
Description |
id |
✅ |
Unique identifier |
type |
✅ |
Node type |
role |
agent |
Agent persona |
goal |
agent |
Task description |
handler |
compute |
Function name |
condition |
condition |
Branch function |
next |
✅ |
Next node IDs |
Example Workflows
Code Review
workflows:
code_review:
nodes:
- id: analyze
type: agent
role: reviewer
goal: "Review the code for issues"
next: [check]
- id: check
type: condition
condition: has_issues
branches:
"yes": fix
"no": done
- id: fix
type: agent
role: developer
goal: "Fix identified issues"
next: [done]
- id: done
type: agent
role: summarizer
goal: "Summarize the review"
next: []
Parallel Research
workflows:
deep_research:
nodes:
- id: parallel_search
type: parallel
nodes:
- id: web
type: agent
goal: "Search the web"
- id: docs
type: agent
goal: "Search documentation"
- id: code
type: agent
goal: "Search codebase"
join_strategy: merge_all
next: [synthesize]
- id: synthesize
type: agent
goal: "Combine all findings"
next: []
Escape Hatches
Python functions for complex logic:
# victor/myvertical/escape_hatches.py
def has_issues(ctx: dict) -> str:
"""Check if code has issues."""
issues = ctx.get("issues", [])
return "yes" if issues else "no"
def merge_results(ctx: dict) -> dict:
"""Merge parallel results."""
return {"merged": list(ctx.values())}
CONDITIONS = {"has_issues": has_issues}
TRANSFORMS = {"merge_results": merge_results}
Scheduling
# Add scheduled workflow
victor scheduler add my_workflow --cron "0 9 * * *"
# List scheduled workflows
victor scheduler list
# Remove schedule
victor scheduler remove <id>
Best Practices
| Practice |
Description |
| Start simple |
Begin with linear workflows |
| Use conditions |
Branch on quality/confidence |
| Parallelize |
Run independent steps together |
| Human in loop |
Add HITL for critical decisions |
| Escape hatches |
Python for complex logic only |
See Also