Code-first · What a workflow looks like
If you can read code, you can read your workflow.
A workflow is a graph of task endpoints. Each task is a normal function with typed
inputs and typed outputs. Microbus carries shared state through the graph, fans out and
fans back in, and reduces overlapping fields deterministically.
- Tasks are addressable on the bus, like any other endpoint
- State is JSON. Reducers are conventional and predictable
- Foreman orchestrates the graph; you author the steps
- Every step is persisted; a crashed flow resumes right where it left off
// ResearchArticle defines the workflow graph that fetches an article, summarizes it,
// critiques the summary, and translates the critique into French and Spanish in parallel.
func (svc *Service) ResearchArticle(ctx context.Context) (graph *workflow.Graph, err error) {
graph = workflow.NewGraph("ResearchArticle")
graph.SetEndpoint("FetchArticle", researchflowapi.FetchArticle.URL())
graph.SetEndpoint("Summarize", researchflowapi.Summarize.URL())
graph.SetEndpoint("Critique", researchflowapi.Critique.URL())
graph.SetEndpoint("TranslateFrench", researchflowapi.TranslateFrench.URL())
graph.SetEndpoint("TranslateSpanish", researchflowapi.TranslateSpanish.URL())
graph.SetEndpoint("Finalize", researchflowapi.Finalize.URL())
graph.SetFanIn("Finalize")
graph.AddTransitionChain("FetchArticle", "Summarize", "Critique")
graph.AddTransitionChain("Critique", "TranslateFrench", "Finalize")
graph.AddTransitionChain("Critique", "TranslateSpanish", "Finalize")
graph.AddTransitionChain("Finalize", workflow.END)
return graph, nil
}
// FetchArticle retrieves the article body for the given URL via the HTTP egress proxy.
func (svc *Service) FetchArticle(ctx context.Context, flow *workflow.Flow, articleURL string) (articleText string, err error) {
if articleURL == "" {
return "", errors.New("articleURL is empty", http.StatusBadRequest)
}
resp, err := httpegressapi.NewClient(svc).Get(ctx, articleURL)
if err != nil {
return "", errors.Trace(err)
}
defer resp.Body.Close()
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return "", errors.New("fetch failed", "url", articleURL, "status", resp.StatusCode, resp.StatusCode)
}
body, err := io.ReadAll(resp.Body)
if err != nil {
return "", errors.Trace(err)
}
return string(body), nil
}
// Summarize produces a concise summary of the article text via the LLM service.
func (svc *Service) Summarize(ctx context.Context, flow *workflow.Flow, articleText string) (summary string, err error) {
if articleText == "" {
return "", errors.New("articleText is empty", http.StatusBadRequest)
}
summary, yield, err := svc.chat(ctx, flow,
"You are a precise summarizer. Reply with a single short paragraph that captures the main idea of the article. Do not preface your answer.",
"Summarize the following article:\n\n"+articleText,
)
if yield {
return "", nil // parked; re-enters when the subflow ends
}
return summary, err
}
// Critique evaluates the summary for accuracy, clarity, and completeness via the LLM service.
func (svc *Service) Critique(ctx context.Context, flow *workflow.Flow, summary string) (critique string, err error) {
if summary == "" {
return "", errors.New("summary is empty", http.StatusBadRequest)
}
critique, yield, err := svc.chat(ctx, flow,
"You are an editor. Reply with a short, candid critique of the summary, focused on clarity, completeness, and tone. Do not preface your answer.",
"Critique the following summary:\n\n"+summary,
)
if yield {
return "", nil // parked; re-enters when the subflow ends
}
return critique, err
}
// TranslateFrench returns a French translation of the critique via the LLM service.
func (svc *Service) TranslateFrench(ctx context.Context, flow *workflow.Flow, critique string) (translationFr string, err error) {
if critique == "" {
return "", errors.New("critique is empty", http.StatusBadRequest)
}
translationFr, yield, err := svc.chat(ctx, flow,
"You translate English into French. Reply with the French translation only, no commentary.",
critique,
)
if yield {
return "", nil // parked; re-enters when the subflow ends
}
return translationFr, err
}
// TranslateSpanish returns a Spanish translation of the critique via the LLM service.
func (svc *Service) TranslateSpanish(ctx context.Context, flow *workflow.Flow, critique string) (translationEs string, err error) {
if critique == "" {
return "", errors.New("critique is empty", http.StatusBadRequest)
}
translationEs, yield, err := svc.chat(ctx, flow,
"You translate English into Spanish. Reply with the Spanish translation only, no commentary.",
critique,
)
if yield {
return "", nil // parked; re-enters when the subflow ends
}
return translationEs, err
}
// Finalize is the fan-in nexus for the parallel translation branches. It reads both
// translations and reports which languages produced output. Having a single converging
// node lets the workflow's lineage validator close the fan-out frame opened at Critique.
func (svc *Service) Finalize(ctx context.Context, flow *workflow.Flow, translationFr string, translationEs string) (languages []string, err error) {
languages = []string{}
if translationFr != "" {
languages = append(languages, "fr")
}
if translationEs != "" {
languages = append(languages, "es")
}
return languages, nil
}
// chat runs a durable system+user exchange against the configured LLM provider
// as a child subgraph and returns the assistant's reply.
func (svc *Service) chat(ctx context.Context, flow *workflow.Flow, systemPrompt string, userPrompt string) (reply string, yield bool, err error) {
items := []llmapi.Item{
llmapi.NewMessage("system", systemPrompt).AsItem(),
llmapi.NewMessage("user", userPrompt).AsItem(),
}
itemsOut, _, yield, err := llmapi.NewSubgraph(flow).ChatLoop(ctx, svc.Provider(), svc.Model(), items, nil, nil)
if err != nil {
return "", false, errors.Trace(err)
}
if yield {
return "", true, nil // parked; re-enters when ChatLoop ends
}
return llmapi.LastAssistantMessage(itemsOut), false, nil
}
