În acest capitol, intrăm în zona de programare pură și inginerie software. Învățăm cum să scriem scriptul central de execuție în Python, cum să integrăm conexiunea securizată prin request-uri HTTP REST către API-ul Gemini pentru a gestiona nativ apelurile de funcții și cum să închidem bucla autonomă pentru ca AI-ul să interogheze baza de date locală în mod direct, securizat și fără intervenție umană.

„Am ridicat zidurile fabricii noastre locale: mașina virtuală rulează în Proxmox, serverul MariaDB este activ, iar variabilele de mediu ne protejează cheile de acces. Acum este momentul să aducem muncitorul pe platformă: scriem scriptul Python care pornește prima buclă dintr-o autonomie reală.”

Bun venit la inima tehnică a proiectului nostru. Până acum am pregătit canalele de comunicare; astăzi scriem logica de execuție care transformă un model de limbaj dintr-un simplu chatbot într-un Agent decizional capabil să interacționeze cu lumea reală.

🔹Pregătirea Mediului de Lucru local (Virtualenv)

O regulă de aur în Python este să nu instalezi pachete direct în sistemul de operare global, pentru a nu strica dependențele Linux-ului. De aceea, în interiorul mașinii noastre virtuale, vom crea un mediu izolat de lucru (Virtual Environment).

Conectează-te prin SSH la serverul tău, creează un director dedicat proiectului și configurează spațiul de lucru rulând comenzile pas cu pas, pe rânduri separate:

# 1. Creăm directorul proiectului și intrăm în el
mkdir -p ~/warehouse_agent && cd ~/warehouse_agent

# 2. Creăm mediul virtual izolat
python3 -m venv venv

# 3. Activăm mediul izolat
source venv/bin/activate

# 4. Instalăm cele două biblioteci esențiale pentru Agentul nostru
pip install requests mysql-connector-python

🔹Scrierea Uneltelor Flexibile (Conexiunea SQL în Python)

Pentru ca Agentul să poată citi stocul din registrul nostru de marfă, trebuie să-i scriem o funcție Python curată. Pentru a evita blocajele în cazul în care utilizatorul trimite cuvinte parazite sau plurale, scriptul va sparge textul în cuvinte cheie și va genera o căutare dinamică tolerantă.

Asigură-te că ești în directorul proiectului (cd ~/warehouse_agent). Deschide fișierul principal folosind editorul text: nano agent.py și introdu codul care se ocupă de infrastructura SQL:

import os
import mysql.connector
import requests

# 1. Definim funcția locală flexibilă care interoghează baza de date nativă
def check_hardware_stock(component: str) -> str:
    try:
        # Deschidem conexiunea pe localhost folosind credențialele din sistem
        connection = mysql.connector.connect(
            host="127.0.0.1",
            user="agent_user",
            password=os.environ.get("DB_PASSWORD"),
            database="agent_warehouse"
        )
        cursor = connection.cursor()
        
        # Spargem string-ul în cuvinte individuale (ex: "module RAM" -> ['module', 'ram'])
        words = component.lower().strip().split()
        
        # Eliminăm sufixele comune de plural pentru o căutare flexibilă
        cleaned_words = []
        for word in words:
            if len(word) > 3:
                if word.endswith('uri'): word = word[:-3]
                elif word.endswith('ele'): word = word[:-3]
                elif word.endswith('e') or word.endswith('i') or word.endswith('s'): word = word[:-1]
            cleaned_words.append(word)
            
        # Generăm dinamic condițiile SQL de tip OR pentru fiecare cuvânt cheie
        conditions = " OR ".join(["LOWER(component_name) LIKE %s" for _ in cleaned_words])
        query = f"SELECT component_name, specs, stock_count FROM hardware_inventory WHERE {conditions}"
        params = [f"%{c}%" for c in cleaned_words]
        
        cursor.execute(query, params)
        result = cursor.fetchone()
        
        cursor.close()
        connection.close()
        
        if result:
            return f"Rezultat Local: Componenta '{result[0]}' ({result[1]}) are un stoc curent de {result[2]} unități."
        else:
            return f"Rezultat Local: Nicio componentă potrivită pentru '{component}' nu a fost găsită în baza de date."
            
    except Exception as e:
        return f"Eroare SQL locală: {str(e)}"

🔹Legarea Creierului și Circuitul de Function Calling

Pentru a înțelege în mod transparent mecanismul de Function Calling și structura exactă a mesajelor schimbate cu modelul, vom comunica direct cu API-ul Gemini prin protocolul HTTP REST. În cadrul acestui proiect folosim apeluri REST directe către API-ul Gemini, ceea ce ne oferă control complet asupra payload-ului și evită eventualele incompatibilități întâlnite în anumite versiuni ale SDK-urilor. În plus, această abordare ne permite să observăm clar fiecare etapă a dialogului dintre aplicația locală și modelul AI, fără straturi suplimentare de abstractizare.

Adaugă în continuare, în același fișier agent.py, logica centrală de control a buclei unificate și clientul REST pentru trimiterea parametrilor în Cloud:

# 2. Clientul REST pentru Gemini (Compatibil nativ cu noile chei de securitate)
class RESTGeminiClient:
    def __init__(self):
        self.api_key = os.environ.get("API_KEY")
        self.url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:generateContent?key={self.api_key}"
        self.headers = {"Content-Type": "application/json"}
        self.history = []
        
        self.system_instruction = (
            "Ești un gestionar de depozit precis. Rolul tău exclusiv este să răspunzi la întrebări legate de stocul hardware "
            "folosind unealta check_hardware_stock. Dacă utilizatorul îți pune întrebări din afara acestui domeniu, "
            "cum ar fi cultură generală, istorie sau capitale, refuză politicos să răspunzi, explicând că poți verifica doar stocul de componente."
        )
        
        # Declarăm semnătura funcției noastre locale pentru ca modelul din Cloud să o poată înțelege
        self.tools = [{
            "function_declarations": [{
                "name": "check_hardware_stock",
                "description": "Verifică stocul disponibil în depozit pentru o anumită componentă hardware.",
                "parameters": {
                    "type": "OBJECT",
                    "properties": {
                        "component": {
                            "type": "STRING",
                            "description": "Numele componentei hardware pentru care se face căutarea."
                        }
                    },
                    "required": ["component"]
                }
            }]
        }]

    def send_message(self, user_text: str, function_response=None):
        if function_response:
            self.history.append({
                "role": "function",
                "parts": [{
                    "functionResponse": {
                        "name": "check_hardware_stock",
                        "response": {"result": function_response}
                    }
                }]
            })
        else:
            self.history.append({
                "role": "user",
                "parts": [{"text": user_text}]
            })

        payload = {
            "contents": self.history,
            "tools": self.tools,
            "systemInstruction": {"parts": [{"text": self.system_instruction}]}
        }

        http_response = requests.post(self.url, headers=self.headers, json=payload)
        if http_response.status_code != 200:
            raise Exception(f"{http_response.status_code} {http_response.text}")
            
        json_data = http_response.json()
        response_part = json_data.get("candidates", [{}])[0].get("content", {}).get("parts", [{}])[0]
        
        self.history.append({
            "role": "model",
            "parts": [response_part]
        })
        return response_part

def run_agent(user_question: str, agent_client: RESTGeminiClient):
    print(f"\n[Utilizator]: {user_question}")
    print("[Sistem]: Se trimite contextul în Cloud...")
    
    try:
        response_part = agent_client.send_message(user_text=user_question)
        
        # Verificăm dacă modelul AI a emis o cerere de rulare a funcției noastre locale SQL
        if "functionCall" in response_part:
            function_call = response_part["functionCall"]
            function_name = function_call.get("name")
            arguments = function_call.get("args", {})
            
            if function_name == "check_hardware_stock":
                component_filter = arguments.get("component", "")
                local_db_result = check_hardware_stock(component=component_filter)
                print(f"[Sistem]: Se execută funcția locală -> {local_db_result}")
                
                # Trimitem rezultatul bazei de date înapoi în Cloud pentru formularea propoziției finale
                final_response = agent_client.send_message(user_text="", function_response=local_db_result)
                print(f"[Agent Autonom]: {final_response.get('text')}\n")
        else:
            print(f"[Agent Autonom]: {response_part.get('text')}\n")
            
    except Exception as e:
        print(f"[Eroare API]: {str(e)}\n")

# 3. Punctul de intrare pentru testul nostru de producție
if __name__ == "__main__":
    agent = RESTGeminiClient()
    
    # Testăm capacitatea de discernământ și execuție a Agentului
    run_agent("Mai avem module de RAM în depozit?", agent)
    run_agent("Câte plăci de bază Intel NUC mai sunt pe stoc?", agent)
    run_agent("Care este capitala Franței?", agent) # Un test care va lovi bariera de siguranță (Guardrail)

Salvează și închide fișierul (apasă CTRL+X, apoi Y și Enter).

⚠️ Notă pentru producție

Pentru simplitate didactică, exemplul utilizează fetchone() și o logică simplificată de normalizare a cuvintelor. Într-un sistem real se recomandă tratarea rezultatelor multiple, validarea suplimentară a inputului și mecanisme de căutare mai robuste, adaptate structurii bazei de date și cerințelor aplicației.

🔹Anatomia unei Decizii Autonome (Ce se întâmplă în spate)

Înainte de a apăsa enter pe execuție, este momentul să deschidem cutia neagră și să înțelegem coregrafia ascunsă din spatele codului. Mulți cred că AI-ul citește direct baza de date din Proxmox, dar realitatea este mult mai inteligentă și securizată.

Iată cele 4 etape ascunse care se execută la fiecare întrebare:

  1. Evaluarea Contextului în Cloud: Când trimiți textul „Mai avem module de RAM în depozit?”, textul ajunge la serverele Google Gemini împreună cu „contractul” funcției tale din Python. Modelul analizează intenția și observă instrucțiunea de sistem. Înțelege că utilizatorul vrea stocuri și că are la dispoziție o unealtă numită check_hardware_stock care primește un argument numit component.
  2. Generarea Argumentului și Cererea de Apel (Function Calling): Modelul AI NU returnează un răspuns text direct către utilizator. În schimb, el generează o instrucțiune specială în format JSON: {"name": "check_hardware_stock", "args": {"component": "module de RAM"}}. Agentul a dedus singur că din toată propoziția ta, cuvântul cheie de care are nevoie funcția locală se află acolo.
  3. Execuția Locală rigidă pe Localhost: Scriptul local Python interceptează această cerere JSON primită din Cloud, oprește temporar dialogul și rulează funcția ta pe mașina virtuală. Scriptul preia valoarea brută, o sparge în cuvinte, elimină pluralele, deschide ușa MariaDB prin localhost și rulează interogarea securizată cu wildcard-uri. Baza de date întoarce rândul real din stoc.
  4. Sinteza Finală a Răspunsului: Scriptul Python reîmpachetează valoarea brută a stocului și o trimite înapoi în Cloud într-al doilea apel API. Gemini citește datele fizice venite din mașina ta virtuală și, folosindu-și abilitățile de limbaj natural, formulează propoziția finală pe care o vezi pe ecran.

🔹Rularea testului secvențial

Este momentul adevărului. Cu mediul virtual activ și variabilele încărcate în sesiune, execută scriptul direct în terminalul serverului tău:

python agent.py

Urmărește cu atenție ce se întâmplă în consolă. Primele două interogări de stoc vor extrage instant datele din tabelele MariaDB de pe localhost. La a treia întrebare („Care este capitala Franței?”), vei observa validarea perfectă a logicii noastre: Agentul va refuza politicos să răspundă, demonstrând că bariera de siguranță (Guardrail) funcționează perfect.

🔹Evoluția către Chatbot-ul Interactiv Continuu

Testul liniar ne-a dovedit că Agentul știe să extragă date, dar pentru o utilizare reală într-o companie, avem nevoie de o conversație fluidă. AI-ul trebuie să aștepte comenzi noi și, foarte important, să își amintească ce s-a discutat anterior (persistența contextului).

Pentru a realiza acest lucru, vom crea un fișier nou: nano chat_agent.py și vom înveli logica REST de mai devreme într-o buclă infinită while True. Acest script va transforma terminalul nostru într-o fereastră de chat live:

import os
import mysql.connector
import requests

# 1. Funcția locală SQL rămâne cu variabile în engleză
def check_hardware_stock(component: str) -> str:
    try:
        connection = mysql.connector.connect(
            host="127.0.0.1", user="agent_user", password=os.environ.get("DB_PASSWORD"), database="agent_warehouse"
        )
        cursor = connection.cursor()
        words = component.lower().strip().split()
        cleaned_words = [w[:-3] if w.endswith('uri') or w.endswith('ele') else w[:-1] if w.endswith('e') or w.endswith('i') or w.endswith('s') else w for w in words if len(w) > 3]
        
        conditions = " OR ".join(["LOWER(component_name) LIKE %s" for _ in cleaned_words])
        query = f"SELECT component_name, specs, stock_count FROM hardware_inventory WHERE {conditions}"
        params = [f"%{c}%" for c in cleaned_words]
        
        cursor.execute(query, params)
        result = cursor.fetchone()
        cursor.close()
        connection.close()
        
        if result: return f"Rezultat Local: Componenta '{result[0]}' ({result[1]}) are un stoc curent de {result[2]} unități."
        else: return f"Rezultat Local: Nicio componentă potrivită pentru '{component}' nu a fost găsită în baza de date."
    except Exception as e:
        return f"Eroare SQL locală: {str(e)}"

# 2. Clientul REST pentru Chat Interactiv
class RESTGeminiChatSession:
    def __init__(self):
        self.api_key = os.environ.get("API_KEY")
        self.url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:generateContent?key={self.api_key}"
        self.headers = {"Content-Type": "application/json"}
        self.history = [] # Aici se salvează memoria conversației
        
        self.system_instruction = "Ești un gestionar de depozit precis. Rolul tău exclusiv este să răspunzi la întrebări legate de stocul hardware folosind unealta check_hardware_stock. Dacă utilizatorul îți pune întrebări din afara acestui domeniu, refuză politicos să răspunzi."
        self.tools = [{"function_declarations": [{"name": "check_hardware_stock", "description": "Verifică stocul disponibil în depozit pentru o anumită componentă hardware.", "parameters": {"type": "OBJECT", "properties": {"component": {"type": "STRING", "description": "Numele componentei hardware pentru care se face căutarea."}}, "required": ["component"]}}]}]

    def send_message(self, user_text: str, function_response=None):
        if function_response:
            self.history.append({"role": "function", "parts": [{"functionResponse": {"name": "check_hardware_stock", "response": {"result": function_response}}}]})
        else:
            self.history.append({"role": "user", "parts": [{"text": user_text}]})

        payload = {"contents": self.history, "tools": self.tools, "systemInstruction": {"parts": [{"text": self.system_instruction}]}}
        http_response = requests.post(self.url, headers=self.headers, json=payload)
        
        if http_response.status_code != 200: raise Exception(f"{http_response.status_code} {http_response.text}")
        response_part = http_response.json().get("candidates", [{}])[0].get("content", {}).get("parts", [{}])[0]
        
        self.history.append({"role": "model", "parts": [response_part]})
        return response_part

# 3. Bucla infinită care menține chat-ul activ în terminal
if __name__ == "__main__":
    print("[Sistem]: Se inițializează sesiunea de chat REST live cu Agentul...")
    chat = RESTGeminiChatSession()
    print("\n=== Chatbot Depozit Hardware Activ ===")
    print("Tastează întrebarea ta mai jos. Tastează 'exit' sau 'quit' pentru a închide.\n")
    
    while True:
        try:
            user_input = input("[Utilizator]: ")
            if user_input.lower() in ['exit', 'quit']:
                print("[Sistem]: Sesiune închisă. La revedere!")
                break
            if not user_input.strip(): continue
                
            print("[Sistem]: Se trimite mesajul în Cloud...")
            response_part = chat.send_message(user_text=user_input)
            
            if "functionCall" in response_part:
                function_call = response_part["functionCall"]
                if function_call.get("name") == "check_hardware_stock":
                    local_db_result = check_hardware_stock(component=function_call.get("args", {}).get("component", ""))
                    print(f"[Sistem]: Se execută funcția locală -> {local_db_result}")
                    final_response = chat.send_message(user_text="", function_response=local_db_result)
                    print(f"[Agent Autonom]: {final_response.get('text')}\n")
            else:
                print(f"[Agent Autonom]: {response_part.get('text')}\n")
        except Exception as e:
            print(f"[Eroare API]: {str(e)}\n")

Rulează acum în terminal python chat_agent.py. Terminalul tău a devenit un prompt viu. Poți pune o serie de întrebări legate (“Mai avem RAM?”, urmată de “Dar SSD-uri?”), iar AI-ul va păstra firul narativ, apelând transparent baza de date ori de câte ori are nevoie de cifrele reale. Pentru a închide programul, tastează pur și simplu exit.

💡 Magia din Spate: Circuitul Explicit de Function Calling Este esențial să înțelegi logica acestui capitol. În timp ce alte framework-uri abstractizează complet execuția din spatele scenei, comunicația directă prin REST ne oferă control total și securitate maximă asupra modului în care datele locale părăsesc infrastructura noastră locală.Scriptul tău Python local acționează ca un veritabil punct de frontieră (Gateway). Cloud-ul nu are acces direct la IP-ul tău, la serverul MariaDB sau la containerele Proxmox; el doar inspectează structura simbolică a funcției și emite o cerere de tip „Function Call”. Scriptul tău local interceptează această cerere, validează parametrii, interoghează baza de date pe localhost și retrimite doar rezultatul procesat înapoi în Cloud. Acest control explicit garantează că nicio interogare malițioasă nu poate altera structura bazei tale de date.

🛠️ Ce urmează în Episodul 7? Muncitorul nostru a intrat în fabrică: scriptul Python rulează perfect, comunică pe localhost cu MariaDB și traduce inteligent cererile umane în interogări matematice de stoc.În episodul următor, ridicăm privirea din consolă pentru a înțelege marea imagine de ansamblu. Vom analiza modul în care această arhitectură de Function Calling pe care am construit-o este complementară sistemelor de tip RAG (Retrieval-Augmented Generation) și vom învăța cum să alegem corect tehnologia în funcție de natura datelor noastre pentru a asigura eficiență maximă în producție!

Infrastructura ta locală a fost pusă în mișcare cu succes. Bucla de Function Calling a fost închisă nativ, atât pentru comenzi secvențiale, cât și pentru chat live.

Stay Free! Stay Hidden! Stay Autonomous!

⚙️ Appendix: Micro-Curs de Inginerie AI (Clarificări Tehnice)

  • Python Virtual Environment (venv): Un instrument care creează un director izolat ce conține propria sa instalare de Python și propriile biblioteci, împiedicând conflictele dintre pachetele diferitelor proiecte locale.
  • SQL Parameterization: Tehnică de programare în care variabilele dintr-o comandă SQL sunt transmise separat către serverul de baze de date (folosind argumente precum %s), blocând atacurile cibernetice de tip SQL Injection prin tratarea inputului strict ca text, nu ca instrucțiune executabilă.
  • REST HTTP API Communication: Metodă universală de interogare a modelelor de limbaj prin pachete HTTP standard direct la punctele terminale Google Cloud, eliminând complet dependențele rigide sau bug-urile din pachetele software intermediare.
  • System Instruction & Guardrails: Set de reguli comportamentale injectate în configurarea unui LLM pentru a-i dicta rolul și limitele operaționale, asigurându-se că Agentul refuză execuțiile sau răspunsurile din afara ariei sale stabilite de lucru (Out of Scope).