Önceki makalede, Deep Agents CLI aracılığıyla Deep Agent'ın Skills modelini nasıl kullandığını incelemiştik. Artık LangChain bu özelliği yerel olarak destekliyor ve geliştirme sürecini büyük ölçüde basitleştiriyor. Bu makale, bu özelliği derinlemesine deneyimlemenize ve daha akıllı bir SQL asistanı oluşturmanıza rehberlik edecek.

Karmaşık bir AI Agent oluştururken, geliştiriciler genellikle bir ikilemle karşı karşıya kalırlar: Tüm bağlamı (veritabanı tablo yapısı, API belgeleri, iş kuralları) tek seferde System Prompt'a enjekte etmek, bağlam penceresinin (Context Window) taşmasına ve modelin dikkatini dağıtmasına mı neden olur? Yoksa maliyetli ve sık sık yapılan ince ayarları (Fine-tuning) mı seçmek gerekir?

Skills Modeli (Skills Pattern), zarif bir ara yol sunar. Gerekli bilgileri dinamik olarak yükleyerek, bağlamın verimli bir şekilde kullanılmasını sağlar. LangChain'in bu modeli yerel olarak desteklemesi, "isteğe bağlı öğrenme" yeteneğine sahip bir Agent'ı daha kolay oluşturabileceğimiz anlamına gelir.

Bu makale, resmi belgeler olan Build a SQL assistant with on-demand skills ile birlikte, okuyuculara sıfırdan başlayarak "isteğe bağlı bilgi yüklemeyi" destekleyen bir SQL Asistanı oluşturma konusunda rehberlik edecektir.

1. Temel Kavramlar: Neden Skills Modelini Seçmeliyiz?

Geleneksel SQL Agent'ın Sınırlamaları

Geleneksel SQL Agent mimarisinde, genellikle System Prompt'ta eksiksiz bir Veritabanı Şeması sağlamamız gerekir. İş geliştikçe, tablo sayısı yüzlerceye çıktığında, bu yaklaşım önemli sorunlara yol açacaktır:

• Büyük Token Tüketimi: Her konuşma, çok sayıda ilgisiz tablo yapısı taşır ve kaynak israfına neden olur.

• Halüsinasyon Riski Artışı: Çok fazla ilgisiz parazit bilgisi, modelin çıkarım doğruluğunu azaltacaktır.

• Bakım Zorluğu: Tüm iş hatlarının bilgisi sıkıca bağlantılıdır ve bağımsız olarak yinelemek zordur.

Skills Modeli: Aşamalı Açıklamaya Dayalı Çözüm

Skills Modeli, aşamalı açıklama (Progressive Disclosure) ilkesine dayanır ve bilgi edinme sürecini katmanlara ayırır:

• Agent'ın Başlangıç Durumu: Yalnızca hangi "becerilere" (Skills) sahip olduğunu ve bunların kısa açıklamalarını (Description) bilir, hafif kalır.

• Çalışma Zamanında Yükleme: Belirli bir sorunla (örneğin, "stok sorgulama") karşılaştığında, Agent, bu becerinin ayrıntılı bağlamını (Şema + İstek) yüklemek için aracı (load_skill) aktif olarak çağırır.

• Görevi Yürütme: Yüklenen kesin bağlama dayanarak, belirli bir görevi yürütür (örneğin, SQL yazma ve yürütme).

Bu model, sınırsız genişlemeyi ve ekip ayrışmasını etkili bir şekilde destekler ve Agent'ın giderek karmaşıklaşan iş senaryolarına uyum sağlamasına olanak tanır.

2. Sistem Mimarisi Tasarımı

Bu uygulama projesi, bu modelin pratik uygulamasını göstermek için iki temel Skills içeren bir SQL Asistanı oluşturacaktır:

• Sales Analytics (Satış Analizi):sales_datatablosundan sorumludur, gelir istatistikleri, sipariş trend analizi vb. ile ilgilenir.

• Inventory Management (Stok Yönetimi):inventory_itemstablosundan sorumludur, stok seviyesi izleme, konum sorgulama vb. ile ilgilenir.

3. Geliştirme Ortamı Kurulumu

Bu proje, verimli bağımlılık yönetimi için Pythonuv kullanır.

Temel Bağımlılıkların Yüklenmesi

uv add langchain langchain-openai langgraph psycopg2-binary python-dotenv langchain-community

PostgreSQL Ortam Yapılandırması

Yerel olarak bir Postgres örneği başlatın veagent_platformveritabanını oluşturun. Tablo yapısını ve test verilerini otomatik olarak başlatmak içinsetup_db.pykomut dosyasını sağladık (ayrıntılar için makalenin sonundaki kaynak koduna bakın).

4. Temel Uygulama Adımlarının Ayrıntılı Açıklaması### Adım Bir: Alan Becerilerini Tanımlama (Bilgi)

Becerileri, dosya sisteminden veya veritabanından yükleme sürecini simüle eden bir sözlük yapısı olarak tanımlayacağız. Lütfen description (Agent'ın karar verme seçimi için) ve content (gerçek yüklenen ayrıntılı bağlam) arasındaki farka dikkat edin.

SKILLS = {"sales_analytics": {"description":"Satış gelirini, trendleri analiz etmek için kullanışlıdır...","content":"""... Tablo Şeması: sales_data ..."" },"inventory_management": {"description":"Stok seviyelerini kontrol etmek için kullanışlıdır...","content":"""... Tablo Şeması: inventory_items ..."" }}

Adım İki: Temel Araçları Uygulama (Yetenekler)

Agent, görevleri tamamlamak için iki temel araca güvenir:

• load_skill(skill_name): Belirtilen becerinin ayrıntılarını çalışma zamanında dinamik olarak yükler.

• run_sql_query(query): Belirli SQL ifadelerini yürütür.

Adım Üç: Agent Mantığını Düzenleme (Beyin)

LangGraph'ı kullanarak ReAct Agent'ı oluşturun. Sistem İstemi burada önemli bir rol oynar ve Agent'ı Identify -> Load -> Query standart çalışma prosedürünü (SOP) kesinlikle izlemesi için yönlendirir.

system_prompt ="""1. İlgili beceriyi belirleyin.2. Şema almak için 'load_skill' kullanın.3. 'run_sql_query' kullanarak SQL yazın ve yürütün....Tablo adlarını tahmin etmeyin. Her zaman önce beceriyi yükleyin.""

5. Çalışma Etkisi Doğrulaması

test_agent.py'yi çalıştırarak, Satış ve Envanter olmak üzere iki farklı alandaki sorguları test ettik. Aşağıda, Agent'ın sorunlara göre becerileri dinamik olarak nasıl yüklediğini gösteren konsolun gerçek çıktı günlükleri bulunmaktadır:

Testing Sales Query...Agent calling tools: [{'name': 'load_skill', 'args': {'skill_name': 'sales_analytics'}, 'id': 'call_f270d76b7ce4404cb5f61bf2', 'type': 'tool_call'}]Tool output:You are a Sales Analytics Expert.You have access to the 'sales_data' table.Table Schema:- id: integer...Agent calling tools: [{'name': 'run_sql_query', 'args': {'query': 'SELECT SUM(amount) as total_revenue FROM sales_data;'}, 'id': 'call_b4f3e686cc7f4f22b3bb9ea7', 'type': 'tool_call'}]Tool output: [(Decimal('730.50'),)]...Agent response: The total revenue is $730.50.Testing Inventory Query...Agent calling tools: [{'name': 'load_skill', 'args': {'skill_name': 'inventory_management'}, 'id': 'call_18c823b2d5064e95a0cfe2e3', 'type': 'tool_call'}]Tool output:You are an Inventory Management Expert.You have access to the 'inventory_items' table.Table Schema...Agent calling tools: [{'name': 'run_sql_query', 'args': {'query': "SELECT warehouse_location FROM inventory_items WHERE product_name = 'Laptop';"}, 'id': 'call_647ee3a444804bd98a045f00', 'type': 'tool_call'}]Tool output: [('Warehouse A',)]...Agent response: The Laptop is located in **Warehouse A**.## 6. Tam Kaynak Kodu Referansı

Aşağıda, veritabanı başlatma betiği ve Agent ana programı dahil olmak üzere projenin tam kaynak kodu bulunmaktadır.

1. Veritabanı Başlatma (setup_db.py)

`import psycopg2 from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT import os from dotenv import load_dotenv

load_dotenv()

Lütfen .env dosyasında veritabanı bağlantı bilgilerini yapılandırdığınızdan emin olun

DB_HOST = os.getenv("DB_HOST", "localhost") DB_PORT = os.getenv("DB_PORT", "5432") DB_USER = os.getenv("DB_USER", "postgres") DB_PASSWORD = os.getenv("DB_PASSWORD", "your_password") # Lütfen gerçek şifreyle değiştirin DB_NAME = os.getenv("DB_NAME", "agent_platform")

def create_database(): try: # Yeni veritabanı oluşturmak için varsayılan 'postgres' veritabanına bağlanın conn = psycopg2.connect( host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD, dbname="postgres", ) conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT) cur = conn.cursor()

    # Veritabanının var olup olmadığını kontrol edin
    cur.execute(f"SELECT 1 FROM pg_catalog.pg_database WHERE datname = '{DB_NAME}'")
    exists = cur.fetchone()

    if not exists:
        print(f"Veritabanı oluşturuluyor {DB_NAME}...")
        cur.execute(f"CREATE DATABASE {DB_NAME}")
    else:
        print(f"Veritabanı {DB_NAME} zaten mevcut.")

    cur.close()
    conn.close()
except Exception as e:
    print(f"Veritabanı oluşturma hatası: {e}")

def create_tables_and_data(): try: conn = psycopg2.connect( host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD, dbname=DB_NAME, ) cur = conn.cursor()

    # Satış Tablosu Oluştur
    print("sales_data tablosu oluşturuluyor...")
    cur.execute(
        """
        CREATE TABLE IF NOT EXISTS sales_data (
            id SERIAL PRIMARY KEY,
            transaction_date DATE,
            product_id VARCHAR(50),
            amount DECIMAL(10, 2),
            region VARCHAR(50)
        )
        """
    )

    # Envanter Tablosu Oluştur
    print("inventory_items tablosu oluşturuluyor...")
    cur.execute(
        """
        CREATE TABLE IF NOT EXISTS inventory_items (
            id SERIAL PRIMARY KEY,
            product_id VARCHAR(50),
            product_name VARCHAR(100),
            stock_count INTEGER,
            warehouse_location VARCHAR(50)
        )
        """
    )

    # Sahte Veri Ekle
    print("Sahte veri ekleniyor...")
    cur.execute("TRUNCATE sales_data, inventory_items")

    sales_data = [
        ('2023-01-01', 'P001', 100.00, 'North'),
        ('2023-01-02', 'P002', 150.50, 'South'),
        ('2023-01-03', 'P001', 120.00, 'East'),
        ('2023-01-04', 'P003', 200.00, 'West'),
        ('2023-01-05', 'P002', 160.00, 'North'),
    ]
    cur.executemany(
        "INSERT INTO sales_data (transaction_date, product_id, amount, region) VALUES (%s, %s, %s, %s)",
        sales_data,
    )

    inventory_data = [
        ('P001', 'Laptop', 50, 'Warehouse A'),
        ('P002', 'Mouse', 200, 'Warehouse B'),
        ('P003', 'Keyboard', 150, 'Warehouse A'),
        ('P004', 'Monitor', 30, 'Warehouse C'),
    ]
    cur.executemany(
        "INSERT INTO inventory_items (product_id, product_name, stock_count, warehouse_location) VALUES (%s, %s, %s, %s)",
        inventory_data,
    )

    conn.commit()
    cur.close()
    conn.close()
    print("Veritabanı kurulumu tamamlandı.")
except Exception as e:
    print(f"Tabloları ayarlama hatası: {e}")

if name == "main": create_database() create_tables_and_data() `### 2. Ajan Ana Programı (main.py)

import os
from typing import Annotated, Literal, TypedDict, Union, Dict
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage
from langchain_community.utilities import SQLDatabase
from langchain_community.agent_toolkits import SQLDatabaseToolkit
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.prebuilt import ToolNode, tools_condition

load_dotenv()

# --- Configuration ---
BASE_URL = os.getenv("BASIC_MODEL_BASE_URL")
API_KEY = os.getenv("BASIC_MODEL_API_KEY")
MODEL_NAME = os.getenv("BASIC_MODEL_MODEL")
DB_URI = f"postgresql://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}@{os.getenv('DB_HOST')}:{os.getenv('DB_PORT')}/{os.getenv('DB_NAME')}"

# --- Database Setup ---
db = SQLDatabase.from_uri(DB_URI)

# --- Skills Definition ---
SKILLS: Dict[str, Dict[str, str]] = {
    "sales_analytics": {
        "description": "Satış gelirini, trendleri ve bölgesel performansı analiz etmek için kullanışlıdır.",
        "content": """
        Sen bir Satış Analizi Uzmanısın.
        'sales_data' tablosuna erişimin var.
        Tablo Şeması:
        - id: integer (birincil anahtar)
        - transaction_date: date
        - product_id: varchar(50)
        - amount: decimal(10, 2)
        - region: varchar(50)
        Ortak sorgular:
        - Toplam gelir: SUM(amount)
        - Bölgeye göre gelir: GROUP BY region
        - Satış trendi: GROUP BY transaction_date
        """
    },
    "inventory_management": {
        "description": "Stok seviyelerini, ürün konumlarını ve depo yönetimini kontrol etmek için kullanışlıdır.",
        "content": """
        Sen bir Envanter Yönetimi Uzmanısın.
        'inventory_item"
```- id: integer (birincil anahtar)
- product_id: varchar(50)
- product_name: varchar(100)
- stock_count: integer
- warehouse_location: varchar(50)
Ortak sorgular:
- Stoğu kontrol et: WHERE product_name = '...'
- Düşük stok: WHERE stock_count < threshold"""  }}# --- Tools ---@tooldefload_skill(skill_name: str)-> str:"""
  Belirli bir beceri için ayrıntılı istemi ve şemayı yükleyin.
  Mevcut beceriler:
  - sales_analytics: Satış, gelir ve işlem analizi için.
  - inventory_management: Stok, ürün ve depo sorguları için.
  """
  skill = SKILLS.get(skill_name)
  if not skill:
    return f"Hata: '{skill_name}' becerisi bulunamadı. Mevcut beceriler: {list(SKILLS.keys())}"
  return skill["content"]

@tool
def run_sql_query(query: str) -> str:
  """
  Veritabanına karşı bir SQL sorgusu yürütün.
  Şemayı anlamak için yalnızca uygun beceriyi yükledikten SONRA bu aracı kullanın.
  """
  try:
    return db.run(query)
  except Exception as e:
    return f"SQL yürütülürken hata: {e}"

@tool
def list_tables() -> str:
  """Veritabanındaki tüm kullanılabilir tabloları listeleyin."""
  return str(db.get_usable_table_names())

araclar = [load_skill, run_sql_query, list_tables]

# --- Aracı Kurulumu ---
llm = ChatOpenAI(
    base_url=BASE_URL,
    api_key=API_KEY,
    model=MODEL_NAME,
    temperature=0
)

llm_with_tools = llm.bind_tools(araclar)

# --- Grafik Tanımı ---
class AgentState(MessagesState):
  # Gerekirse özel durum ekleyebiliriz, ancak basit sohbet için MessagesState yeterlidir
  pass


def agent_node(state: AgentState):
  messages = state["messages"]
  response = llm_with_tools.invoke(messages)
  return {"messages": [response]}


workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)workflow.add_node("tools", ToolNode(tools))workflow.add_edge(START,"agent")workflow.add_conditional_edges("agent", tools_condition)workflow.add_edge("tools","agent")app = workflow.compile()# --- Ana Yürütme ---if__name__ =="__main__":  system_prompt ="""Sen yardımcı bir SQL Asistanısın.Veritabanı şemaları ve alan bilgisi içeren uzmanlaşmış becerilere sahipsin.Bir kullanıcının sorusunu yanıtlamak için:1. İlgili beceriyi (sales_analytics veya inventory_management) belirleyin.2. şema ve talimatları almak için 'load_skill' aracını kullanın.3. Yüklenen beceriye dayanarak, 'run_sql_query' kullanarak bir SQL sorgusu yazın ve yürütün.4. Sorgu sonuçlarına göre kullanıcının sorusunu yanıtlayın.Tablo adlarını tahmin etmeyin. Her zaman önce beceriyi yükleyin."""  print("SQL Asistanı başlatıldı. Çıkmak için 'quit' yazın.")  print("-"*50)  messages = [SystemMessage(content=system_prompt)]# Önceden ısıtma bağlantı kontrolütry:    print(f"Veritabanına bağlanıldı:{DB_URI.split('@')[-1]}")exceptExceptionase:    print(f"Veritabanı bağlantı uyarısı:{e}")whileTrue:try:      user_input = input("Kullanıcı: ")ifuser_input.lower()in["quit","exit"]:break      messages.append(HumanMessage(content=user_input))# Yürütmeyi yayınlayın      print("Aracı: ", end="", flush=True)      final_response =Noneforeventinapp.stream({"messages": messages}, stream_mode="values"):# 'values' modunda, tam durumu alırız. Sadece son mesajın yeni olup olmadığını görmek istiyoruz.        last_message = event["messages"][-1]# Mesaj geçmişimizi en son durumla güncelleyinpass# Yayın bittikten sonra, son durumda nihai cevap bulunur      final_state = app.invoke({"messages": messages})

 last_msg = final_state["messages"][-1]

if isinstance(last_msg, AIMessage):

 print(last_msg.content)

 messages = final_state["messages"]

# Geçmişi güncelle

 print("-"*50)

except Exception as e:

 print(f"\nHata:{e}")

 break`