embedder

The embedder.example microservice is the worked example for the framework’s Python integration via the pyvenv library. It loads a real sentence-transformers model into an in-process Python virtual environment and exposes two typed Go endpoints — Embed for a vector, Similarity for the cosine score between two strings — that delegate into the venv.

It demonstrates the full Go-Python bridge end-to-end:

A venv that the microservice owns directly (no separate Python service to deploy, no bus round-trip per call).
The deferred-subscription lifecycle: the microservice goes live on the bus immediately while the venv provisions in the background; Python-backed endpoints only come onto the bus when the venv reports ready.
A demo HTML page that drives venv start from a button click and surfaces the multi-second pip install + model download via tailed stdout/stderr.

The example is intentionally not bundled in the fabric repo’s main/main.go. The guided tour adds it explicitly so dev environments without Python remain unaffected. Add embedder.NewService() to your app.Add(...) block to bring it online.

Depends On

A Python 3 toolchain on the host (python3 -m venv and pip must work).
The github.com/microbus-io/pyvenv library — already a transitive dependency of the fabric module.
Network access on first start, to install sentence-transformers from PyPI and download the MiniLM model (~80 MB). Subsequent starts reuse the on-disk venv and skip both steps.

Try It

With the examples app running and embedder.NewService() added:

http://localhost:8080/embedder.example/demo

You’ll see a setup card with an “Initialize Python VM” button.

Click it. The button is wired to POST /demo/init, which kicks off venv.Start(ctx) on a background goroutine and returns immediately.
The page long-polls GET /demo/status and renders the tailed pip-install output as it arrives. First run takes 30 s – a few minutes depending on bandwidth; subsequent runs are seconds because the venv is cached on disk.
When the venv reports ready, the page swaps the setup card for the action cards. /embed returns the 384-dimensional vector for a string; /similarity returns the cosine score between two strings.

Behind the scenes:

GET /embedder.example/embed?text=Hello%20world → { "vector": [0.012, -0.034, ...] }
GET /embedder.example/similarity?a=Hello&b=Hi → { "score": 0.78 }

Code Walkthrough

Source: exampleservices/embedder/service.go.

The Service struct holds the venv handle and a small initialization-status block that backs the demo’s long-poll endpoint:

type Service struct {
    *Intermediate // IMPORTANT: Do not remove

    venv *pyvenv.Venv

    initMu     sync.Mutex
    initStatus string // "not_started" | "pending" | "ready" | "error"
    initError  string
}

venv is the long-lived Python subprocess handle. initStatus + initError are the demo’s view of the venv’s startup lifecycle; the production-style guard against using the venv before it’s ready lives on svc.venv.Ready() in the handler bodies.

Wiring (intermediate.go)

Embed and Similarity are wired with two unusual subscription options:

svc.Subscribe(
    "Embed", svc.doEmbed,
    sub.At(embedderapi.Embed.Method, embedderapi.Embed.Route),
    sub.Description(`Embed returns the sentence-embedding vector for the input text.`),
    sub.Function(embedderapi.EmbedIn{}, embedderapi.EmbedOut{}),
    sub.Manual(), sub.Tag("python"),
)
svc.Subscribe(
    "Similarity", svc.doSimilarity,
    sub.At(embedderapi.Similarity.Method, embedderapi.Similarity.Route),
    sub.Description(`Similarity returns the cosine similarity between the embeddings of strings a and b.`),
    sub.Function(embedderapi.SimilarityIn{}, embedderapi.SimilarityOut{}),
    sub.Manual(), sub.Tag("python"),
)

sub.Manual() tells the framework not to bring the subscription onto the bus at startup. sub.Tag("python") groups both subscriptions under a name the microservice can activate as a batch. The onVenvLiveness callback below activates every python-tagged subscription when the venv reaches StateReady, and deactivates them on StateDied. This is the deferred-subscription pattern: while the venv is provisioning (or has crashed), calls to /embed and /similarity get a clean 404 ack-timeout from the bus rather than a 503 or a slow hang.

The three demo endpoints are subscribed normally with sub.Web() — they have nothing to do with the venv being up, so they should always be reachable.

A single config rounds out the wiring:

svc.DefineConfig(
    "MaxWorkers",
    cfg.Description(`MaxWorkers caps how many calls into the Python venv may run concurrently.`),
    cfg.DefaultValue("2"),
    cfg.Validation("int [1,]"),
)

MaxWorkers sizes the Python ThreadPoolExecutor that dispatches concurrent Calls. For GIL-releasing work (NumPy, PyTorch, pandas, I/O) the threads run truly in parallel; for GIL-bound code, the right scaling knob is replica count instead.

OnStartup

func (svc *Service) OnStartup(ctx context.Context) (err error) {
    sources, err := readPythonSources()
    if err != nil {
        return errors.Trace(err)
    }
    svc.venv, err = pyvenv.New(pyvenv.Config{
        Sources:          sources,
        Requirements:     parseRequirements(pythonRequirements),
        MaxWorkers:       svc.MaxWorkers(),
        Logger:           svc,
        LivenessCallback: svc.onVenvLiveness,
    })
    if err != nil {
        return errors.Trace(err)
    }
    svc.setInitStatus("not_started", "")
    return nil
}

OnStartup reads the embedded *.py sources and requirements.txt, constructs the *pyvenv.Venv with MaxWorkers + a LivenessCallback, and returns. No subprocess is spawned yet. That deferral is deliberate — pip install sentence-transformers plus model download can take minutes, and surfacing that cost in the demo UI (rather than blocking startup) is the point of the example. The microservice goes live on the bus immediately for control-plane traffic; the venv comes up later.

In a production Python-backed service, you’d typically launch venv.Start(ctx) from OnStartup itself via svc.Go(...) so the venv comes up automatically. The demo’s button-driven start is example-specific scaffolding.

onVenvLiveness

func (svc *Service) onVenvLiveness(state pyvenv.State, err error) {
    ctx := svc.Lifetime()
    switch state {
    case pyvenv.StateReady:
        actErr := svc.activatePythonSubs()
        if actErr != nil {
            svc.LogError(ctx, "Activating python subs", "error", actErr)
        }
        svc.setInitStatus("ready", "")
    case pyvenv.StateDied:
        svc.LogWarn(ctx, "Python venv died", "error", err)
        dErr := svc.deactivatePythonSubs()
        if dErr != nil {
            svc.LogError(ctx, "Deactivating python subs", "error", dErr)
        }
        msg := "python subprocess exited"
        if err != nil {
            msg = err.Error()
        }
        svc.setInitStatus("error", msg)
    }
}

The liveness callback is the heart of the deferred-subscription pattern. pyvenv fires StateReady from its own goroutine when the worker is up and defined; the handler responds by activating every python-tagged subscription, which finally puts Embed and Similarity on the bus. StateDied (worker crash, OOM, etc.) is the inverse: deactivate the subs immediately so the bus stops routing calls to a replica whose Python is gone. Recovery is fast — the on-disk venv is reused and pip install is skipped, so only the worker spawn and define re-run.

Start failures (pip install fails, model import fails, ctx expires before ready) don’t go through StateDied per pyvenv’s contract — they’re returned from Start and handled by the DemoInit goroutine that called it.

Embed and Similarity

func (svc *Service) Embed(ctx context.Context, text string) (vector []float64, err error) {
    if svc.venv == nil || !svc.venv.Ready() {
        return nil, errors.New("venv not ready", http.StatusServiceUnavailable)
    }
    in := embedderapi.EmbedIn{
        Text: text,
    }
    var out embedderapi.EmbedOut
    err = svc.venv.CallAndAwait(ctx, "embed", in, &out)
    if err != nil {
        return nil, errors.Trace(err)
    }
    return out.Vector, nil
}

func (svc *Service) Similarity(ctx context.Context, a string, b string) (score float64, err error) {
    if svc.venv == nil || !svc.venv.Ready() {
        return 0, errors.New("venv not ready", http.StatusServiceUnavailable)
    }
    in := embedderapi.SimilarityIn{
        A: a,
        B: b,
    }
    var out embedderapi.SimilarityOut
    err = svc.venv.CallAndAwait(ctx, "similarity", in, &out)
    if err != nil {
        return 0, errors.Trace(err)
    }
    return out.Score, nil
}

Each handler is shaped the same way: pack a typed *api.<Name>In struct, call svc.venv.CallAndAwait(ctx, "<funcName>", in, &out), and return the typed result. pyvenv serializes in to JSON, dispatches to the Python <funcName> function over the worker’s local pipe, and json.Unmarshals the response straight into out.

The !svc.venv.Ready() guard is defense-in-depth. With sub.Manual() + sub.Tag("python"), the subscription is normally off the bus until the liveness callback brings it online, so callers see a clean 404 ack-timeout rather than a 503 during provisioning. The 503 guard only fires if a caller somehow reaches the handler while the venv is starting up or recovering — for example, in tests that bypass the subscription layer.

Python Side

The Python functions called from Embed and Similarity live in a sibling service.py (embedded via //go:embed *.py). Python sees typed JSON in and returns typed JSON out — it has no awareness of ctx, tracing, claims, or the bus:

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("all-MiniLM-L6-v2")

def embed(args):
    vector = model.encode(args["text"]).tolist()
    return {"vector": vector}

def similarity(args):
    a_vec = model.encode(args["a"])
    b_vec = model.encode(args["b"])
    score = float(util.cos_sim(a_vec, b_vec).item())
    return {"score": score}

The keys in the returned dicts (vector, score) line up with the json:"vector" / json:"score" tags on EmbedOut / SimilarityOut. No mapping layer in between.

Demo, DemoInit, DemoStatus

The three web endpoints back the interactive demo page. Demo serves a single self-contained HTML template; DemoInit and DemoStatus drive the venv lifecycle from the browser.

func (svc *Service) DemoInit(w http.ResponseWriter, r *http.Request) (err error) {
    svc.initMu.Lock()
    switch svc.initStatus {
    case "pending", "ready":
        status := svc.initStatus
        svc.initMu.Unlock()
        return writeJSON(w, map[string]any{"status": status})
    }
    svc.initStatus = "pending"
    svc.initError = ""
    svc.initMu.Unlock()

    svc.Go(r.Context(), func(ctx context.Context) error {
        err := svc.StartPyVenv(ctx)
        if err != nil {
            svc.LogError(ctx, "Python venv Start failed", "error", err)
            svc.setInitStatus("error", err.Error())
            return err
        }
        // Success: the LivenessCallback flips status to "ready" and activates subs.
        return nil
    })
    return writeJSON(w, map[string]any{"status": "pending"})
}

DemoInit is idempotent — clicking the button again while a start is in progress (or already ready) returns the current status without restarting. The actual start runs on svc.Go(...), a framework-tracked goroutine that the connector drains on shutdown. The endpoint itself returns immediately with status: "pending"; the browser then long-polls DemoStatus to watch progress.

DemoStatus is an ETag-driven long-poll endpoint that holds the connection open until the status changes (or returns 204 No Content close to the deadline so the client can reconnect). Each response includes the current status plus the most recent stdout+stderr from the venv via svc.venv.TailStdOut() / TailStdErr(). This is what makes the pip install and model download visible to the user as it happens, rather than appearing as a frozen spinner.

OnShutdown

func (svc *Service) OnShutdown(ctx context.Context) (err error) {
    if svc.venv != nil {
        err := svc.venv.Close(ctx)
        if err != nil {
            svc.LogError(ctx, "Closing python venv failed", "error", err)
        }
    }
    return nil
}

venv.Close kills the subprocess, drains any pending calls with ErrClosed, and removes the on-disk venv directory. The kernel cleans up the child process if the parent crashes, so there’s no orphan-subprocess story to manage.