Building a Private Local AIOps Assistant: Part 4 (The State, Stream, and Thinking Layer

In Part 3, we achieved a massive milestone by lifting open-ended network instructions out of real-time application streams and baking our exact physical topology maps directly into an Ollama Modelfile binary. devops-buddy transformed from a slow, hallucinating loop engine into a highly accurate, deterministic configuration renderer.

But as our testing sessions progressed, we hit the final classic hurdle of raw LLM orchestration scripts: Total Context Amnesia.

Every time you execute a standalone client prompt loop, or restart your native terminal wrapper, the model wakes up with a completely blank canvas. It cannot remember what you discussed one minute ago. If you ask it to generate an underlay protocol for J1, and then follow up with "Now show me its neighbor's management IP," it loses track of what "its" refers to.

Today, we are going to fix this. We will transition devops-buddy into a fully stateful application by writing a custom, reusable SQLite3 connection handler from scratch, introducing a sliding-window history cache, and implementing live chunk streaming that captures the model's hidden "thinking" processes in real time.

Prerequisite: Make sure you have Ollama installed to follow along with this tutorial, you can download for FREE from: https://ollama.com

Part 1: Decoupling the Storage Core (db_connections.py)

My first structural goal was to avoid writing messy, tangled database calls directly into my main execution wrapper. In software engineering, you want a separate Data Access layer. This way, if I decide to upgrade my homelab later from an embedded local file to an enterprise PostgreSQL instance, I only have to modify my database wrapper class—my main application loop won't change at all.

I wrote a standalone module named db_connections.py. It uses Python's built-in sqlite3 driver to dynamically handle file connections, construct database tables safely on the fly from list-of-dictionary structures, and format parameterized insertion blocks to completely block SQL syntax errors.

Here is the decoupled backend code:

import sys
import sqlite3

class SQLiteConnection:

    def __init__(self, db_name: str):
        self.cursor = None
        self.connection = None
        try:
            self.connection = sqlite3.connect(db_name)
            self.cursor = self.connection.cursor()
        except sqlite3.Error as e:
            print(f"Database connection could not be established. Error: {e}")
            sys.exit(1)

    def test_connection(self):
        if self.cursor is not None:
            return True

    def create_table(self, table_name: str, fields: list[dict]) -> bool:
        exec_string = f"CREATE TABLE IF NOT EXISTS {table_name} ("
        for field in fields:
            exec_string += f" {field['field_name']} {field['type']} "
            if field.get('primary_key', False):
                exec_string += ' PRIMARY KEY'
            if field.get('auto_increment', False):
                exec_string += ' AUTOINCREMENT'
            if field.get('required', False):
                exec_string += ' NOT NULL'
            exec_string.strip(' ')
            exec_string += ','
        exec_string = exec_string[:-1]
        exec_string += ")"
        try:
            self.cursor.execute(exec_string)
            self.connection.commit()
        except sqlite3.Error as e:
            print(e)
            return False
        return True

    def describe_table(self, table_name: str):
        self.cursor.execute(f'PRAGMA table_info({table_name})')
        print(self.cursor.fetchall())

    def insert_into_table(self, table_name: str, data: dict):
        try:
            keys = ','.join(data.keys())
            placeholders = ','.join(['?' for _ in data.values()])
            query = f"INSERT INTO {table_name} ({keys}) VALUES({placeholders})"
            self.cursor.execute(query, tuple(data.values()))
            self.connection.commit()
        except sqlite3.Error as e:
            print(e)
            return False
        return True

    def get_table_data(self, table_name: str, no_of_rows: int = 4) -> list[str]:
        self.cursor.execute(f"SELECT * FROM {table_name} ORDER BY id DESC LIMIT {no_of_rows}")
        return self.cursor.fetchall()

    def delete_all(self, table_name: str):
        try:
            query = f'DELETE FROM {table_name};'
            self.cursor.execute(query)
            self.connection.commit()
        except sqlite3.Error as e:
            print(e)
            return False
        return True

Part 2: The Sliding Window, Live Streaming, and Thinking Loop

With the storage engine built, I developed the primary application script: data_loader.py.

To keep my 12-core Xeon CPU from lagging over long conversational sessions, I used the database to create a Sliding-Window Memory Cache. Every line we chat is saved permanently to the .db file file on disk, but our code uses get_table_data(no_of_rows=4) to extract only the 4 most recent chronological messages to pass to Ollama. This keeps our prompt context window tightly controlled and consistently fast.

Furthermore, running an advanced reasoning model like Gemma 4 12B on a CPU creates a massive silent delay if you use standard blocking API chat calls. The model spends its first several seconds mapping out an internal chain of logic, leaving the engineer staring at a frozen prompt screen.

To solve this, I activated stream=True and think=True inside the Ollama client wrapper. I configured a custom loop to instantly intercept the text fragments as they drop. If the model is outputting its hidden reasoning tokens, it prints them immediately in custom terminal Grey text, before seamlessly transitioning to the final answer payload.

Here is the orchestration app engine:

import subprocess
import ollama
from db_connections import SQLiteConnection

GREY = "\033[90m"
RESET = "\033[0m"

class DevopsBuddyChat:

    def __init__(self, db_name: str='devops_buddy_chats.db'):
        self.db_connection = SQLiteConnection(db_name)
        self.db_connection.create_table(table_name='chats', fields=[
            {'field_name': 'id', 'type': 'INTEGER', 'primary_key': True, 'auto_increment': True},
            {'field_name': 'role', 'type': 'TEXT', 'required': True},
            {'field_name': 'content', 'type': 'TEXT', 'required': True}
        ])
        if not self.db_connection.test_connection:
            raise ValueError("Couldn't initiate database connection.")

    def chat(self):
        user_input = ''
        while True:
            try:
                print()
                user_input = input("Ask: [bye to quit, delete to delete all chats, clear to clear screen] ")
            except (KeyboardInterrupt, EOFError):
                print("\nSession ended by user interrupt.")
                break

            if user_input.lower() == 'bye':
                break
            if user_input.lower() == 'clear':
                subprocess.run('clear')
                continue
            if user_input.lower() == 'delete':
                self.db_connection.delete_all('chats')
                continue
            if not user_input.strip():
                continue

            # Save your query directly to the on-disk storage file
            self.db_connection.insert_into_table('chats', {
                'role': 'user', 'content': user_input.lower()
            })

            messages = []
            assistant_content = ''
            rows = self.db_connection.get_table_data(table_name='chats')

            # Map the rows into the standard message matrix timeline
            for id, role, content in rows:
                messages.append({"role": role, "content": content})

            # Fire the live-streaming reasoning connection
            assistant_response = ollama.chat(
                model='devops-buddy-gemma4.12b:v1',
                messages=messages,
                stream=True,
                think=True
            )

            # Process and capture the text pieces smoothly
            for chunk in assistant_response:
                if chunk.message.thinking:
                    print(f"{GREY}{chunk['message']['thinking']}{RESET}", end='', flush=True)
                if chunk.message.content:
                    assistant_content += chunk['message']['content']
                    print(chunk['message']['content'], end='', flush=True)

            # Securely log the final compiled response string back to SQL rows
            self.db_connection.insert_into_table('chats', {
                'role': 'assistant', 'content': assistant_content
            })

if __name__ == "__main__":
    buddy = DevopsBuddyChat()
    buddy.chat()

The Verification: Flawless Graph Memory

With our decoupled state backend and chunk tracking filters completely up and running, I launched python data_loader.py to challenge devops-buddy-gemma4.12b:v1 on a cross-reference relationship test.

Look at this incredibly clean, zero-hallucination interactive terminal run:

python data_loader.py

Ask: [bye to quit, delete to delete all chats, clear to clear screen] is there any link between J2 and J4, if there is a link what is the ip address on both sides of the link?

Yes, there is a link between J2 and J4 (identified as P2P Link 3 in the topology).

The IP addresses for this link are:
* J2 side: 172.16.24.2/24 on interface em3
* J4 side: 172.16.24.4/24 on interface em3

Ask: [bye to quit, delete to delete all chats, clear to clear screen]

The application successfully achieved architectural alignment. By pairing the compiled Modelfile blueprint with a persistent SQLite sliding-window transaction history database, the model accurately isolated Link 3, tracked that J2 maps to em3 with .2, and cross-referenced that J4 maps to em3 with .4.

No invented interfaces, no command leakage, and no CPU context stalls.

Architectural Conclusion & Next Steps

Our localized NetDevOps ecosystem has transformed significantly across these past four phases: * The Environment (Part 1): Built a private CPU-based hypervisor workspace runtime to ensure complete corporate configuration privacy. * The Compilation Core (Part 2 & Part 3): Discovered the limits of open-ended prompt calculations, moving to explicit, static hardware caching inside model weights. * The Memory Management Engine (Part 4): Wrote a persistent SQLite3 layer to give our assistant continuous session integrity without token bloat.

Now that the system-prompt storage variables and session timeline states are fully functional, we are completely ready to expand our platform's operational abilities.

In Part 5, we will look into building some memory for our netdevops-buddy, at the moment we are storing conversations but our assistant should also be capable of storing and remembering its responses in a way that if a same question is asked it should not re-think, it should just provide the same answer or kind of same answer!