Dış API Entegrasyonu: LangChain ile Özel Araç (Tool) Yazma

Yazar:komut.io ekibi
0 Yorum

Bir LLM’i gerçekten güçlü kılan şey, sadece metin üretmesi değil, dış dünyayla etkileşime girebilmesidir. Hava durumu sorduğunda internete bakan, veritabanına yazabilen, hatta şirketinizin iç API’larını çağırabilen bir yapay zeka asistanı hayal edin. İşte LangChain’in “Tool” konsepti tam olarak bunu sağlıyor. Bu yazıda, gerçek dünya senaryolarıyla dış API entegrasyonunu nasıl yapacağınızı, kendi araçlarınızı nasıl yazacağınızı ve bunları bir agent’a nasıl bağlayacağınızı adım adım inceleyeceğiz.

LangChain Tool Nedir ve Neden Önemlidir?

LangChain’de bir Tool, LLM’in çağırabileceği bir Python fonksiyonudur. Agent mimarisinde LLM, hangi aracı ne zaman kullanacağına karar verir, aracı çağırır, sonucu alır ve bu sonucu kullanıcıya anlamlı bir cevap üretmek için kullanır. Bu döngü, ReAct (Reasoning + Acting) paradigması olarak bilinir.

Sysadmin perspektifinden bakarsak, bu yapıyı şöyle düşünebilirsiniz: LLM bir orkestrasyon katmanı, Tool’lar ise gerçek işi yapan scriptler. Tıpkı Ansible playbook’unuzda modüller nasıl belirli işleri yapıyorsa, Tool’lar da LLM’in emirlerini gerçek API çağrılarına dönüştürür.

Önce ortamı hazırlayalım:

pip install langchain langchain-openai langchain-community requests python-dotenv httpx
# .env dosyası oluştur
cat > .env << 'EOF'
OPENAI_API_KEY=sk-...
WEATHER_API_KEY=your_openweathermap_key
GITHUB_TOKEN=ghp_...
INTERNAL_API_URL=http://your-internal-api:8080
EOF

Temel Tool Yazımı: @tool Dekoratörü

LangChain’de tool yazmanın en basit yolu @tool dekoratörüdür. Fonksiyonun docstring’i kritik önem taşır çünkü LLM hangi aracı kullanacağına bu açıklamaya bakarak karar verir. Açıklama ne kadar net olursa, LLM o kadar doğru araç seçimi yapar.

from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents import create_openai_tools_agent, AgentExecutor
from langchain import hub
import requests
import os
from dotenv import load_dotenv

load_dotenv()

@tool
def get_server_status(hostname: str) -> str:
    """
    Belirtilen sunucunun HTTP durum kodunu kontrol eder.
    Sunucunun ayakta olup olmadığını öğrenmek için kullanın.
    hostname parametresi tam URL olmalıdır, örneğin: https://example.com
    """
    try:
        response = requests.get(hostname, timeout=5)
        return f"Sunucu {hostname} - Durum: {response.status_code}, Yanıt süresi: {response.elapsed.total_seconds():.2f}s"
    except requests.exceptions.ConnectionError:
        return f"HATA: {hostname} adresine bağlanılamadı. Sunucu kapalı olabilir."
    except requests.exceptions.Timeout:
        return f"HATA: {hostname} zaman aşımına uğradı (5 saniye)."
    except Exception as e:
        return f"Beklenmeyen hata: {str(e)}"

Bu basit örnekte bile birkaç önemli nokta var. Fonksiyon her zaman string döndürüyor çünkü LLM string üzerinde çalışır. Hata yönetimi fonksiyonun içinde yapılıyor ve exception fırlatmak yerine anlamlı hata mesajı döndürüyoruz. LLM bir exception stack trace’i yorumlayamaz ama “Sunucu kapalı olabilir” metnini anlayabilir.

Gerçek Dünya Senaryosu 1: Hava Durumu API Entegrasyonu

OpenWeatherMap API’sini kullanarak daha kapsamlı bir tool yazalım. Bu örnek, parametre doğrulama ve veri formatlama konularını da gösteriyor:

@tool
def get_weather(city: str) -> str:
    """
    Belirtilen şehir için güncel hava durumu bilgisini getirir.
    Sıcaklık, nem, rüzgar hızı ve genel hava koşullarını döndürür.
    Şehir adını Türkçe veya İngilizce girebilirsiniz.
    """
    api_key = os.getenv("WEATHER_API_KEY")
    if not api_key:
        return "Hata: WEATHER_API_KEY environment variable tanımlanmamış."
    
    base_url = "http://api.openweathermap.org/data/2.5/weather"
    params = {
        "q": city,
        "appid": api_key,
        "units": "metric",
        "lang": "tr"
    }
    
    try:
        response = requests.get(base_url, params=params, timeout=10)
        
        if response.status_code == 404:
            return f"'{city}' şehri bulunamadı. Şehir adını kontrol edin."
        
        if response.status_code == 401:
            return "API anahtarı geçersiz. Lütfen WEATHER_API_KEY değerini kontrol edin."
        
        response.raise_for_status()
        data = response.json()
        
        result = f"""
Şehir: {data['name']}, {data['sys']['country']}
Durum: {data['weather'][0]['description'].capitalize()}
Sıcaklık: {data['main']['temp']}°C (Hissedilen: {data['main']['feels_like']}°C)
Nem: {data['main']['humidity']}%
Rüzgar: {data['wind']['speed']} m/s
Görünürlük: {data.get('visibility', 'N/A')} metre
        """.strip()
        
        return result
        
    except requests.exceptions.Timeout:
        return "Hava durumu servisi yanıt vermedi. Lütfen tekrar deneyin."
    except Exception as e:
        return f"Hava durumu alınamadı: {str(e)}"

Gerçek Dünya Senaryosu 2: GitHub API ile Repository Yönetimi

Bir DevOps aracı yazıyorsunuz ve LLM’in GitHub repository bilgilerine erişmesini istiyorsunuz. Bu senaryo, authentication header yönetimini de gösteriyor:

@tool
def get_github_repo_info(repo_full_name: str) -> str:
    """
    GitHub repository hakkında detaylı bilgi getirir.
    repo_full_name parametresi 'kullanici/repo' formatında olmalıdır.
    Örnek: 'torvalds/linux' veya 'microsoft/vscode'
    Yıldız sayısı, fork sayısı, açık issue'lar ve son commit bilgisini döndürür.
    """
    token = os.getenv("GITHUB_TOKEN")
    headers = {
        "Accept": "application/vnd.github.v3+json"
    }
    if token:
        headers["Authorization"] = f"Bearer {token}"
    
    url = f"https://api.github.com/repos/{repo_full_name}"
    
    try:
        response = requests.get(url, headers=headers, timeout=10)
        
        if response.status_code == 404:
            return f"Repository '{repo_full_name}' bulunamadı veya erişim yetkiniz yok."
        
        if response.status_code == 403:
            remaining = response.headers.get('X-RateLimit-Remaining', 'bilinmiyor')
            return f"GitHub API rate limit aşıldı. Kalan istek hakkı: {remaining}"
        
        response.raise_for_status()
        data = response.json()
        
        # Son commit bilgisini al
        commits_url = f"https://api.github.com/repos/{repo_full_name}/commits?per_page=1"
        commits_response = requests.get(commits_url, headers=headers, timeout=10)
        last_commit = "Alınamadı"
        if commits_response.status_code == 200:
            commits_data = commits_response.json()
            if commits_data:
                last_commit = commits_data[0]['commit']['message'].split('n')[0]
        
        result = f"""
Repository: {data['full_name']}
Açıklama: {data.get('description') or 'Açıklama yok'}
Dil: {data.get('language') or 'Belirtilmemiş'}
Yıldız: {data['stargazers_count']:,}
Fork: {data['forks_count']:,}
Açık Issue: {data['open_issues_count']:,}
Varsayılan Branch: {data['default_branch']}
Son Commit: {last_commit}
Boyut: {data['size']} KB
Lisans: {data['license']['name'] if data.get('license') else 'Lisans belirtilmemiş'}
        """.strip()
        
        return result
        
    except Exception as e:
        return f"GitHub bilgisi alınamadı: {str(e)}"

Pydantic ile Güçlü Tip Kontrolü

Karmaşık parametreler için BaseTool ve Pydantic şemalarını kullanmak çok daha sağlam bir yapı sunar. Özellikle birden fazla parametre alan ve bu parametrelerin doğrulanması gereken durumlar için idealdir:

from langchain.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Optional, Type

class ServerHealthCheckInput(BaseModel):
    hostname: str = Field(description="Kontrol edilecek sunucunun hostname veya IP adresi")
    port: int = Field(default=80, description="Bağlantı portu, varsayılan 80")
    protocol: str = Field(default="http", description="Protokol: http veya https")
    timeout: int = Field(default=5, description="Bekleme süresi saniye cinsinden, max 30")

class ServerHealthCheckTool(BaseTool):
    name: str = "server_health_check"
    description: str = """
    Sunucu sağlık kontrolü yapar. HTTP/HTTPS endpoint'lerini kontrol eder.
    Durum kodu, yanıt süresi ve bağlantı durumu hakkında bilgi verir.
    Monitoring ve troubleshooting görevleri için kullanın.
    """
    args_schema: Type[BaseModel] = ServerHealthCheckInput
    
    def _run(self, hostname: str, port: int = 80, 
             protocol: str = "http", timeout: int = 5) -> str:
        # Timeout sınırlaması
        timeout = min(timeout, 30)
        
        url = f"{protocol}://{hostname}:{port}"
        if (protocol == "http" and port == 80) or 
           (protocol == "https" and port == 443):
            url = f"{protocol}://{hostname}"
        
        try:
            response = requests.get(url, timeout=timeout, 
                                   allow_redirects=True)
            status = "SAĞLIKLI" if response.status_code < 400 else "SORUNLU"
            
            return (f"[{status}] {url} - "
                   f"HTTP {response.status_code} - "
                   f"Yanıt: {response.elapsed.total_seconds():.3f}s - "
                   f"İçerik boyutu: {len(response.content)} byte")
                   
        except requests.exceptions.SSLError:
            return f"SSL sertifika hatası: {url}"
        except requests.exceptions.ConnectionError:
            return f"[KAPALI] {url} - Bağlantı reddedildi veya host bulunamadı"
        except requests.exceptions.Timeout:
            return f"[ZAMAN AŞIMI] {url} - {timeout} saniyede yanıt alınamadı"

Şirket İçi API Entegrasyonu

Gerçek sysadmin senaryolarında en çok ihtiyaç duyulan durum, şirket içi sistemlere erişimdir. CMDB (Configuration Management Database) veya bir monitoring API’sini sorgulayan bir tool yazalım:

@tool
def query_cmdb(hostname: str) -> str:
    """
    Şirket CMDB sisteminden sunucu bilgilerini sorgular.
    Sunucu sahibi, lokasyon, işletim sistemi, servis durumu gibi bilgileri döndürür.
    IT operasyon görevleri için kullanın.
    """
    api_url = os.getenv("INTERNAL_API_URL", "http://cmdb-api:8080")
    api_token = os.getenv("INTERNAL_API_TOKEN")
    
    headers = {
        "Content-Type": "application/json",
        "X-API-Token": api_token
    }
    
    try:
        # CMDB'den sunucu bilgisi çek
        response = requests.get(
            f"{api_url}/api/v1/servers/{hostname}",
            headers=headers,
            timeout=10,
            verify=True  # SSL doğrulama açık, internal CA için verify="/etc/ssl/company-ca.crt" yapın
        )
        
        if response.status_code == 404:
            return f"'{hostname}' CMDB'de kayıtlı değil. Kayıt için IT ekibine başvurun."
        
        response.raise_for_status()
        data = response.json()
        
        # Maintenance penceresi kontrolü
        in_maintenance = data.get('maintenance_mode', False)
        maintenance_note = " [BAKIM MODUNDA]" if in_maintenance else ""
        
        return f"""
Hostname: {data['hostname']}{maintenance_note}
IP Adresi: {data['ip_address']}
İşletim Sistemi: {data['os']} {data['os_version']}
Lokasyon: {data['datacenter']} / {data['rack']}
Sahip Ekip: {data['owner_team']}
Servis Katmanı: {data['service_tier']}
Son Yama Tarihi: {data['last_patched']}
Monitoring Durumu: {data['monitoring_status']}
        """.strip()
        
    except requests.exceptions.ConnectionError:
        return f"CMDB API'sine bağlanılamadı ({api_url}). VPN bağlantınızı kontrol edin."
    except Exception as e:
        return f"CMDB sorgusu başarısız: {str(e)}"

Agent Oluşturma ve Tool’ları Bağlama

Tüm tool’larımızı bir agent’a bağlayalım. Bu noktada LLM, hangi soruya hangi tool’u kullanacağına kendisi karar verecek:

from langchain_openai import ChatOpenAI
from langchain.agents import create_openai_tools_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# LLM tanımla
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0,  # Araç seçimi için deterministik davranış
    api_key=os.getenv("OPENAI_API_KEY")
)

# Tool listesi
tools = [
    get_server_status,
    get_weather,
    get_github_repo_info,
    ServerHealthCheckTool(),
    query_cmdb
]

# Sistem promptu
prompt = ChatPromptTemplate.from_messages([
    ("system", """Sen deneyimli bir sistem yöneticisi asistanısın. 
    Kullanıcıların IT operasyon sorularını yanıtlamak için gerekli araçları kullanırsın.
    Her zaman Türkçe yanıt ver. Teknik bilgileri net ve anlaşılır şekilde açıkla.
    Emin olmadığın durumlarda araçları kullanarak bilgiyi doğrula."""),
    MessagesPlaceholder(variable_name="chat_history", optional=True),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad")
])

# Agent oluştur
agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,       # Debug için tool çağrılarını göster
    max_iterations=5,   # Sonsuz döngüyü önle
    return_intermediate_steps=True  # Hangi tool'ların kullanıldığını göster
)

# Test et
if __name__ == "__main__":
    sorular = [
        "İstanbul'un hava durumu nasıl?",
        "microsoft/vscode repository'si hakkında bilgi ver",
        "web01.internal.company.com sunucusunu CMDB'de kontrol et"
    ]
    
    for soru in sorular:
        print(f"n{'='*50}")
        print(f"Soru: {soru}")
        print('='*50)
        result = agent_executor.invoke({"input": soru})
        print(f"Yanıt: {result['output']}")

Hata Yönetimi ve Retry Mekanizması

Production ortamında tool’larınızın güvenilir olması şarttır. Retry mekanizması ve circuit breaker pattern’i ekleyelim:

import time
from functools import wraps

def retry_on_failure(max_retries: int = 3, delay: float = 1.0):
    """Tool fonksiyonları için retry dekoratörü"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except requests.exceptions.Timeout as e:
                    last_error = e
                    if attempt < max_retries - 1:
                        wait_time = delay * (2 ** attempt)  # Exponential backoff
                        time.sleep(wait_time)
                except requests.exceptions.ConnectionError as e:
                    last_error = e
                    if attempt < max_retries - 1:
                        time.sleep(delay)
                except Exception as e:
                    # Retry edilemeyecek hatalar için direkt dön
                    return f"Kritik hata: {str(e)}"
            
            return f"Servis {max_retries} denemeden sonra yanıt vermedi: {str(last_error)}"
        return wrapper
    return decorator

@tool
@retry_on_failure(max_retries=3, delay=2.0)
def get_prometheus_metrics(job_name: str) -> str:
    """
    Prometheus'tan belirli bir job için anlık metrikleri çeker.
    CPU kullanımı, bellek kullanımı ve özel uygulama metriklerini döndürür.
    job_name parametresi Prometheus'taki job etiketi değeridir.
    """
    prometheus_url = os.getenv("PROMETHEUS_URL", "http://prometheus:9090")
    
    queries = {
        "CPU Kullanımı": f'100 - (avg by(instance)(rate(node_cpu_seconds_total{{mode="idle",job="{job_name}"}}[5m])) * 100)',
        "Bellek Kullanımı": f'(1 - (node_memory_MemAvailable_bytes{{job="{job_name}"}} / node_memory_MemTotal_bytes{{job="{job_name}"}})) * 100',
        "Disk I/O": f'rate(node_disk_io_time_seconds_total{{job="{job_name}"}}[5m])'
    }
    
    results = []
    for metric_name, query in queries.items():
        response = requests.get(
            f"{prometheus_url}/api/v1/query",
            params={"query": query},
            timeout=10
        )
        response.raise_for_status()
        data = response.json()
        
        if data['status'] == 'success' and data['data']['result']:
            value = float(data['data']['result'][0]['value'][1])
            results.append(f"{metric_name}: {value:.2f}%")
        else:
            results.append(f"{metric_name}: Veri yok")
    
    return f"Job '{job_name}' Metrikleri:n" + "n".join(results)

Tool Çıktısını Yapılandırma ve Loglama

Production ortamında tool çağrılarını loglamak ve izlemek kritik önem taşır:

import logging
import json
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('/var/log/langchain-tools.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger('langchain_tools')

def log_tool_call(tool_name: str, params: dict, result: str, duration: float):
    """Tool çağrılarını structured log olarak kaydet"""
    log_entry = {
        "timestamp": datetime.utcnow().isoformat(),
        "tool": tool_name,
        "parameters": params,
        "result_length": len(result),
        "duration_ms": round(duration * 1000, 2),
        "success": not result.startswith("HATA") and not result.startswith("Kritik")
    }
    logger.info(json.dumps(log_entry, ensure_ascii=False))

@tool
def get_disk_usage(server: str, mount_point: str = "/") -> str:
    """
    Belirtilen sunucudaki disk kullanım bilgisini getirir.
    server parametresi CMDB'deki hostname olmalıdır.
    mount_point varsayılan olarak kök dizini kontrol eder.
    Yüksek disk kullanımı uyarısı için kullanın.
    """
    start_time = time.time()
    api_url = os.getenv("MONITORING_API_URL", "http://monitoring:8080")
    
    try:
        response = requests.get(
            f"{api_url}/api/disk/{server}",
            params={"mount": mount_point},
            timeout=10
        )
        response.raise_for_status()
        data = response.json()
        
        used_percent = data['used_percent']
        
        # Kritiklik seviyesi belirle
        if used_percent >= 90:
            severity = "KRİTİK"
        elif used_percent >= 75:
            severity = "UYARI"
        else:
            severity = "NORMAL"
        
        result = (f"[{severity}] {server}:{mount_point} - "
                 f"Kullanım: {used_percent}% "
                 f"({data['used_gb']:.1f}GB / {data['total_gb']:.1f}GB)")
        
        duration = time.time() - start_time
        log_tool_call("get_disk_usage", 
                      {"server": server, "mount_point": mount_point},
                      result, duration)
        return result
        
    except Exception as e:
        error_msg = f"Disk bilgisi alınamadı ({server}): {str(e)}"
        logger.error(error_msg)
        return error_msg

Async Tool Desteği

Çok sayıda API çağrısı yapıyorsanız async kullanmak performansı ciddi artırır:

import asyncio
import httpx
from langchain.tools import StructuredTool

async def _check_multiple_servers_async(hostnames: list) -> dict:
    """Birden fazla sunucuyu eş zamanlı kontrol et"""
    async with httpx.AsyncClient(timeout=10.0) as client:
        tasks = []
        for hostname in hostnames:
            tasks.append(client.get(f"https://{hostname}"))
        
        results = {}
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        
        for hostname, response in zip(hostnames, responses):
            if isinstance(response, Exception):
                results[hostname] = f"HATA: {str(response)}"
            else:
                results[hostname] = f"HTTP {response.status_code} ({response.elapsed.total_seconds():.2f}s)"
        
        return results

def check_multiple_servers(hostnames_csv: str) -> str:
    """
    Virgülle ayrılmış sunucu listesini eş zamanlı olarak kontrol eder.
    Toplu sağlık kontrolü için kullanın. Tüm sunucular paralel olarak sorgulanır.
    Örnek: 'web01.example.com,web02.example.com,api.example.com'
    """
    hostnames = [h.strip() for h in hostnames_csv.split(",")]
    
    if len(hostnames) > 20:
        return "En fazla 20 sunucu aynı anda kontrol edilebilir."
    
    results = asyncio.run(_check_multiple_servers_async(hostnames))
    
    output_lines = [f"Toplam {len(hostnames)} sunucu kontrol edildi:n"]
    for hostname, status in results.items():
        output_lines.append(f"- {hostname}: {status}")
    
    return "n".join(output_lines)

# StructuredTool olarak kaydet
multi_server_check_tool = StructuredTool.from_function(
    func=check_multiple_servers,
    name="check_multiple_servers",
    description="Birden fazla sunucuyu aynı anda kontrol eder. Toplu sağlık kontrolü için idealdir."
)

Güvenlik Konuları

Tool’larla dış API’lara erişirken güvenlik göz ardı edilemez:

  • API anahtarlarını asla kod içine yazmayın, her zaman environment variable veya vault sistemi kullanın (HashiCorp Vault, AWS Secrets Manager)
  • Input sanitization yapın: LLM’den gelen parametreler doğrudan SQL sorgusuna veya shell komutuna geçirilmemeli
  • Rate limiting uygulayın: LLM bir döngüye girip saniyede yüzlerce API çağrısı yapabilir
  • Timeout değerlerini daima belirtin, varsayılana bırakmayın
  • Sadece okuma izni olan API tokenları kullanın, mümkünse yazma işlemleri için ayrı onay mekanizması ekleyin
  • SSL doğrulamasını devre dışı bırakmayın (verify=False), iç CA sertifikalarınızı sisteme tanıtın
  • Hassas bilgileri tool çıktısından filtreleyin, parolalar veya tam bağlantı stringleri LLM’e iletilmemeli

Sonuç

LangChain tool sistemi, LLM’leri gerçek dünya sistemleriyle entegre etmek için son derece esnek bir yapı sunuyor. Basit @tool dekoratöründen başlayıp Pydantic şemaları, async destek ve retry mekanizmalarına kadar ölçeklenebilen bu yapı, kurumsal kullanım için de oldukça uygun.

Dikkat etmeniz gereken en kritik noktalar şunlar: Tool açıklamalarını (docstring) çok iyi yazın, çünkü LLM’in doğru araç seçimi tamamen bu metne bağlı. Hata yönetimini her zaman fonksiyon içinde yapın ve anlamlı mesajlar döndürün. Production’da tüm tool çağrılarını loglayın, hem debug hem de audit açısından bu veriler değerli. Son olarak güvenliği sonradan eklemeye çalışmayın, başından itibaren tasarıma dahil edin.

Bu temel üzerine ilerleyen yazılarda LangGraph ile multi-agent sistemler, tool sonuçlarını hafızaya alma ve RAG entegrasyonu konularını ele alacağız.

Bir yanıt yazın

E-posta adresiniz yayınlanmayacak. Gerekli alanlar * ile işaretlenmişlerdir