Flow engineering

2 min read
On trails:AGCTXstep 9/11

Flow engineering means designing the sequence of steps, or state machine, a task moves through, instead of handing it to one big prompt. Use the probabilistic model only where judgement is needed. Decompose the task into stages, make the stages that need no model deterministic nodes, and add branching or routing between them. No one-shot big prompt.

A support flow: classify the ticket with an LLM, branch on the category. Billing tickets go to a policy lookup in code and then an LLM draft. Other tickets go to an escalate step in code. Both end in a reply. classify LLM branch billing lookup_policy code draft LLM reply other escalate code HIGHLIGHTED = LLM CALL (PROBABILISTIC). PLAIN = ORDINARY CODE (DETERMINISTIC).
The model classifies and writes. Code does the rest.

In LangGraph

The steps are nodes in a state graphState graphsAn agent drawn as nodes that read and write one shared state, joined by edges that say what runs next.Opens in a new tab, and how the state and edges are shaped is graph engineeringGraph engineeringDesigning the graph itself: what lives in state, how big a node is, where it loops and where it stops.Opens in a new tab.

Each LLM step gets one small job and a small prompt. They are the only places the flow can go wrong in a new way. The policy lookup and the escalation are code, so they behave the same every time and can be unit tested. The branch is an ordinary function on the state: routing is a decision made in code on top of a classification the model made.

Python
from typing_extensions import TypedDictfrom langgraph.graph import StateGraph, START, END class State(TypedDict):    ticket: str    category: str    policy: str    reply: str def classify(state: State) -> dict:                      # LLM: judgement    out = llm.invoke(f"Answer billing or other: {state['ticket']}")    return {"category": out.content.strip().lower()} def lookup_policy(state: State) -> dict:                 # code: no model needed    return {"policy": POLICIES["billing"]} def draft(state: State) -> dict:                         # LLM: wording    out = llm.invoke(f"Policy: {state['policy']}\nReply to: {state['ticket']}")    return {"reply": out.content} def escalate(state: State) -> dict:                      # code    return {"reply": "Passed to a human agent."} def branch(state: State) -> str:    return "lookup_policy" if state["category"] == "billing" else "escalate" flow = (    StateGraph(State)    .add_node("classify", classify)    .add_node("lookup_policy", lookup_policy)    .add_node("draft", draft)    .add_node("escalate", escalate)    .add_edge(START, "classify")    .add_conditional_edges("classify", branch, ["lookup_policy", "escalate"])    .add_edge("lookup_policy", "draft")    .add_edge("draft", END)    .add_edge("escalate", END)    .compile())

Why a flow

Anthropic separates workflows, where models and tools follow predefined code paths, from agents, where the model directs its own process, and advises adding complexity only when it demonstrably helps. Prompt chaining suits a task that splits cleanly into fixed steps. Routing allows separation of concerns and more specialised prompts, since tuning one prompt for every kind of input hurts some of them. The multi-agent version of routing is the router patternRouter patternA manager that only decides which specialist handles the request.Opens in a new tab.

A flow is also what makes an agent checkable: every node has an input and an output to log and test (see traceabilityTraceabilityLog the whole reasoning trace, so any answer can be explained afterwards.Opens in a new tab and agent evaluationAgent evaluationTest the flow an agent took and what it left behind, against the goal and the guardrails.Opens in a new tab).

Sources