Introduction: Why Agents Need State Machines

When building complex agents, we often fall into a common pitfall: treating the Agent as a stateless black box, starting from scratch on every call. However, in real-world business, agents often require multi-step reasoning, tool calls, user confirmations, and other interactions. Any single external call failure (such as API timeout or tool error) can cause the entire conversation chain to collapse. State machines provide a structured solution: they divide the agent's lifecycle into finite states (such as idle, thinking, tool_call, responding) and define legal transitions between states. By explicitly managing state, we can precisely control each step of the agent's behavior and also roll back to a safe state or feed error information back to the user when anomalies occur.

Based on practical experience with the DeepSeek API, this article delves into how to design state machines for agents and combines fault-tolerant orchestration strategies to build an agent that runs stably even in the face of unstable networks and tool failures. We will start from the principles, gradually implement an agent framework with a state machine, and share pitfalls encountered in engineering along with solutions. If you are already familiar with basic agent development, this tutorial will help you improve the robustness and maintainability of your system.

Core Elements of a State Machine: States, Events, Transitions

A state machine consists of three elements: states, events, and transitions. In the agent context, states can be idle (waiting for user input), processing (currently processing), tool_call (waiting for tool results), error (an error occurred), etc. Events trigger state changes, such as user_message, api_response, tool_timeout, etc. Transitions define the actions the agent should perform and the next state to transition to when a specific event is received in the current state.

When designing a state machine, I strongly recommend using configuration files to define transition rules rather than hardcoding them in business logic. The advantages are: better readability, easier extensibility, and convenient visualization. Below is an example of a state machine definition based on a Python dictionary, which clearly describes the response of each state under events.

STATE_MACHINE = {
    "idle": {
        "user_message": {"action": "handle_user", "next": "processing"},
        "error": {"action": "notify_user", "next": "idle"}
    },
    "processing": {
        "llm_done": {"action": "generate_reply", "next": "responding"},
        "llm_error": {"action": "retry_llm", "next": "processing"},
        "tool_required": {"action": "call_tool", "next": "tool_call"}
    },
    "tool_call": {
        "tool_success": {"action": "notify_llm", "next": "processing"},
        "tool_error": {"action": "handle_tool_error", "next": "error"},
        "tool_timeout": {"action": "abort_tool", "next": "error"}
    },
    "responding": {
        "message_sent": {"action": "reset", "next": "idle"},
        "message_failed": {"action": "retry_send", "next": "responding"}
    },
    "error": {
        "user_retry": {"action": "reset", "next": "idle"}
    }
}

This state machine is a simplified model, but it demonstrates how to embed fault tolerance into state transitions. For example, in the processing state, if the LLM call fails, we trigger the retry_llm action and stay in the same state rather than directly entering the error state, providing an opportunity for retry. In the tool_call state, a timeout is treated as an error, transitioning to the error state and waiting for user decision.

DeepSeek API Integration and State Management

DeepSeek provides an OpenAI-compatible API with base_url https://api.deepseek.com and model deepseek-chat. In actual coding, we need to integrate DeepSeek calls into the state machine. A common pitfall is API key leakage or configuration errors causing call failures. Therefore, when initializing the state machine, we should first validate the API configuration and provide clear error messages.

The following code demonstrates how to call the DeepSeek API in the processing state of the state machine. We use the requests library directly to better control timeouts and retries.

import requests
import json

def call_deepseek(messages, max_retries=2, timeout=30):
    headers = {
        "Authorization": "Bearer your-deepseek-api-key",
        "Content-Type": "application/json"
    }
    payload = {
        "model": "deepseek-chat",
        "messages": messages,
        "temperature": 0.7
    }
    response = None
    for attempt in range(max_retries):
        try:
            response = requests.post(
                "https://api.deepseek.com/chat/completions",
                headers=headers,
                json=payload,
                timeout=timeout
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.Timeout as e:
            print(f"Attempt {attempt+1} timed out: {e}")
        except requests.exceptions.ConnectionError as e:
            print(f"Attempt {attempt+1} connection error: {e}")
        except requests.exceptio
ns.HTTPError as e:
            code = response.status_code if response else None
            if code in (429, 500, 502, 503):
                print(f"Attempt {attempt+1} HTTP {code}, retrying...")
            else:
                raise e
    # Final failure, return error message
    return {"error": "DeepSeek API call failed", "detail": str(e)}
re>
Note that we have implemented a simple retry mechanism and catch timeouts and connection errors. In the state machine, when the return value contains an 'error' key, the llm_error event should be triggered to enter retry or error handling logic. If multiple retries still fail, notify the user and enter the error state.
Fault-Tolerant Orchestration: Retry, Timeout, and Degradation Strategies
Fault-tolerant orchestration is not just adding retries to API calls; it requires systematically designing strategies to handle different failures. Based on the stability of external dependencies, we can classify three levels: Level 1, transient faults (e.g., network jitter), usually solvable by retries; Level 2, partial faults (e.g., a model unavailable), can consider degrading to a backup model or API; Level 3, long-term faults (e.g., invalid API keys), require manual intervention.
In my engineering practice, I have established a fault classification table to guide the state machine's behavior decisions. The table below lists common fault types and recommended strategies:
Fault TypeExampleStrategyState Transition
Transient timeoutLLM call timeoutRetry 2 times with exponential backoffprocessing → processing
Rate limiting (429)Too frequent requestsWait and retry, or degrade to slow queueprocessing → processing
Model unavailableReturn 404Degrade to another model, e.g., deepseek-chat-0712processing → processing
Invalid API key401 UnauthorizedStop retrying, notify user to check configurationprocessing → error
Tool failureThird-party API returns errorRetry 1 time, if fails provide partial resultstool_call → tool_call
An important principle is: retries are not infinite loops. Each state should have a maximum retry count, and exceeding it must transition to an error state; otherwise, it wastes resources and causes infinite loops. Additionally, using exponential backoff during retries reduces pressure on the service, e.g., wait 1 second first, then 2 seconds, then 4 seconds.
Engineering Pitfalls: Context Loss and Recovery in State Machines
The biggest pitfall I encountered when implementing state machines is context loss. When an Agent handles multi-turn conversations, we need to store the conversation history (messages) in the state machine's context, but during abnormal transitions or service restarts, if not persisted, all context is lost, forcing the user to re-describe the problem.
The solution is to serialize the state machine and context together to a database or file. For example, we can convert the state object to JSON, including the current state, unsent message history, and necessary temporary variables. Persist after each transition. This way, even if the process crashes, it can resume from the last state. Below is an example of persisting state:
def save_state(session_id, state_machine, context):
    state_snapshot = {
        "session_id": session_id,
        "state": context.state,
        "messages": context.messages,
        "data": context.data
    }
    with open(f"sessions/{session_id}.json", "w") as f:
        json.dump(state_snapshot, f)

def load_state(session_id):
    try:
        with open(f"sessions/{session_id}.json", "r") as f:
            snapshot = json.load(f)
        # Restore context
        ctx = Context()
        ctx.state = snapshot["state"]
        ctx.messages = snapshot["messages"]
        ctx.data = snapshot["data"]
        return ctx
    except FileNotFoundError:
        return None

Additionally, in the state machine, event triggering may have concurrency issues, such as users sending multiple messages quickly. It is necessary to add locks or use unique session_id for isolation. If using a distributed environment, it is recommended to use Redis or other distributed locks. In my project, I simply used file locks, but production environments should adopt more robust solutions.
Real Case Analysis: Degradation after Tool Call Failure
A real case is when building a weather query Agent that relies on a third-party weather API. Once, that API returned 500 errors for several hours. Our state machine captured the error in the tool_call state, first retried 2 times, both failed. Then we triggered a degradation strategy: use a backup weather data source (e.g., a static cache or another free API), but if the backup also fails, tell the user "weather service temporarily unavailable" and preserve the conversation context for later retry.
During this process, the state machine went through transitions: tool_call → tool_call (retry) → tool_call (degradation) → tool_call (success). The key is that we did not lose the user's request context during degradation, and we recorded error information for log analysis. Ultimately, the user completed the query, albeit with a slightly degraded experience, but the entire system did not crash.
To ensure reliability of degradation, I also set up independent degradation functions for each external call and tested their return structure consistency. In the state machine, degradation flags are passed via event parameters, so business logic can clearly know which data source is currently used.
Performance Optimization: Timeout and Concurrency Control in State Machines
In high-concurrency scenarios, performance optimization of state machines is crucial. A common issue is waiting without timeout, e.g., LLM calls without setting timeout, causing threads to block indefinitely. We must set reasonable timeouts in API calls, and the entire state machine's single flow should also have a total timeout limit (e.g., 10 seconds), after which it forcibly transitions to the error state.
For concurrency control, you can use Python's asyncio to handle multiple sessions asynchronously. Each session's state machine can run independently, but attention must be paid to shared resource access, such as database connection pools. In my implementation, I adopted a single-threaded event loop with one task per session, avoiding lock complexity, but the premise is that external calls must be non-blocking (using async libraries or executed in a thread pool).
Below is a simple timeout control example using concurrent.futures to limit state machine processing time.
from concurrent.futures import ThreadPoolExecutor, TimeoutError

def run_state_machine_with_timeout(session_id, input, timeout_seconds=15):
    executor 
= ThreadPoolExecutor(max_workers=1)
    future = executor.submit(process_event, session_id, input)
    try:
        result = future.result(timeout=timeout_seconds)
        return result
    except TimeoutError:
        # Force switch to error state
        context = load_state(session_id)
        if context:
            context.state = "error"
            save_state(session_id, context)
        return {"error": "Processing timeout"}