Workflow Patterns

This content was generated with the assistance of AI and should be verified against the official documentation of each tool and framework before being relied on in production.

This section’s bibliography lists the reference material consulted while preparing these pages.

This page’s examples target Python 3.12 with LangChain 1.4.2 / langchain-core 1.6.5 / LangGraph 1.2.12 and langchain-ollama 1.1.0, Java 21 with LangChain4j 1.20.2 (langchain4j-agentic 1.20.2-beta30), and Spring Boot 4.1 with Spring AI 2.0.1, all calling DocsAssistant's Ollama llama3.1:8b chat model, as checked on 2026-09-28 (see the AI Agents baseline).

A workflow is a system where LLM calls and tools are composed through predefined code paths; an agent is one where the model itself decides the path and how many steps to take. Anthropic’s Building effective agents draws that line and names five workflow patterns that cover most production systems. They are composable, mostly deterministic pipelines: the code decides the order of the calls, and each call is a narrow step. That is the difference from Agent Loop & Architectures, which is about how a single agent step reasons when the model decides what comes next. In practice the two nest: a workflow step can be a small agent, and an agent’s tool can be a workflow.

The guidance that comes with the patterns is the most useful part: start with the simplest thing (one call with good retrieval and examples), add a workflow when a task decomposes cleanly, and reach for an open-ended agent only when the path genuinely cannot be known in advance. See the workflows vs. agents discussion on the landing page.

Pattern What the code fixes What the model decides

Prompt chaining

The sequence of steps, and the gates between them

The content of each step

Routing

The set of routes and what each one does

Which route an input takes

Parallelisation

The branches and how their outputs are aggregated

Each branch’s output (or vote)

Orchestrator-workers

That there is a decompose / work / synthesise shape

Which subtasks exist and how many

Evaluator-optimizer

The generate-evaluate loop and its retry budget

The draft, the verdict and the feedback

Prompt chaining

Break a task into a fixed sequence of calls, each consuming the previous one’s output, with optional programmatic gates between steps that check the intermediate result before continuing. Chaining trades latency for accuracy: each call does one easier thing. DocsAssistant uses it to turn a retrieved answer into a short, cited reply, with a gate that refuses to polish a draft that has no citation.

flowchart LR Q[Question] --> D[Draft with citations] D --> G{Gate: has a citation?} G -->|no| D G -->|yes| P[Polish to three sentences]
LangGraph: a two-step chain with a gate
from typing import TypedDict

from langchain_ollama import ChatOllama
from langgraph.graph import END, START, StateGraph

from docsassistant.search import search_docs

llm = ChatOllama(model="llama3.1:8b", temperature=0)


class ChainState(TypedDict):
    question: str
    draft: str
    answer: str


def draft(state: ChainState) -> dict:
    chunks = search_docs(state["question"], k=5)
    reply = llm.invoke(f"Answer from these chunks, citing source_path in brackets:\n{chunks}\n\n"
                       f"Question: {state['question']}")
    return {"draft": reply.content}


def has_citation(state: ChainState) -> str:          # programmatic gate between steps
    return "polish" if "[docs/" in state["draft"] else "draft"


def polish(state: ChainState) -> dict:
    reply = llm.invoke(f"Rewrite in at most three sentences, keeping every citation:\n{state['draft']}")
    return {"answer": reply.content}


builder = StateGraph(ChainState)
builder.add_node("draft", draft)
builder.add_node("polish", polish)
builder.add_edge(START, "draft")
builder.add_conditional_edges("draft", has_citation, ["polish", "draft"])
builder.add_edge("polish", END)
chain = builder.compile()

The gate is ordinary code, not a model call, which is what keeps a chain predictable. In production, give the retry edge its own counter (as the evaluator-optimizer below does) so a draft that never cites cannot loop until recursion_limit. See LangGraph v1 release notes.

In LangChain4j’s agentic module the same chain is a sequence workflow: each sub-agent writes its result to a named variable in the shared AgenticScope, and the next agent reads it through a @V parameter of the same name.

LangChain4j: sequenceBuilder()
public interface Drafter {
    @UserMessage("Answer from the docs, citing source paths in brackets: {{question}}")
    @Agent(description = "Drafts a cited answer", outputKey = "draft")
    String draft(@V("question") String question);
}

public interface Polisher {
    @UserMessage("Rewrite in at most three sentences, keeping every citation: {{draft}}")
    @Agent(description = "Shortens a cited draft", outputKey = "answer")
    String polish(@V("draft") String draft);
}

Drafter drafter = AgenticServices.agentBuilder(Drafter.class)
        .chatModel(chatModel).tools(new DocsTools()).build();
Polisher polisher = AgenticServices.agentBuilder(Polisher.class)
        .chatModel(chatModel).build();

UntypedAgent answerChain = AgenticServices.sequenceBuilder()
        .subAgents(drafter, polisher)
        .outputKey("answer")
        .build();

String answer = (String) answerChain.invoke(Map.of("question", "What does ef_search control?"));

See LangChain4j — Agents and Agentic AI (sequential workflow). Spring AI’s chain example is a loop over system prompts with plain ChatClient calls, each step’s output becoming the next step’s input:

Spring AI: the chain workflow
public class DocsAnswerChain {

    private static final String[] STEPS = {
        "Answer the question from the documentation context, citing source paths in brackets.",
        "Rewrite the answer in at most three sentences, keeping every citation."
    };

    private final ChatClient chatClient;

    public DocsAnswerChain(ChatClient.Builder builder, DocsTools docsTools) {
        this.chatClient = builder.defaultTools(docsTools).build();
    }

    public String run(String question) {
        String current = question;
        for (String step : STEPS) {
            current = chatClient.prompt()
                    .system(step)
                    .user(current)
                    .call()
                    .content();
        }
        return current;
    }
}

See Spring AI — Building Effective Agents and the runnable agentic-patterns examples.

Routing

Classify an input first, then send it to a specialised prompt, model or pipeline. Routing lets each route be tuned without degrading the others, and lets cheap inputs go to a cheap path. It works when the categories are distinct and the classifier is accurate; a misroute is the typical failure, so keep an explicit "none of the above" route.

flowchart LR Q[Question] --> C{Classify} C -->|concept| A[Explain with sources] C -->|howto| B[Steps and code] C -->|out of scope| D[Decline]
LangGraph: structured-output router
from typing import Literal, TypedDict

from langchain_ollama import ChatOllama
from langgraph.graph import END, START, StateGraph
from pydantic import BaseModel

from docsassistant.search import search_docs

llm = ChatOllama(model="llama3.1:8b", temperature=0)


class Route(BaseModel):
    kind: Literal["concept", "howto", "out_of_scope"]


class RouteState(TypedDict):
    question: str
    kind: str
    answer: str


def classify(state: RouteState) -> dict:
    route = llm.with_structured_output(Route).invoke(
        "Classify this question about the documentation site: concept (what/why), "
        f"howto (steps/code), or out_of_scope.\n{state['question']}"
    )
    return {"kind": route.kind}


def concept(state: RouteState) -> dict:
    chunks = search_docs(state["question"], k=5)
    return {"answer": llm.invoke(f"Explain the concept, citing sources: {chunks}").content}


def howto(state: RouteState) -> dict:
    chunks = search_docs(state["question"], k=8)
    return {"answer": llm.invoke(f"Give numbered steps and the code, citing sources: {chunks}").content}


def decline(state: RouteState) -> dict:
    return {"answer": "DocsAssistant only answers questions about this documentation site."}


builder = StateGraph(RouteState)
builder.add_node("classify", classify)
builder.add_node("concept", concept)
builder.add_node("howto", howto)
builder.add_node("decline", decline)
builder.add_edge(START, "classify")
builder.add_conditional_edges(
    "classify", lambda s: s["kind"],
    {"concept": "concept", "howto": "howto", "out_of_scope": "decline"},
)
for node in ("concept", "howto", "decline"):
    builder.add_edge(node, END)
router = builder.compile()

The Literal type on Route.kind constrains the classifier’s output to exactly the three route names. See LangChain — Structured output.

LangChain4j: conditionalBuilder()
public enum QuestionKind { CONCEPT, HOWTO, OUT_OF_SCOPE }

public interface QuestionClassifier {
    @UserMessage("Classify as CONCEPT, HOWTO or OUT_OF_SCOPE: {{question}}")
    @Agent(description = "Classifies a documentation question", outputKey = "kind")
    QuestionKind classify(@V("question") String question);
}

UntypedAgent routed = AgenticServices.conditionalBuilder()
        .subAgents(scope -> scope.readState("kind", QuestionKind.OUT_OF_SCOPE) == QuestionKind.CONCEPT,
                conceptExplainer)
        .subAgents(scope -> scope.readState("kind", QuestionKind.OUT_OF_SCOPE) == QuestionKind.HOWTO,
                howtoWriter)
        .subAgents(scope -> scope.readState("kind", QuestionKind.OUT_OF_SCOPE) == QuestionKind.OUT_OF_SCOPE,
                AgenticServices.agentAction(scope -> scope.writeState("answer",
                        "DocsAssistant only answers questions about this documentation site.")))
        .build();

UntypedAgent docsRouter = AgenticServices.sequenceBuilder()
        .subAgents(classifier, routed)
        .outputKey("answer")
        .build();

Each subAgents(Predicate<AgenticScope>, agents…​) call registers one route; the classifier agent runs first in the enclosing sequence and writes kind to the scope. The out-of-scope route needs no model call, so AgenticServices.agentAction(…​) wraps a plain lambda that writes the decline message to answer. See LangChain4j — Agents and Agentic AI (conditional workflow).

Spring AI: routing with a typed classification
public class DocsRouter {

    record Classification(String route, String reasoning) {}

    private static final Map<String, String> ROUTES = Map.of(
            "concept", "Explain the concept from the documentation, citing sources.",
            "howto", "Give numbered steps and the code from the documentation, citing sources.");

    private final ChatClient chatClient;

    public DocsRouter(ChatClient.Builder builder, DocsTools docsTools) {
        this.chatClient = builder.defaultTools(docsTools).build();
    }

    public String answer(String question) {
        Classification c = chatClient.prompt()
                .user("Classify as one of " + ROUTES.keySet() + " or out_of_scope: " + question)
                .call()
                .entity(Classification.class);
        String system = ROUTES.get(c.route());
        if (system == null) {
            return "DocsAssistant only answers questions about this documentation site.";
        }
        return chatClient.prompt().system(system).user(question).call().content();
    }
}

.entity(Classification.class) asks Spring AI to convert the reply into the record. See Spring AI — Building Effective Agents and Spring AI — ChatClient.

Parallelisation

Run independent calls at the same time and aggregate the results in code. It comes in two variants:

  • Sectioning — split a task into independent parts that run side by side, for example answering the question while a separate call screens it for abuse. Separating concerns this way beats asking one call to do both.

  • Voting — run the same task several times (or with different prompts) and aggregate, for example three reviewers that must all flag a page before DocsAssistant opens an issue. This is self-consistency applied at the workflow level.

flowchart LR Q[Question] --> A[Answer] Q --> S[Screen for abuse] A --> J[Join: answer only if allowed] S --> J
LangGraph: sectioning with two edges out of START
from typing import TypedDict

from langchain_ollama import ChatOllama
from langgraph.graph import END, START, StateGraph
from pydantic import BaseModel

from docsassistant.search import search_docs

llm = ChatOllama(model="llama3.1:8b", temperature=0)


class Screen(BaseModel):
    allowed: bool


class ParallelState(TypedDict):
    question: str
    answer: str
    allowed: bool
    final: str


def answer(state: ParallelState) -> dict:          # section 1: the actual work
    chunks = search_docs(state["question"], k=5)
    return {"answer": llm.invoke(f"{state['question']}\n{chunks}").content}


def screen(state: ParallelState) -> dict:          # section 2: an independent guardrail
    verdict = llm.with_structured_output(Screen).invoke(
        f"Is this a legitimate question about software documentation? {state['question']}"
    )
    return {"allowed": verdict.allowed}


def join(state: ParallelState) -> dict:
    return {"final": state["answer"] if state["allowed"] else "Request declined."}


builder = StateGraph(ParallelState)
builder.add_node("answer", answer)
builder.add_node("screen", screen)
builder.add_node("join", join)
builder.add_edge(START, "answer")                  # two edges out of START: both run in parallel
builder.add_edge(START, "screen")
builder.add_edge(["answer", "screen"], "join")     # join waits for both
builder.add_edge("join", END)
sectioned = builder.compile()

Because answer and screen write different state keys, no reducer is needed; add_edge with a list of sources makes join wait for both. The dynamic (number-of-branches-decided-at-runtime) form uses Send, as on query decomposition. See LangGraph v1 release notes.

LangChain4j: parallelBuilder() with an aggregating output
UntypedAgent answerAndScreen = AgenticServices.parallelBuilder()
        .subAgents(answerer, abuseScreener)        // write "answer" and "allowed"
        .executor(Executors.newFixedThreadPool(2))
        .output(scope -> scope.readState("allowed", false)
                ? scope.readState("answer")
                : "Request declined.")
        .build();

output(Function<AgenticScope, Object>) is the aggregation step; parallelMapperBuilder() is the variant that runs one agent over each item of a list (the voting form). See LangChain4j — Agents and Agentic AI (parallel workflow).

Spring AI: parallel ChatClient calls on an executor
public List<String> reviewInParallel(String proposedFix, List<String> reviewerPrompts) {
    ExecutorService pool = Executors.newFixedThreadPool(reviewerPrompts.size());
    try {
        List<CompletableFuture<String>> votes = reviewerPrompts.stream()
                .map(prompt -> CompletableFuture.supplyAsync(() -> chatClient.prompt()
                        .system(prompt)
                        .user(proposedFix)
                        .call()
                        .content(), pool))
                .toList();
        return votes.stream().map(CompletableFuture::join).toList();
    } finally {
        pool.shutdown();
    }
}

This mirrors Spring AI’s parallelization example, which fans the same prompt out over several inputs with a bounded pool. See Spring AI — Building Effective Agents.

Orchestrator-workers

A central LLM (the orchestrator) looks at the task, decides which subtasks it needs — how many and what each should do — dispatches them to workers, and a final step synthesises the results. It looks like parallelisation, but the subtasks are not known in advance: the orchestrator invents them per input. DocsAssistant uses it when a user reports that "the memory docs contradict each other": the orchestrator decides which pages to check and what to check on each.

flowchart LR R[Request] --> O[Orchestrator: decide subtasks] O --> W1[Worker: page 1] O --> W2[Worker: page 2] O --> W3[Worker: page n] W1 --> S[Synthesise] W2 --> S W3 --> S
LangGraph: dynamic subtasks with Send
import operator
from typing import Annotated, TypedDict

from langchain_ollama import ChatOllama
from langgraph.graph import END, START, StateGraph
from langgraph.types import Send
from pydantic import BaseModel

from docsassistant.search import search_docs

llm = ChatOllama(model="llama3.1:8b", temperature=0)


class Subtask(BaseModel):
    page: str            # which documentation page the worker should check
    instruction: str


class Assignment(BaseModel):
    subtasks: list[Subtask]


class OrchestratorState(TypedDict):
    request: str
    subtasks: list[Subtask]
    results: Annotated[list[str], operator.add]
    summary: str


def orchestrate(state: OrchestratorState) -> dict:
    plan = llm.with_structured_output(Assignment).invoke(
        f"Decide which documentation pages must be checked, and what to check on each, for: "
        f"{state['request']}"
    )
    return {"subtasks": plan.subtasks}


def assign(state: OrchestratorState) -> list[Send]:
    return [Send("worker", {"subtask": t}) for t in state["subtasks"]]


def worker(state: dict) -> dict:
    t: Subtask = state["subtask"]
    chunks = search_docs(t.page, k=4)
    return {"results": [f"{t.page}: " + llm.invoke(f"{t.instruction}\n{chunks}").content]}


def synthesize(state: OrchestratorState) -> dict:
    return {"summary": llm.invoke("Merge these findings:\n" + "\n".join(state["results"])).content}


builder = StateGraph(OrchestratorState)
builder.add_node("orchestrate", orchestrate)
builder.add_node("worker", worker)
builder.add_node("synthesize", synthesize)
builder.add_edge(START, "orchestrate")
builder.add_conditional_edges("orchestrate", assign, ["worker"])
builder.add_edge("worker", "synthesize")
builder.add_edge("synthesize", END)
orchestrator = builder.compile()

See LangGraph v1 release notes. When the workers are full agents with their own tools and context windows, this pattern becomes the supervisor topology on Multi-Agent Systems — and Claude Code subagents are the same shape for coding agents (see Orchestration Patterns).

In LangChain4j, the workflow form is a parallelMapperBuilder() over a list the orchestrator agent wrote to the scope; the fully model-driven form is the supervisorBuilder() agent, where the planner picks the next worker at each step (see Multi-Agent Systems). See LangChain4j — Agents and Agentic AI.

Spring AI: orchestrator and workers with typed plans
public class PageCheckOrchestrator {

    record Subtask(String page, String instruction) {}
    record Assignment(String analysis, List<Subtask> subtasks) {}

    private final ChatClient chatClient;

    public PageCheckOrchestrator(ChatClient.Builder builder, DocsTools docsTools) {
        this.chatClient = builder.defaultTools(docsTools).build();
    }

    public String process(String request) {
        Assignment plan = chatClient.prompt()
                .user("Decide which documentation pages to check, and what to check on each: " + request)
                .call()
                .entity(Assignment.class);

        List<String> findings = plan.subtasks().parallelStream()
                .map(t -> t.page() + ": " + chatClient.prompt()
                        .user(t.instruction() + " (page: " + t.page() + ")")
                        .call()
                        .content())
                .toList();

        return chatClient.prompt()
                .user("Merge these findings into one report:\n" + String.join("\n", findings))
                .call()
                .content();
    }
}

See Spring AI — Building Effective Agents and its orchestrator-workers example in spring-ai-examples.

Evaluator-optimizer

One call generates, another evaluates against explicit criteria and returns feedback, and the loop repeats until the evaluator passes the output or a round budget runs out. It works when there are clear evaluation criteria and when a human’s feedback would demonstrably improve the output — which is why it is the workflow form of reflection.

flowchart LR Q[Question] --> G[Generate] G --> E{Evaluate against rubric} E -->|revise + feedback| G E -->|pass or budget spent| A([Answer])
LangGraph: generator and evaluator with a round budget
from typing import Literal, TypedDict

from langchain_ollama import ChatOllama
from langgraph.graph import END, START, StateGraph
from pydantic import BaseModel

from docsassistant.search import search_docs

generator = ChatOllama(model="llama3.1:8b", temperature=0.3)
evaluator = ChatOllama(model="llama3.1:8b", temperature=0)


class Grade(BaseModel):
    verdict: Literal["pass", "revise"]
    feedback: str


class EvalState(TypedDict):
    question: str
    answer: str
    feedback: str
    rounds: int


def generate(state: EvalState) -> dict:
    chunks = search_docs(state["question"], k=5)
    prompt = f"{state['question']}\n{chunks}"
    if state.get("feedback"):
        prompt += f"\nRevise your previous answer:\n{state['answer']}\nFeedback: {state['feedback']}"
    return {"answer": generator.invoke(prompt).content, "rounds": state.get("rounds", 0) + 1}


def evaluate(state: EvalState) -> dict:
    grade = evaluator.with_structured_output(Grade).invoke(
        "Grade against the rubric: cites a source for every claim, answers the question asked, "
        f"no speculation.\nQuestion: {state['question']}\nAnswer: {state['answer']}"
    )
    return {"feedback": "" if grade.verdict == "pass" else grade.feedback}


def route(state: EvalState) -> str:
    return END if not state["feedback"] or state["rounds"] >= 3 else "generate"


builder = StateGraph(EvalState)
builder.add_node("generate", generate)
builder.add_node("evaluate", evaluate)
builder.add_edge(START, "generate")
builder.add_edge("generate", "evaluate")
builder.add_conditional_edges("evaluate", route, ["generate", END])
evaluator_optimizer = builder.compile()

Using a lower temperature for the evaluator than the generator is a common, cheap choice: the generator benefits from some variety, the grader from repeatability. A different (stronger) model for the evaluator — for example claude-sonnet-5 grading llama3.1:8b — is the usual next step. See LangGraph v1 release notes.

LangChain4j’s version is the loop workflow shown on Agent Loop & Architectures: a scorer agent and a reviser agent inside loopBuilder(), with exitCondition(…​) on the score and maxIterations(…​) as the budget. See LangChain4j — Agents and Agentic AI.

Spring AI: evaluator-optimizer loop
public class AnswerRefiner {

    record Evaluation(String verdict, String feedback) {}   // verdict: PASS or NEEDS_IMPROVEMENT

    private static final int MAX_ROUNDS = 3;
    private final ChatClient chatClient;

    public AnswerRefiner(ChatClient.Builder builder, DocsTools docsTools) {
        this.chatClient = builder.defaultTools(docsTools).build();
    }

    public String refine(String question) {
        String answer = chatClient.prompt().user(question).call().content();
        for (int round = 1; round < MAX_ROUNDS; round++) {
            Evaluation e = chatClient.prompt()
                    .system("Grade: every claim cited, answers the question, no speculation. "
                            + "Reply with verdict PASS or NEEDS_IMPROVEMENT and feedback.")
                    .user("Question: " + question + "\nAnswer: " + answer)
                    .call()
                    .entity(Evaluation.class);
            if ("PASS".equals(e.verdict())) {
                break;
            }
            answer = chatClient.prompt()
                    .user("Revise this answer to: " + question + "\n" + answer + "\nFeedback: " + e.feedback())
                    .call()
                    .content();
        }
        return answer;
    }
}

See Spring AI — Building Effective Agents and Spring AI — Building Effective Agents with Spring AI (blog).

Combining patterns

The patterns compose freely, and real systems are usually two or three deep. DocsAssistant’s "report a wrong page" flow is a router (is this a report or a question?) whose report route is an orchestrator-workers pass over the pages involved, whose synthesis step is an evaluator-optimizer loop that drafts the GitHub issue — and whose final "open the issue" step is a write action behind an approval gate (see Human-in-the-Loop). Keep each layer independently testable, which is exactly what component evaluation needs.

In other frameworks

Framework Workflow support

OpenAI Agents SDK

No workflow DSL; chains and routing are ordinary Python around Runner.run, routing is often done with handoffs, and agents can be exposed as tools for orchestrator-workers.

Claude Agent SDK

Orchestrator-workers maps to subagents declared with AgentDefinition in ClaudeAgentOptions.agents; other patterns are ordinary code around query().

Microsoft Agent Framework

Graph workflows via WorkflowBuilder (fan-out/fan-in edge groups, conditional edges), plus ready-made sequential, concurrent, handoff, group-chat and Magentic orchestrations.

CrewAI

Flows are the workflow layer: @start, @listen and @router decorators chain and route steps, and a flow step can kick off a Crew of agents.

See OpenAI Agents SDK, Claude Agent SDK, Microsoft Agent Framework and CrewAI.

References