Unifying Execution and Business States

In the previous two lessons, you built a stateless reducer agent that externalizes prompts and serializes context. The agent works correctly, solving complex problems like quadratic equations through multi-step tool use. However, when you look at the run() method's return signature, you see it returns a tuple of (context, status, final_answer). This scattered state makes the system harder to maintain, harder to debug, and more fragile to pause and resume because there is no single, structured object that captures everything about the agent's current situation. The solution is to create a unified State class that implements Factor 5 of the 12-Factor Agents methodology: unify execution state and business state.

Before building the unified State class, it helps to clarify what execution state and business state mean and why they both belong in the same object.

• Execution state refers to the metadata about where the agent is in its processing lifecycle, including how many steps have been taken, what the current status is, what work is pending, and whether any fatal errors have occurred.
• Business state refers to the domain-specific information the agent is working with, including the conversation context, the final answer once the agent completes its work, and any domain-specific data structures you might add later.

By merging execution state and business state into a single State object, you create a unified representation that is easier to reason about, persist, and pass between systems, enabling capabilities like pausing an agent mid-execution and resuming it later.

Before we implement the unified State class, let's understand how we need to extend our project structure to support state management. We're going to add a new models/ directory to organize state-related data structures and prepare for future state-related utilities.

We're going to extend the existing structure to include a dedicated directory for state models:

src/
├── core/
│   ├── agent.py
│   ├── models/                     # NEW: State and domain models
│   │   └── state.py                # Unified state representation
│   ├── prompts/
│   │   ├── base_system.md
│   │   └── context_format.md
│   ├── utils/
│   │   └── context_serializer.py
│   └── tools/
│       ├── schemas/
│       │   ├── math.json
│       │   └── final_answer.json
│       └── functions/
│           └── math.py
└── main.py

Here's what the new component adds to our architecture:

• models/ directory — Houses all data models that represent agent state and domain concepts, treating state as a first-class structured object
• state.py — Defines the State class that unifies execution state (steps, status, pending work) and business state (context, answers, errors) into a single validated structure

This structure separates state representation from agent logic, making it clear that the State object is a pure data container while the Agent class handles the processing logic. By organizing state models this way, you can validate state transitions independently, serialize state for persistence or debugging, and extend the state schema without modifying the agent's control flow. Now let's create the State class and refactor the agent to use it.

The unified State class uses Pydantic's BaseModel to provide validation, default values, and a clear structure. The class lives at src/core/models/state.py and defines seven fields that capture both execution and business state.

Create the file src/core/models/state.py and add the following implementation:

from typing import List, Any, Optional
from pydantic import BaseModel, Field


class State(BaseModel):
    # Unique identifier for tracking and persistence
    id: str
    
    # Execution state: processing step count
    steps: int = 0
    status: str = "running"
    
    # Business state: conversation history
    # Using default_factory ensures each instance gets its own list
    context: List[Any] = Field(default_factory=list)
    
    # Execution state: tool calls queued for the next step
    pending_tool_calls: List[Any] = Field(default_factory=list)
    
    # Optional fields populated during execution
    error: Optional[str] = None
    final_answer: Optional[str] = None

Here's what each field represents in the unified state:

• id — A unique identifier for this state instance, enabling tracking across systems and persistence layers
• steps — Counts how many processing cycles the agent has completed, used for both debugging and enforcing maximum step limits
• status — Tracks the agent's lifecycle stage ("running", "complete", "max_steps_reached", or "failed"), making control flow explicit
• context — Holds the complete conversation history including user messages, assistant responses, tool calls, and tool outputs
• pending_tool_calls — Stores tool calls that have been requested by the LLM but not yet executed, representing queued work
• error — Captures any fatal exception message that causes the agent loop to crash (used when status="failed")
• final_answer — Stores the agent's final response once processing completes successfully

The use of Field(default_factory=list) prevents the common Python pitfall where all instances share the same mutable default value, and the Optional type annotations make it explicit that error and final_answer can be None. With this class defined, you can refactor the agent to use it everywhere, starting with the _next_step method where most state mutations happen.

The _next_step method is where most state mutations happen, so this is the most significant refactoring. The method now receives a State object and returns a modified State object, implementing the reducer pattern at the state level. The method begins by incrementing the steps counter directly on the state object, then iterates through pending_tool_calls and processes each one. For each function call, the method extracts the call_name, call_arguments, and call_id, then persists the tool call into the unified context history.

def _next_step(self, state: State):
    # State carries both execution and business data (Factor 5)
    state.steps += 1

    # Process all queued tool calls
    for function_call in list(state.pending_tool_calls):
        call_name = function_call["name"]
        call_arguments = function_call["arguments"]
        call_id = function_call["call_id"]

        # Persist the tool call in the same state object
        state.context.append({
            "type": "function_call",
            "name": call_name,
            "arguments": json.dumps(call_arguments),
            "call_id": call_id
        })

This approach makes it clear how the execution progresses step by step, with all mutations flowing through the same State object. Now you need to handle the actual tool execution and update the state accordingly.

After appending the function call to the context, the match/case block handles tool execution. When the final_answer tool is called, the method clears pending_tool_calls, sets the status to "complete", stores the answer in state.final_answer, and returns immediately. For math tools like sum_numbers, the method executes the function within a try-except block to catch any errors, formats the result as JSON, and then removes the processed function call from pending_tool_calls before appending the output to the context.

        # Execute the tool based on its name
        match call_name:
            case "final_answer":
                # Agent is done - transition to complete status
                state.pending_tool_calls = []
                state.status = "complete"
                state.final_answer = call_arguments.get("answer")
                return state
                
            case "sum_numbers":
                # Execute tool with error handling
                try:
                    result = sum_numbers(**call_arguments)
                    output = json.dumps({"result": result})
                except Exception as e:
                    output = json.dumps({"result": f"Error: {str(e)}"})
            # ... other cases ...

        # Remove processed call and store output
        state.pending_tool_calls.remove(function_call)
        # Store tool output in the same state object
        state.context.append({
            "type": "function_call_output",
            "call_id": call_id,
            "output": output
        })

By flowing all results back into the State object, you maintain a complete execution trace that can be inspected, persisted, or replayed. After processing all pending tool calls, you need to call the LLM to get the next set of actions.

After executing all pending tool calls and appending their outputs to context, the method calls the LLM with the updated context using the existing _call_llm helper. The response contains new function calls, which are extracted and converted into dictionaries. These new tool calls are then added to state.pending_tool_calls for the next step, and the updated State is returned.

    # Call LLM with updated context including tool results
    response = self._call_llm(state.context)
    
    # Extract function calls from response
    function_calls = [item for item in response.output if item.type == "function_call"]

    # Convert to dictionaries for easier manipulation
    function_call_dicts = [
        {
            "name": fc.name,
            "arguments": json.loads(fc.arguments),
            "call_id": fc.call_id,
            "type": fc.type
        }
        for fc in function_calls
    ]

    # Queue new tool calls for the next step
    state.pending_tool_calls.extend(function_call_dicts)
    return state

This explicit queueing of pending work inside the State object makes the agent's control flow transparent and debuggable. Now you need to update the run() method to orchestrate these steps using the unified State and handle unexpected crashes.

The run() method becomes significantly simpler with unified state because it only needs to track a single object. To adhere to the stateless principle, the method creates a deep copy of the incoming state.

Crucially, we now handle resilience by wrapping the execution loop in a try/except block. Before starting, we ensure state.error is cleared. If a fatal error occurs (like an API outage or unhandled exception), we catch it, set state.status to "failed", and record the exception message in state.error.

def run(self, state: State):
    # Create a deep copy to avoid mutating the original
    state = state.model_copy(deep=True)

    # Initialize execution status and clear stale errors
    state.status = "running"
    state.error = None

    try:
        # Process steps until completion or max_steps reached
        while state.status == "running" and state.steps < self.max_steps:
            state = self._next_step(state)
            
    except Exception as e:
        # Capture fatal errors in the state object
        state.status = "failed"
        state.error = str(e)
        return state

    # Handle max_steps timeout
    if state.status == "running":
        state.status = "max_steps_reached"

    return state

This control flow is robust: success, timeout, and failure are all captured within the returned State object.

The main script demonstrates how to use the agent with the unified state pattern. You start by creating a State object with a unique id generated by uuid.uuid4() and initial context containing the user's request. You then call agent.run(state) and access the results directly through the returned State object.

import uuid
from core.agent import Agent
from core.models.state import State

agent = Agent()

# Create initial state with unique ID and user request
initial_state = State(
    id=str(uuid.uuid4()),
    context=[
        {
            "role": "user",
            "content": "Solve the root of this equation: x^2 - 5x + 6 = 0"
        }
    ],
    status="running"
)

# Run agent and receive a new state object (original is not mutated)
final_state = agent.run(initial_state)

print(f"Status: {final_state.status}")
if final_state.status == "failed":
    print(f"Error: {final_state.error}")
elif final_state.final_answer:
    print(f"Final answer: {final_state.final_answer}")

Running this script produces the following output, showing that the agent successfully solved the quadratic equation:

Status: complete
Final answer: The equation x^2 - 5x + 6 = 0 has roots x = 2 and x = 3.

The unified state pattern makes the API much cleaner than the previous tuple unpacking approach, and all execution information is available in one object for easy debugging and monitoring.

You have implemented Factor 5 by creating a single source of truth that captures everything about the agent's current situation in one State object. By using model_copy(deep=True), you've ensured that your agent functions as a pure transformer of state, taking one version and returning another without side effects on the input. This design not only simplifies the code but also enables powerful capabilities you will explore in future lessons: serializing state to JSON for persistence, replaying executions for debugging, and scaling horizontally by having different worker processes continue the work. In the upcoming practice exercises, you will extend the State class to handle more complex scenarios and explore how unified state simplifies error handling and recovery.