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
// Research defines the workflow graph: fetch -> summarize -> critique
// -> (translate-french || translate-spanish) -> end.
func (svc *Service) Research(ctx context.Context) (graph *workflow.Graph, err error) {
fetchArticle := researchflowapi.FetchArticle.URL()
summarize := researchflowapi.Summarize.URL()
critique := researchflowapi.Critique.URL()
translateFrench := researchflowapi.TranslateFrench.URL()
translateSpanish := researchflowapi.TranslateSpanish.URL()
graph = workflow.NewGraph(researchflowapi.Research.URL())
graph.DeclareInputs("articleURL")
graph.DeclareOutputs("summary", "critique", "french", "spanish")
graph.AddTransition(fetchArticle, summarize)
graph.AddTransition(summarize, critique)
graph.AddTransition(critique, translateFrench)
graph.AddTransition(critique, translateSpanish)
graph.AddTransition(translateFrench, workflow.END)
graph.AddTransition(translateSpanish, workflow.END)
return graph, nil
}
// FetchArticle retrieves the article body 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 required", http.StatusBadRequest)
}
resp, err := httpegressapi.NewClient(svc).Get(ctx, articleURL)
if err != nil {
return "", errors.Trace(err)
}
body, err := io.ReadAll(resp.Body)
if err != nil {
return "", errors.Trace(err)
}
return string(body), nil
}
// Summarize asks the LLM for a concise summary of the article.
func (svc *Service) Summarize(ctx context.Context, flow *workflow.Flow, articleText string) (summary string, err error) {
prompt := "Summarize the following article in 3-5 sentences. " +
"Respond with the summary only.\n\n" + articleText
return svc.askLLM(ctx, prompt)
}
// Critique asks the LLM to critique the summary for accuracy and gaps.
func (svc *Service) Critique(ctx context.Context, flow *workflow.Flow, summary string) (critique string, err error) {
prompt := "Critique the following summary for accuracy, clarity, " +
"and missing context. Respond with the critique only.\n\n" + summary
return svc.askLLM(ctx, prompt)
}
// TranslateFrench asks the LLM to translate the critique into French.
func (svc *Service) TranslateFrench(ctx context.Context, flow *workflow.Flow, critique string) (french string, err error) {
prompt := "Translate the following text into French. " +
"Respond with the translation only.\n\n" + critique
return svc.askLLM(ctx, prompt)
}
// TranslateSpanish asks the LLM to translate the critique into Spanish.
func (svc *Service) TranslateSpanish(ctx context.Context, flow *workflow.Flow, critique string) (spanish string, err error) {
prompt := "Translate the following text into Spanish. " +
"Respond with the translation only.\n\n" + critique
return svc.askLLM(ctx, prompt)
}
// askLLM sends a single user message and returns the assistant's reply.
func (svc *Service) askLLM(ctx context.Context, userMessage string) (string, error) {
messages := []llmapi.Message{{Role: "user", Content: userMessage}}
out, _, err := llmapi.NewClient(svc).Chat(ctx, svc.Provider(), svc.Model(), messages, nil, nil)
if err != nil {
return "", errors.Trace(err)
}
for i := len(out) - 1; i >= 0; i-- {
if out[i].Role == "assistant" && out[i].Content != "" {
return strings.TrimSpace(out[i].Content), nil
}
}
return "", errors.New("no assistant reply")
}
