Bangun Agent AI Sederhana dengan Language Natural

Era Baru AI yang Melampaui Batas Lama

idfuturetechin.com – Dulu, analisis data atau pembuatan aplikasi pintar selalu memerlukan tim developer, data engineer, dan pakar SQL yang mumpuni. Tapi belakangan ini, dengan hadirnya Large Language Models (LLMs) seperti ChatGPT, GPT‑4, Claude, dan Gemini, banyak hal yang dulu mustahil untuk pemula jadi mungkin.

LLMs sekarang dapat Mengonversi bahasa sehari‑hari ke query SQL, Membuat ringkasan data besar dalam sekejap, Membangun prototipe aplikasi tanpa menulis satu baris kode rumit.

Natural Language to SQL

Salah satu breakthrough paling nyata adalah Natural Language to SQL (NL2SQL). Kita tinggal ngomong atau ngetik pertanyaan, LLM yang dipadukan dengan toolkit SQL otomatis mengonversi ke query

Selain SQL, LLMs juga jago Generate skrip Python/JavaScript untuk ETL (Extract, Transform, Load), Membangun pipeline data di Airflow, Prefect, atau dbt, Debugging otomatis: minta LLM perbaiki kode yang error

Membuat Databricks Catalog Assistant

Pertama dan terutama, kita perlu memilih tumpukan teknologi kita. Kita memerlukan penyedia model, tools untuk membantu kita menegakkan struktur dalam alur agen kita, konektor ke basis data kita, dan UI sederhana untuk mendukung pengalaman mengobrol.

Secara singkat, kalian punya asisten yang bisa menjawab:

“Tabel mana yang simpan data transaksi 2024?”
“Tulis query untuk hitung total sales per customer bulan lalu.”

Pydantic AI

Apa itu Pydantic AI

Pydantic AI adalah ekstensi dari Pydantic (library validasi data di Python) yang mengintegrasikan LLM (Large Language Model) untuk parse natural language jadi instance model Pydantic.

from pydantic import BaseModel

class CatalogQuery(BaseModel):
    database: str
    schema: str
    action: str       # e.g. "list_tables", "describe_table"
    table: str = None

Dengan Pydantic AI, pengguna cukup ketik “list all tables in sales_schema” dan library akan bikin CatalogQuery(database="...", schema="sales_schema", action="list_tables")—langsung siap kita eksekusi.

Apa itu Databricks Unity Catalog?

Databricks Unity Catalog adalah sistem metadata management terpusat di Databricks. Unity Catalog menyimpan informasi:

• Database & schema
• Tabel & view
• Kolom & tipe data
• Permissions & lineage

Dengan REST API atau Databricks SQL endpoints, kita bisa tarik metadata ini lalu tampilkan di assistant. Sebelum masuk, pastikan kalian udah nyiapin:

1. Akun Databricks aktif dengan Unity Catalog sudah di-enable.
2. Token akses (Personal Access Token) untuk API Databricks.
3. Python 3.10+ terinstall di lokal atau di Databricks Workspace.
4. Install library: pip install pydantic pydantic-ai openai databricks-sdk
   • pydantic-ai buat parse NL → objek
   • openai (atau library LLM lain) untuk LLM backend
   • databricks-sdk untuk panggil REST API Databricks

Step 1: Definisikan Model Pydantic

Buka file models.py, lalu:

from pydantic import BaseModel, Field
from enum import Enum

class Action(str, Enum):
    LIST_DATABASES = "list_databases"
    LIST_SCHEMAS   = "list_schemas"
    LIST_TABLES    = "list_tables"
    DESCRIBE_TABLE = "describe_table"

class CatalogQuery(BaseModel):
    action: Action = Field(..., description="Jenis aksi yang diinginkan")
    database: str  = Field(None, description="Nama database di Unity Catalog")
    schema: str    = Field(None, description="Nama schema di Unity Catalog")
    table: str     = Field(None, description="Nama tabel, kalau DESCRIBE_TABLE")

Penjelasan:

• Action enum memetakan jenis query.
• CatalogQuery memuat parameter yang dibutuhkan.

Step 2: Setup LLM dengan Pydantic AI

Buat file assistant.py, lalu:

from pydantic_ai import create_llm_model
from models import CatalogQuery
import os

# Pilih provider LLM, misal OpenAI
LLM_API_KEY = os.getenv("OPENAI_API_KEY")
llm = create_llm_model(
    CatalogQuery,
    provider="openai",
    api_key=LLM_API_KEY,
    model_name="gpt-4o"  # atau model lain yang support function calling
)

Dengan create_llm_model, Pydantic AI melakukan function calling:

• Minta LLM parse teks user jadi payload dict sesuai CatalogQuery
• Validasi otomatis via Pydantic

Step 3: Buat Fungsi Pemanggil Databricks API

Di databricks_client.py:

from databricks import sql
import os

# Buat koneksi Databricks SQL Endpoint
connection = sql.connect(
    server_hostname=os.getenv("DATABRICKS_HOST"),
    http_path=os.getenv("DATABRICKS_HTTP_PATH"),
    access_token=os.getenv("DATABRICKS_TOKEN")
)

def list_databases():
    with connection.cursor() as cursor:
        cursor.execute("SHOW DATABASES")
        return cursor.fetchall()

def list_schemas(database: str):
    with connection.cursor() as cursor:
        cursor.execute(f"SHOW SCHEMAS IN {database}")
        return cursor.fetchall()

def list_tables(database: str, schema: str):
    with connection.cursor() as cursor:
        cursor.execute(f"SHOW TABLES IN {database}.{schema}")
        return cursor.fetchall()

def describe_table(database: str, schema: str, table: str):
    with connection.cursor() as cursor:
        cursor.execute(f"DESCRIBE TABLE {database}.{schema}.{table}")
        return cursor.fetchall()

Catatan: Semua query di sini menggunakan SQL natif Databricks; sesuai fokus artikel.

Step 4: Hubungkan Parser & API

Tambah di assistant.py:

from databricks_client import (
    list_databases, list_schemas, list_tables, describe_table
)
from typing import Any

def handle_query(user_text: str) -> Any:
    # Parse text user jadi objek CatalogQuery
    query: CatalogQuery = llm.parse(user_text)
    # Jalankan aksi sesuai atribut
    if query.action == Action.LIST_DATABASES:
        return list_databases()
    if query.action == Action.LIST_SCHEMAS and query.database:
        return list_schemas(query.database)
    if query.action == Action.LIST_TABLES and query.database and query.schema:
        return list_tables(query.database, query.schema)
    if query.action == Action.DESCRIBE_TABLE and query.database and query.schema and query.table:
        return describe_table(query.database, query.schema, query.table)
    return "Maaf, format perintah kurang tepat. Coba: 'list_tables in my_db.my_schema'."

Sekarang handle_query akan:

1. Transformasi teks → CatalogQuery
2. Eksekusi fungsi SQL
3. Return hasil (list of tuples)

Step 5: Buat Interface Sederhana

Untuk demo di blog atau internal, kita bisa pakai Streamlit:

# app.py
import streamlit as st
from assistant import handle_query

st.title("Databricks Catalog Assistant")
user_input = st.text_input("Tanya metadata Databricks:")

if st.button("Submit"):
    try:
        result = handle_query(user_input)
        st.write(result)
    except Exception as e:
        st.error(f"Error: {e}")

Jalankan:

streamlit run app.py

Dengan Pydantic AI + Databricks SQL, membuat Catalog Assistant jadi gampang dan cepat.

Dalam tahap demi tahap yang sudah kita bahas di ats, kalian telah belajar untuk menelusuri mekanisme awal pembuatan agen AI sederhana. Fokusnya adalah membuat prototipe ringan untuk membantu kalian memahami cara menyusun alur agen dan bereksperimen dengan perkakas AI modern.

jika ada salah kata dan salah penulisan atau bahkan salah menaruh rasa kami mohon maaf.

Sekian dan terima kasih !

Jangan lupa follow akun Instagram, Facebook, Youtube Digivestasi agar Anda tetap update