Agent Workflows

Between a fixed chain and a fully autonomous loop sits most of the work worth building. A router that picks one of three chains. A retrieval step that retries once with a rephrased query when nothing clears the relevance floor. A drafting step that stops and waits for a person before anything is sent.

These are workflows: branching and repetition that you designed, with the model making bounded decisions inside a structure you control. The distinction from an agent is not how clever the flow is. It is who owns the graph. In a workflow you drew the edges, so you can read them, test them and cost them. In an agent the edges are chosen at runtime.

from dataclasses import dataclass, field

@dataclass
class FlowState:
    """Everything the flow knows. Not a transcript — fields you can assert on."""
    question: str
    caller: User
    route: str | None = None
    passages: list[Passage] = field(default_factory=list)
    retrieval_attempts: int = 0
    draft: str | None = None
    tokens_used: int = 0
    needs_approval: bool = False
    outcome: str | None = None

MAX_RETRIEVAL_ATTEMPTS = 2
TOKEN_BUDGET = 8_000

def handle(question: str, caller: User) -> FlowState:
    state = FlowState(question=question, caller=caller)
    state.route = classify(question, state)          # "policy" | "data" | "unsupported"

    if state.route == "unsupported":
        state.outcome = "out of scope"
        return state

    if state.route == "data":
        return answer_from_api(state)                # tool calling, validated per call

    # Policy route: retrieve, and allow one rephrased retry
    while not state.passages and state.retrieval_attempts < MAX_RETRIEVAL_ATTEMPTS:
        query = state.question if state.retrieval_attempts == 0 else rephrase(state)
        state.passages = search(query, caller=caller, top_k=5)
        state.retrieval_attempts += 1
        if state.tokens_used > TOKEN_BUDGET:
            state.outcome = "budget exhausted"
            return state

    if not state.passages:
        state.outcome = "no relevant documents"
        return state

    state.draft = generate_grounded_answer(state)
    state.outcome = "answered"
    log.info("flow complete", extra=state.audit())
    return state

• FlowState is the whole point of this sample. Every decision the flow made is a named field you can read, log and write a test against. Keeping state only in a growing message transcript means the same information exists as prose, where asserting on it requires parsing text the model wrote.
• classify is one cheap call that returns one of three fixed labels, validated against that list. The model is choosing among options you enumerated, which is a narrow and testable grant of autonomy. A fourth label coming back is a validation failure, not a new route.
• The unsupported branch exists so the flow can decline. Without an explicit out-of-scope route, every question gets pushed down whichever path is closest, and questions your system was not built for get answered anyway.
• The retrieval loop is bounded by a counter in the condition, not by the model deciding it has tried enough. One rephrased retry is worth having because a question worded unlike your documents is a common miss. A second retry rarely helps, and the cap says so.
• The budget check sits inside the loop, so a flow cannot spend past it between iterations. Checking only at the end tells you what it cost after you have paid.
• state.audit() emits one structured line holding the route, attempt count, chunk ids, tokens and outcome. When somebody asks why a question was declined, that line is the answer, and it does not require rerunning anything.

{
  "run_id": "flow-2f7a91c4",
  "caller_id": "EMP04417",
  "question_chars": 63,
  "route": "policy",
  "retrieval_attempts": 2,
  "rephrased": true,
  "chunk_ids": ["handbook-expenses#c0041", "handbook-expenses#c0042"],
  "top_similarity": 0.61,
  "citations_returned": [1],
  "model": "gpt-4.1-mini",
  "tokens_prompt": 3140,
  "tokens_completion": 212,
  "tokens_used_total": 3352,
  "duration_ms": 2870,
  "needs_approval": false,
  "outcome": "answered"
}

• route and retrieval_attempts together tell you which path ran and whether it struggled. A rising share of runs needing the rephrased retry is an early signal that your chunking or your corpus has drifted away from how people ask.
• chunk_ids and top_similarity let you reconstruct what the model was given without storing the documents again. A wrong answer with a top similarity of 0.61 is a retrieval problem; the same answer at 0.9 is a generation problem.
• citations_returned crossed against chunk_ids is how you spot an answer with no grounding. Log it per run and the rate becomes something you can watch rather than something you discover.
• Token fields belong here rather than only in an aggregate. Averages hide the runs that cost ten times the norm, and those are the ones worth reading.
• outcome is a small closed set — answered, no relevant documents, out of scope, budget exhausted, awaiting approval. Counting outcomes weekly is the cheapest health check this system has.
• Note what is absent: the question text and the answer text. Both may contain personal or commercial detail, so store them only where your data policy allows and keep the operational log safe to query widely.