Factor 08 Own Your Control Flow - 12 Factor Agents

8. Own your control flow

If you own your control flow, you can do lots of fun things.

Build your own control structures that make sense for your specific use case. Specifically, certain types of tool calls may be reason to break out of the loop and wait for a response from a human or another long-running task like a training pipeline. You may also want to incorporate custom implementation of:

summarization or caching of tool call results
LLM-as-judge on structured output
context window compaction or other memory management
logging, tracing, and metrics
client-side rate limiting
durable sleep / pause / "wait for event"

The below example shows three possible control flow patterns:

request_clarification: model asked for more info, break the loop and wait for a response from a human
fetch_git_tags: model asked for a list of git tags, fetch the tags, append to context window, and pass straight back to the model
deploy_backend: model asked to deploy a backend, this is a high-stakes thing, so break the loop and wait for human approval

python

def handle_next_step(thread: Thread):

  while True:
    next_step = await determine_next_step(thread_to_prompt(thread))
    
    # inlined for clarity - in reality you could put 
    # this in a method, use exceptions for control flow, or whatever you want
    if next_step.intent == 'request_clarification':
      thread.events.append({
        type: 'request_clarification',
          data: nextStep,
        })

      await send_message_to_human(next_step)
      await db.save_thread(thread)
      # async step - break the loop, we'll get a webhook later
      break
    elif next_step.intent == 'fetch_open_issues':
      thread.events.append({
        type: 'fetch_open_issues',
        data: next_step,
      })

      issues = await linear_client.issues()

      thread.events.append({
        type: 'fetch_open_issues_result',
        data: issues,
      })
      # sync step - pass the new context to the LLM to determine the NEXT next step
      continue
    elif next_step.intent == 'create_issue':
      thread.events.append({
        type: 'create_issue',
        data: next_step,
      })

      await request_human_approval(next_step)
      await db.save_thread(thread)
      # async step - break the loop, we'll get a webhook later
      break

This pattern allows you to interrupt and resume your agent's flow as needed, creating more natural conversations and workflows.

Example - the number one feature request I have for every AI framework out there is we need to be able to interrupt a working agent and resume later, ESPECIALLY between the moment of tool selection and the moment of tool invocation.

Without this level of resumability/granularity, there's no way to review/approve the tool call before it runs, which means you're forced to either:

Pause the task in memory while waiting for the long-running thing to complete (think while...sleep) and restart it from the beginning if the process is interrupted
Restrict the agent to only low-stakes, low-risk calls like research and summarization
Give the agent access to do bigger, more useful things, and just yolo hope it doesn't screw up

You may notice this is closely related to factor 5 - unify execution state and business state and factor 6 - launch/pause/resume with simple APIs, but can be implemented independently.

← Contact Humans With Tools | Compact Errors →