An AI-Driven Architecture for Autonomous Network Operations (NetOps)

from langchain.text_splitter import MarkdownHeaderTextSplitter
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

def build_knowledge_base(markdown_text):
 # 1. Split specific technical sections (e.g., "Error Codes", "Troubleshooting")
 headers_to_split_on = [
 ("#", "Header 1"),
 ("##", "Header 2"),
 ]
 markdown_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
 docs = markdown_splitter.split_text(markdown_text)

 # 2. Create Vector Store (The "Brain")
 # This converts text into numerical vectors that represent semantic meaning
 db = Chroma.from_documents(
 documents=docs, 
 embedding=OpenAIEmbeddings(),
 persist_directory="./network_knowledge_db"
 )
 db.persist()
 print("Knowledge Base Indexing Complete.")

import json
from langchain.chat_models import ChatOpenAI
from langchain.schema import SystemMessage, HumanMessage

def analyze_incident(log_entry):
 # 1. Retrieve Context
 db = Chroma(persist_directory="./network_knowledge_db", embedding=OpenAIEmbeddings())
 # Search for similar error codes or symptoms in the manual
 docs = db.similarity_search(log_entry, k=3)
 context_text = "\n\n".join([d.page_content for d in docs])

 # 2. Construct Prompt with Context
 system_prompt = """
 You are a Network Operations AI. 
 Analyze the log based ONLY on the provided context. 
 Output your decision as a JSON object with keys: "root_cause", "recommended_action", "confidence".
 Allowed actions: ["BLOCK_IP", "RESTART_SERVICE", "ESCALATE"].
 """

 user_prompt = f"""
 Context from Manuals:
 {context_text}

 Log Entry:
 {log_entry}
 """

 # 3. Get Decision
 llm = ChatOpenAI(temperature=0, model="gpt-4")
 response = llm.predict_messages([
 SystemMessage(content=system_prompt),
 HumanMessage(content=user_prompt)
 ])

    return json.loads(response.content)

def execute_remediation(decision):
 action = decision.get("recommended_action")
 confidence = decision.get("confidence")

 print(f"AI suggests: {action} with {confidence}% confidence.")

 # Guardrail: Only auto-execute high confidence actions
 if confidence < 90:
 return "Manual Intervention Required: Confidence too low."

 # Deterministic Execution Map
 if action == "BLOCK_IP":
 # Call actual Firewall API here
 return run_firewall_block_script()
 
 elif action == "RESTART_SERVICE":
 # Call SSH restart script
 return run_service_restart()
 
 elif action == "ESCALATE":
 return send_pagerduty_alert()
 
 else:
 return "Action not permitted."

def run_firewall_block_script():
 # Simulation of a network library call (e.g., Netmiko)
    return "SUCCESS: Firewall rule applied."

# Simulated incoming syslog message
incoming_log = "Apr 10 10:00:00 firewall-01 ALERT: Multiple failed login attempts from IP 192.168.1.50. Malware signature detected in payload."

# Step 1: Analyze
decision = analyze_incident(incoming_log)

# Step 2: Act
result = execute_remediation(decision)

print(f"Final Outcome: {result}")