Building AI applications purely on LLM chains usually ends in tears. One day, the model outputs crisp JSON, and the next, it returns a polite paragraph that breaks your downstream code. On the flip side, traditional state machines are bulletproof but rigid—they fail the moment a user submits ambiguous text or an unpredictable input.
I spent weeks debugging broken multi-prompt pipelines before realizing the solution was simple: we need to fuse deterministic state machines with LLM reasoning nodes. By using a durable workflow engine like Temporal to manage state transitions while letting LLMs make bounded decisions, you get reliable, deterministic AI systems that won't crash in production.
Here is how you can build a hybrid workflow engine from scratch.
A state machine needs explicit states, valid transitions, and strict rules. An LLM node takes raw context, evaluates it, and returns a single decision from a pre-defined set of valid actions.
In a hybrid setup, the state machine handles orchestration, state persistence, retries, and execution limits. The LLM acts purely as a routing function inside a single state. The LLM never updates global state directly. It simply suggests the next transition based on structured data.
If you are scaling enterprise systems, setting up robust workflow automation early prevents massive tech debt later.
+-----------------------------------------------------------+
| Temporal Workflow |
| |
| [State: DRAFT] ---> [Activity: LLM Evaluation Node] |
| | |
| v |
| Structured Decision |
| / \ |
| v v |
| [State: REQUIRES_HUMAN] [State: APPROVED] |
+-----------------------------------------------------------+
Step 2: Set Up Temporal and Dependencies
We will use Temporal for durable execution and Python for our workflow logic. First, spin up a local Temporal server using the Temporal CLI or Docker.
Install the required Python packages:
pip install temporalio openai pydantic
Make sure your Temporal server is running locally on port 7233.
Step 3: Define Rigid State Models with Pydantic
To enforce deterministic AI behavior, the LLM must return strict schema outputs. Never let the LLM output freeform text to control workflow state.
Create a file named schemas.py:
from enum import Enum
from pydantic import BaseModel, Field
class WorkflowState(str, Enum):
DRAFT = "DRAFT"
NEEDS_REVISION = "NEEDS_REVISION"
APPROVED = "APPROVED"
REJECTED = "REJECTED"
class LLMDecision(str, Enum):
APPROVE = "APPROVE"
REQUEST_CHANGES = "REQUEST_CHANGES"
REJECT = "REJECT"
class EvaluationResult(BaseModel):
decision: LLMDecision
reasoning: str = Field(description="Brief reason for the decision")
confidence_score: float = Field(description="Confidence between 0.0 and 1.0")
Step 4: Build the LLM Reasoning Activity
Temporal activities handle side effects like API calls or database operations. If an LLM call fails or hits a rate limit, Temporal automatically handles retries without restarting the whole workflow.
When teams start building complex AI agents, keeping side effects inside isolated activities is the single best decision they can make.
Create activities.py:
import json
import os
from openai import OpenAI
from temporalio import activity
from schemas import EvaluationResult, LLMDecision
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
@activity.defn
async def evaluate_document_activity(document_text: str) -> EvaluationResult:
prompt = f"""
Analyze the following text for compliance and quality.
Select APPROVE if it meets standard criteria.
Select REQUEST_CHANGES if minor edits are needed.
Select REJECT if it violates basic guidelines.
Document Text:
{document_text}
"""
response = client.beta.chat.completions.parse(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "You are a deterministic logic node in a workflow engine."},
{"role": "user", "content": prompt}
],
response_format=EvaluationResult
)
return response.choices[0].message.parsed
Step 5: Implement the State Machine Workflow
Now write the workflow that processes state transitions based on the output of the LLM node. Temporal guarantees that workflow state persists across crashes, server restarts, or network failures.
Create workflows.py:
from datetime import timedelta
from temporalio import workflow
from schemas import WorkflowState, LLMDecision, EvaluationResult
with workflow.unsafe.imports_passed_through():
from activities import evaluate_document_activity
@workflow.defn
class DocumentApprovalWorkflow:
def __init__(self):
self.current_state = WorkflowState.DRAFT
self.history = []
@workflow.run
async def run(self, document_text: str) -> dict:
self.history.append(f"Started in state: {self.current_state}")
# Execute LLM Reasoning Node
result: EvaluationResult = await workflow.execute_activity(
evaluate_document_activity,
document_text,
start_to_close_timeout=timedelta(seconds=30)
)
# Enforce State Machine Transitions based on LLM output
if result.decision == LLMDecision.APPROVE and result.confidence_score > 0.8:
self.current_state = WorkflowState.APPROVED
elif result.decision == LLMDecision.REQUEST_CHANGES:
self.current_state = WorkflowState.NEEDS_REVISION
else:
# Low confidence or explicit reject routes to REJECTED
self.current_state = WorkflowState.REJECTED
self.history.append(f"Transitioned to state: {self.current_state} | Reason: {result.reasoning}")
return {
"final_state": self.current_state,
"confidence": result.confidence_score,
"history": self.history
}
Step 6: Create the Worker and Trigger the Workflow
To run the workflow, set up a worker process to listen on a Temporal task queue.
Create run_worker.py:
import asyncio
from temporalio.client import Client
from temporalio.worker import Worker
from activities import evaluate_document_activity
from workflows import DocumentApprovalWorkflow
async def main():
client = await Client.connect("localhost:7233")
worker = Worker(
client,
task_queue="hybrid-workflow-queue",
workflows=[DocumentApprovalWorkflow],
activities=[evaluate_document_activity],
)
print("Worker started. Listening for tasks...")
await worker.run()
if __name__ == "__main__":
asyncio.run(main())
In a separate terminal, trigger the execution using execute_workflow.py:
import asyncio
from temporalio.client import Client
from workflows import DocumentApprovalWorkflow
async def main():
client = await Client.connect("localhost:7233")
sample_doc = "The policy document outlines terms for user data retention. All data is deleted after 30 days."
result = await client.execute_workflow(
DocumentApprovalWorkflow.run,
sample_doc,
id="doc-approval-001",
task_queue="hybrid-workflow-queue",
)
print("Workflow Execution Result:")
print(result)
if __name__ == "__main__":
asyncio.run(main())
Run the worker terminal first, then run the execution script. You will see Temporal step through the workflow states deterministically, using the LLM output strictly as a routing signal.
Moving Beyond Simple Logic
Building hybrid engines keeps your production code sane. State machines provide observability, deterministic retries, and guardrails. LLMs supply flexible evaluation and contextual understanding. Mixing them prevents unpredictable loop failures and runaway API bills.
If you are expanding your infrastructure and need expert developers to scale these systems, Gaper connects companies with vetted software engineers who specialize in production ready workflow orchestration and AI implementations.
Top comments (0)