Pipelines

A pipeline splits a task into named steps and runs each step as a separate model call. Every call sees the original input and the output of all earlier steps, so later steps can build on, check or correct earlier ones. Giving each stage its own call tends to produce more careful work than asking for everything at once, at the cost of one call per step.

The pipeline is defined by beliefs rather than code:

Belief	Meaning
pipeline(list steps)	The names of the steps, in the order they run
step(string name, string instruction)	The instruction given to the model for each step
stepOutput(string input, string name, string output)	Added by the pipeline: the output of each step for an input

Two classes use these beliefs. MultiStepChainOfThought uses a pipeline to answer a question. Pipeline runs the steps and leaves the agent to decide what to do with their outputs.

Multi-step chain of thought

MultiStepChainOfThought runs a pipeline of reasoning steps, then makes a final call that extracts the answer from the accumulated reasoning. Its default steps are:

Step	Instruction (summarised)
understand	Restate the problem; list the facts, unknowns and constraints
plan	Break the problem into ordered sub-steps
reason	Carry out the plan, showing the working
verify	Check the reasoning and correct any mistakes

The defaults are used only when the agent has no pipeline belief. To use your own steps, declare them as beliefs. This example from the reasoning-agents project replaces the defaults with three steps suited to a calculation:

import astra.langchain4j.reasoning.*;

agent Steps extends MultiStepChainOfThought {
    module System S;

    initial pipeline(["facts", "calculate", "check"]),
        step("facts", "List the quantities given in the problem and what is asked."),
        step("calculate", "Calculate the result, showing each operation."),
        step("check", "Check the calculation and correct it if necessary.");

    rule +!main(list args) {
        try {
            model.loadApiKeyFromEnv("OPENAI_API_KEY");
            model.initialize("gpt-4o-mini",
                "astra.langchain4j.configuration.OpenAiConfiguration", []);

            !solve("A room is 4.5 m by 3.2 m. Carpet costs 20 euro per square metre. "
                + "What does it cost to carpet the room?", string answer);
            console.println("Answer: " + answer);
        } recover {
            console.println("The question could not be answered.");
        }
        S.exit();
    }
}

Run it with mvn -Dastra.main=Steps. The agent prints the output of each step under its name, followed by the answer:

[main] ----- facts -----
...
[main] ----- calculate -----
...
[main] ----- check -----
...
[main] Answer: 288 euro

The example makes four calls: one for each step and one to extract the answer. The exact wording of the answer depends on the model.

Using Pipeline directly

A pipeline is not limited to answering questions. Any task that benefits from staged work, such as reviewing a document or drafting a plan, can be described as steps. Extend Pipeline and adopt:

!runPipeline(string input, string transcript)

transcript is bound to the outputs of all the steps, each under a ### name heading. Each output is also recorded as a stepOutput belief, so a plan can pick out the step it needs. This agent reviews a sentence in three steps and prints only the result of the last:

import astra.langchain4j.reasoning.*;

agent Review extends Pipeline {
    module System S;

    initial pipeline(["summarise", "critique", "improve"]),
        step("summarise", "Summarise the text in one sentence."),
        step("critique", "List up to three weaknesses in the text."),
        step("improve", "Rewrite the text to address the weaknesses.");

    rule +!main(list args) {
        try {
            model.loadApiKeyFromEnv("OPENAI_API_KEY");
            model.initialize("gpt-4o-mini",
                "astra.langchain4j.configuration.OpenAiConfiguration", []);

            string text = "Agents is programs that acts on there own.";
            !runPipeline(text, string transcript);
            foreach (stepOutput(text, "improve", string improved)) {
                console.println("Improved text: " + improved);
            }
        } recover {
            console.println("The pipeline could not be completed.");
        }
        S.exit();
    }
}

stepOutput beliefs remain after the pipeline finishes. They form a record of the agent’s work that other plans can query, but they also accumulate; remove them when they are no longer needed. If the same input is run through the pipeline twice, the beliefs from both runs are present.

Changing the steps at run time

Because the steps are beliefs, an agent can change its pipeline between tasks: add a step, reorder the list or replace an instruction. Change the beliefs before adopting the goal, and keep a single pipeline belief so that the order is unambiguous. A belief can only be removed once its arguments are bound, so query the current list in the plan’s context:

rule +!usePipeline(list steps) : pipeline(list old) {
    -pipeline(old);
    +pipeline(steps);
}

rule +!usePipeline(list steps) {
    +pipeline(steps);
}

With these rules, !usePipeline(["understand", "reason", "verify"]); replaces the steps that the next !solve(...) uses. Every name in the list must have a matching step belief. If one does not, the !step(...) subgoal for that name has no applicable plan and !solve(...) fails.

Where next?

Pipelines use the model’s knowledge alone. ReAct lets the model call tools that the agent provides.