Example 02: Recipe Analyzer

The workflow

Pipeline structure

AI extracts signals from unstructured text; deterministic ops turn those signals into a score; the score gates one of three advice lanes.

HTTPGetOp

fetch / fixture

→

JSONExtractOp

instructions

→

AIExtract (ingredients)

AIExtract (steps)

AIParseNumber (minutes)

→

MulFloatOp / AddFloatOp

score formula

→

easy advice

medium advice

hard advice

→

CoalesceN

det op

Deterministic steps

ContextValOp / HTTPGetOp

Live mode: fetches the TheMealDB API via HTTPGetOp. Fixture mode: reads a pre-captured JSON string from context via ContextValFactory[string]. Same downstream graph either way.

JSONExtractOp (×2)

Pulls meals.0.strInstructions and meals.0.strMeal from the API JSON using a dot-path selector. Zero AI tokens.

SliceLenOp (×2)

Counts the ingredients and steps lists that AI returned. Converts a non-deterministic list into a deterministic integer for the scoring formula.

IntToFloat64Op (×2) + MulFloatOp (×2) + AddFloatOp (×2)

Implements score = ingredient_count + step_count×1.5 + cook_minutes×0.1. Each op is a single typed struct wired together — no scripting or eval.

Predicate gates (×3) + CoalesceNStringOp

score_is_easy, score_is_medium, score_is_hard compare the computed float against thresholds (15 and 30). Exactly one advice vertex fires.

AI op

AI steps

AIExtractStringSliceOp — ingredients

Lists every distinct ingredient from the instructions text, one per item, stripped of quantities and units. Returns []string.

AIExtractStringSliceOp — steps

Lists every discrete cooking step as a flat list. Runs concurrently with the ingredients extractor — same input wire, no dependency between them.

AIParseNumberOp — cook_minutes

Estimates total active + passive cook time in minutes. Also runs concurrently. Handles vague phrasings like "bake for about 30–40 minutes" reliably.

AIComputeStringToStringOp — one advice lane

Whichever difficulty band the score falls in, its lane generates a single targeted cooking tip. Only one of the three lanes fires per run.

Implementation

Walkthrough

1. Three parallel AI extractors over the same input

All three extractors read from the same instructions_text wire. Because they have no dependencies on each other, the dagor engine schedules all three concurrently. The three AI calls complete in roughly the time of one.

          
          
          
        
main.go — Stage 3 (three parallel extractors)
// All three share the same "instructions_text" input wire.
// No mutual dependency → the engine runs them concurrently.
Vertex("ingredients").Op("AIExtractStringSliceOp").
    Params(map[string]string{"operation": "extract every distinct ingredient ..."}).
    Input("Input", "instructions_text").
    Output("Result", "ingredients").

Vertex("steps").Op("AIExtractStringSliceOp").
    Params(map[string]string{"operation": "extract every discrete cooking step ..."}).
    Input("Input", "instructions_text").
    Output("Result", "steps").

Vertex("cook_minutes").Op("AIParseNumberOp").
    Params(map[string]string{"operation": "estimate total cooking time in minutes"}).
    Input("Input", "instructions_text").
    Output("Result", "cook_minutes").
main.py — Stage 3 (three parallel extractors)
# All three share the same "instructions_text" input wire.
# No mutual dependency → the engine runs them concurrently.
b.vertex("ingredients").op("AIExtractStringSliceOp") \
    .params({"operation": "extract every distinct ingredient ..."}) \
    .input("Input", "instructions_text") \
    .output("Result", "ingredients")

b.vertex("steps").op("AIExtractStringSliceOp") \
    .params({"operation": "extract every discrete cooking step ..."}) \
    .input("Input", "instructions_text") \
    .output("Result", "steps")

b.vertex("cook_minutes").op("AIParseNumberOp") \
    .params({"operation": "estimate total cooking time in minutes"}) \
    .input("Input", "instructions_text") \
    .output("Result", "cook_minutes")
index.ts — Stage 3 (three parallel extractors)
// All three share the same "instructions_text" input wire.
// No mutual dependency → the engine runs them concurrently.
const ingredients = wf.ai.extractStringSlice({
    operation: "extract every distinct ingredient ...",
    input: instructionsText,
});

const steps = wf.ai.extractStringSlice({
    operation: "extract every discrete cooking step ...",
    input: instructionsText,
});

const cookMinutes = wf.ai.parseNumber({
    operation: "estimate total cooking time in minutes",
    input: instructionsText,
});

2. Deterministic scoring formula

Once the AI extractions complete, the rest of the scoring is pure arithmetic. SliceLenOp counts the list items, IntToFloat64Op widens to float64, and MulFloatOp/AddFloatOp implement the formula. None of these steps ever calls an LLM.

          
          
          
        
main.go — Stage 4 (deterministic scoring)
Vertex("ingredient_count_int").Op("SliceLenOp").
    Input("Input", "ingredients").
    Output("Result", "ingredient_count_int").

Vertex("ingredient_count").Op("IntToFloat64Op").
    Input("Value", "ingredient_count_int").
    Output("Result", "ingredient_count_f").

Vertex("step_weight").Op("ConstFloat64Op").
    Params(map[string]string{"Value": "1.5"}).
    Output("Result", "step_weight").

Vertex("step_term").Op("MulFloatOp").
    Input("A", "step_count_f").
    Input("B", "step_weight").
    Output("Result", "step_term").

Vertex("difficulty_score").Op("AddFloatOp").
    Input("A", "partial_score").
    Input("B", "cook_term").
    Output("Result", "difficulty_score").
main.py — Stage 4 (deterministic scoring)
b.vertex("ingredient_count_int").op("SliceLenOp") \
    .input("Input", "ingredients") \
    .output("Result", "ingredient_count_int")

b.vertex("ingredient_count").op("IntToFloat64Op") \
    .input("Value", "ingredient_count_int") \
    .output("Result", "ingredient_count_f")

b.vertex("step_weight").op("ConstFloat64Op") \
    .params({"Value": "1.5"}) \
    .output("Result", "step_weight")

b.vertex("step_term").op("MulFloatOp") \
    .input("A", "step_count_f") \
    .input("B", "step_weight") \
    .output("Result", "step_term")

b.vertex("difficulty_score").op("AddFloatOp") \
    .input("A", "partial_score") \
    .input("B", "cook_term") \
    .output("Result", "difficulty_score")
index.ts — Stage 4 (deterministic scoring)
const ingredientCountInt = wf.op.sliceLen({
    input: ingredients,
});

const ingredientCount = wf.op.intToFloat64({
    value: ingredientCountInt,
});

const stepWeight = wf.op.constFloat64({
    value: 1.5,
});

const stepTerm = wf.op.mulFloat({
    a: stepCountF,
    b: stepWeight,
});

const difficultyScore = wf.op.addFloat({
    a: partialScore,
    b: cookTerm,
});

3. Numeric predicate gates on the score

Predicates for continuous scores compare the float wire against thresholds. Here the advice op itself carries the condition rather than a separate gate op — the predicate is evaluated before the vertex runs, and if false, the vertex is skipped along with its output wire.

          
          
          
        
main.go — predicates + lane vertices
// Thresholds: easy < 15, medium 15–30, hard ≥ 30.
predicate.Register("score_is_easy", func(in map[string]any) bool {
    s, ok := in["difficulty_score"].(*float64)
    return ok && s != nil && *s < 15.0
})

// The condition goes on the advice vertex itself — no separate gate.
b.Vertex("easy_advice").Op("AIComputeStringToStringOp").
    Condition("score_is_easy").
    ConditionInput("difficulty_score").
    Params(map[string]string{"operation": "write a one-sentence tip for a beginner cook ..."}).
    Input("Input", "meal_name").
    Output("Result", "easy_advice").
main.py — predicates + lane vertices
# Thresholds: easy < 15, medium 15–30, hard ≥ 30.
def score_is_easy(inputs: dict[str, Any]) -> bool:
    s = inputs.get("difficulty_score")
    return s is not None and s < 15.0
predicate.register("score_is_easy", score_is_easy)

# The condition goes on the advice vertex itself — no separate gate.
b.vertex("easy_advice").op("AIComputeStringToStringOp") \
    .condition("score_is_easy") \
    .condition_input("difficulty_score") \
    .params({"operation": "write a one-sentence tip for a beginner cook ..."}) \
    .input("Input", "meal_name") \
    .output("Result", "easy_advice")
index.ts — predicates + lane vertices
// Thresholds: easy < 15, medium 15–30, hard ≥ 30.
predicate.register("score_is_easy", (inputs: Record<string, any>) => {
    const s = inputs["difficulty_score"];
    return typeof s === "number" && s < 15.0;
});

// The condition goes on the advice vertex itself — no separate gate.
const easyAdvice = wf.ai.computeStringToString({
    operation: "write a one-sentence tip for a beginner cook ...",
    input: mealName,
}, {
    condition: "score_is_easy",
    conditionInput: difficultyScore,
});

Run it

Test with three fixtures spanning all three difficulty bands: Pancakes (easy/medium boundary), Chicken Curry (medium), Beef Wellington (hard).

          
          
          
        
shell
export CLAUDE_API_KEY=<your key>

# Live API
go run ./examples/recipe-analyzer --meal "Pancakes"
go run ./examples/recipe-analyzer --meal "Chicken Curry"
go run ./examples/recipe-analyzer --meal "Beef Wellington"

# Offline fixtures
go run ./examples/recipe-analyzer --fixture examples/recipe-analyzer/testdata/recipes/pancakes.json
shell
export CLAUDE_API_KEY=<your key>

# Live API
python examples/recipe_analyzer/main.py --meal "Pancakes"
python examples/recipe_analyzer/main.py --meal "Chicken Curry"
python examples/recipe_analyzer/main.py --meal "Beef Wellington"

# Offline fixtures
python examples/recipe_analyzer/main.py --fixture examples/recipe_analyzer/testdata/recipes/pancakes.json
shell
export CLAUDE_API_KEY=<your key>

# Live API
npx tsx examples/recipe_analyzer/index.ts --meal "Pancakes"
npx tsx examples/recipe_analyzer/index.ts --meal "Chicken Curry"
npx tsx examples/recipe_analyzer/index.ts --meal "Beef Wellington"

# Offline fixtures
npx tsx examples/recipe_analyzer/index.ts --fixture examples/recipe_analyzer/testdata/recipes/pancakes.json

Run it as an MCP server

The same binary is dual-mode. Pass --mcp and instead of running once and exiting it speaks the Model Context Protocol over stdin/stdout, exposing the entire workflow as a single MCP tool, analyze_recipe. The Go SDK derives the tool's input schema from the workflow's UserInput struct and validates every tools/call request against it, so an agent runs the whole lookup → extract → score → advice-lane pipeline with one tool call.

          
          
          
        
shell
# Speak MCP over stdin/stdout instead of running once
go run ./examples/recipe-analyzer --mcp
shell
# Speak MCP over stdin/stdout instead of running once
python examples/recipe_analyzer/main.py --mcp
shell
# Speak MCP over stdin/stdout instead of running once
npx tsx examples/recipe_analyzer/index.ts --mcp

Register it with any MCP client by pointing the client at that command:

          
          
          
        
.mcp.json
{
  "mcpServers": {
    "recipe-analyzer": {
      "command": "go",
      "args": ["run", "./examples/recipe-analyzer", "--mcp"]
    }
  }
}
.mcp.json
{
  "mcpServers": {
    "recipe-analyzer": {
      "command": "python",
      "args": ["examples/recipe_analyzer/main.py", "--mcp"]
    }
  }
}
.mcp.json
{
  "mcpServers": {
    "recipe-analyzer": {
      "command": "npx",
      "args": ["tsx", "examples/recipe_analyzer/index.ts", "--mcp"]
    }
  }
}

Token usage

Measured on testdata/recipes/pancakes.json. All four AIComputeOp calls are the three parallel extractors plus the matched difficulty lane's description writer.

Op	Calls	Input tokens	Output tokens	Total tokens
AIComputeOp	4	800	177	977
Total	4	800	177	977

← Example 01 Next: README Quality →

Recipe Difficulty Analyzer