Building a DAG Execution Pipeline with FFAI

In this tutorial you will build a multi-step research pipeline using FFAI’s DAG (directed acyclic graph) execution. By the end you will be able to define prompt dependency graphs, validate them, and execute prompts in parallel where possible.

Prerequisites

  • Python >= 3.10

  • pip install ffai

  • A Mistral API key (or any LiteLLM-supported provider) set as MISTRAL_API_KEY

  • Familiarity with the quickstart and basic FFAI usage

Step 1: Set up the async client

DAG execution requires an async client. The execute_graph method runs prompts in topological-parallel order using asyncio.gather:

import os

from ffai.Clients import AsyncFFLiteLLMClient
from ffai.FFAI import FFAI

client = AsyncFFLiteLLMClient(
    model_string="mistral/mistral-small-latest",
    api_key=os.environ["MISTRAL_API_KEY"],
)

ffai = FFAI(client)

Step 2: Define a prompt graph

Define prompts as a list of dicts. Each prompt has a prompt_name and a prompt. Use the history key to declare dependencies — FFAI will ensure that dependent prompts run after their prerequisites:

prompts = [
    {
        "prompt_name": "topic",
        "prompt": "Suggest a topic for a blog post about technology",
    },
    {
        "prompt_name": "outline",
        "prompt": "Create a detailed outline for a blog post about {{topic.response}}",
        "history": ["topic"],
    },
    {
        "prompt_name": "article",
        "prompt": "Write a 3-paragraph blog post based on this outline:\n{{outline.response}}",
        "history": ["outline"],
    },
]

The {{topic.response}} and {{outline.response}} placeholders are resolved at execution time — each prompt receives the output of the prompts it depends on.

Step 3: Validate the graph

Before executing, validate the graph to check for cycles, missing dependencies, and other issues:

graph, warnings = ffai.validate_graph(prompts)

print(f"Graph has {len(graph.nodes)} nodes, {len(graph.edges)} edges")
print(f"Max level: {graph.max_level}")

if warnings:
    for w in warnings:
        print(f"Warning: {w}")

Output (deterministic):

Graph has 3 nodes, 2 edges
Max level: 2

The max_level indicates the depth of the dependency chain (0 = root, 1 = depends on root, etc.). Prompts at the same level run in parallel.

Step 4: Execute the graph

Execute the full pipeline. execute_graph requires an AsyncFFLiteLLMClient and runs prompts level by level:

graph_result = await ffai.execute_graph(prompts, max_concurrency=10)

for name, r in graph_result.results.items():
    print(f"{name}: {r.status} ({r.duration_ms:.0f}ms)")

Output (illustrative):

topic: success (842ms)
outline: success (1203ms)
article: success (2156ms)

execute_graph returns a GraphResult with:

  • results (dict[str, ResponseResult]) — one result per prompt

  • success_count (int) — how many succeeded

  • failed_count (int) — how many failed

  • skipped_count (int) — how many were skipped by conditions

  • aborted (bool) — whether an abort condition was triggered

Step 5: Add parallel branches

When prompts at the same level have no dependencies on each other, they run in parallel. Here, angles and audience both depend on topic but not on each other:

prompts = [
    {
        "prompt_name": "topic",
        "prompt": "Suggest a topic for a blog post about technology",
    },
    {
        "prompt_name": "angles",
        "prompt": "List 3 unique angles on {{topic.response}}",
        "history": ["topic"],
    },
    {
        "prompt_name": "audience",
        "prompt": "Describe the target audience for {{topic.response}}",
        "history": ["topic"],
    },
    {
        "prompt_name": "article",
        "prompt": (
            "Write a blog post about {{topic.response}}\n"
            "Angles: {{angles.response}}\n"
            "Audience: {{audience.response}}"
        ),
        "history": ["angles", "audience"],
    },
]

graph, _ = ffai.validate_graph(prompts)
print(f"Nodes: {len(graph.nodes)}, Edges: {len(graph.edges)}, Levels: {graph.max_level}")

Output (deterministic):

Nodes: 4, Edges: 4, Levels: 2

angles and audience run in parallel at level 1. article waits for both at level 2.

Complete listing

import os

from ffai.Clients import AsyncFFLiteLLMClient
from ffai.FFAI import FFAI

client = AsyncFFLiteLLMClient(
    model_string="mistral/mistral-small-latest",
    api_key=os.environ["MISTRAL_API_KEY"],
)
ffai = FFAI(client)

# Define the prompt graph
prompts = [
    {
        "prompt_name": "topic",
        "prompt": "Suggest a topic for a blog post about technology",
    },
    {
        "prompt_name": "outline",
        "prompt": "Create a detailed outline for a blog post about {{topic.response}}",
        "history": ["topic"],
    },
    {
        "prompt_name": "article",
        "prompt": "Write a 3-paragraph blog post based on this outline:\n{{outline.response}}",
        "history": ["outline"],
    },
]

# Validate
graph, warnings = ffai.validate_graph(prompts)
print(f"Graph: {len(graph.nodes)} nodes, {len(graph.edges)} edges")
if warnings:
    for w in warnings:
        print(f"Warning: {w}")

# Execute
graph_result = await ffai.execute_graph(prompts, max_concurrency=10)
for name, r in graph_result.results.items():
    print(f"{name}: {r.status} ({r.duration_ms:.0f}ms)")

print(f"\nSuccess: {graph_result.success_count}")
print(f"Failed: {graph_result.failed_count}")

Next steps