Function Calling ile OpenAI API’yi Dış Araçlara Bağlama

Yazar:komut.io ekibi
0 Yorum

Yapay zeka modelleriyle çalışırken bir noktada şunu fark ediyorsunuz: ChatGPT veya GPT-4 harika sorular cevaplayabiliyor, ama kendi veritabanınızdaki müşteri bilgisine erişemiyor, sisteminizdeki servisleri kontrol edemiyor ya da anlık hava durumu verisi çekemiyor. İşte tam burada Function Calling devreye giriyor. OpenAI’nin bu özelliği, dil modelini gerçek dünya araçlarına bağlamanın resmi ve güvenilir yolu. Bu yazıda, Function Calling’in nasıl çalıştığını, nasıl implemente edileceğini ve production ortamında nasıl kullanılabileceğini elimden geldiğince pratik bir şekilde anlatacağım.

Function Calling Nedir ve Neden Önemli?

Normalde bir LLM’e soru sorduğunuzda, model eğitim verisine dayanarak bir metin cevabı üretir. Bu cevap bazen yanlış olabilir, bazen güncel olmayabilir, bazen de sisteminizle hiçbir ilgisi olmayabilir. Function Calling ise modele şunu söylemenizi sağlar: “Sen bu fonksiyonları çağırabilirsin, ihtiyaç duyduğunda kullan.”

Model aslında fonksiyonu direkt çalıştırmıyor, bunu anlamak çok önemli. Model size yapılandırılmış bir JSON çıktısı döndürüyor: “Şu fonksiyonu, şu parametrelerle çağırmak istiyorum.” Sonra siz bu çağrıyı yapıyor, sonucu modele geri veriyorsunuz ve model nihai cevabını oluşturuyor. Bu döngüsel yapı, modeli kendi araçlarınıza entegre etmenin temel mekanizması.

Neden önemli? Şunlar için:

  • Gerçek zamanlı veri: Veritabanı sorguları, API çağrıları, servis kontrolleri
  • Güvenilir yapılandırılmış çıktı: JSON parse etmek için regex yazmak zorunda kalmıyorsunuz
  • Aksiyon alma: Sunucuda komut çalıştırmak, ticket açmak, bildirim göndermek
  • Hibrit sistemler: LLM zekasını mevcut altyapınızla birleştirmek

Temel Yapı ve İlk Örnek

Önce gerekli kurulumları yapalım. OpenAI Python kütüphanesinin güncel versiyonunu kullanıyoruz:

pip install openai python-dotenv

Basit bir ortam değişkeni dosyası oluşturalım:

cat > .env << 'EOF'
OPENAI_API_KEY=sk-your-api-key-here
EOF

Şimdi en temel Function Calling örneğine bakalım. Bir sunucunun CPU kullanımını sorgulayan basit bir senaryo:

cat > basic_function_calling.py << 'EOF'
import os
import json
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# Fonksiyon tanımları - modele hangi araçlara sahip olduğumuzu söylüyoruz
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_server_cpu_usage",
            "description": "Belirtilen sunucunun anlık CPU kullanım yüzdesini döndürür",
            "parameters": {
                "type": "object",
                "properties": {
                    "server_name": {
                        "type": "string",
                        "description": "Sunucu adı veya IP adresi"
                    },
                    "interval": {
                        "type": "integer",
                        "description": "Ölçüm aralığı saniye cinsinden, varsayılan 1",
                        "default": 1
                    }
                },
                "required": ["server_name"]
            }
        }
    }
]

# Gerçek fonksiyon implementasyonu (normalde burada psutil veya SSH bağlantısı olurdu)
def get_server_cpu_usage(server_name: str, interval: int = 1) -> dict:
    # Demo amaçlı sahte veri döndürüyoruz
    import random
    return {
        "server": server_name,
        "cpu_usage": round(random.uniform(10, 95), 2),
        "interval": interval,
        "unit": "percent"
    }

# Konuşmayı başlatıyoruz
messages = [
    {"role": "user", "content": "web-server-01 sunucusunun CPU kullanımı ne kadar?"}
]

# İlk API çağrısı
response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools,
    tool_choice="auto"
)

response_message = response.choices[0].message

# Model bir fonksiyon çağırmak istiyor mu kontrol ediyoruz
if response_message.tool_calls:
    tool_call = response_message.tool_calls[0]
    function_name = tool_call.function.name
    function_args = json.loads(tool_call.function.arguments)
    
    print(f"Model şu fonksiyonu çağırmak istiyor: {function_name}")
    print(f"Parametreler: {function_args}")
    
    # Fonksiyonu çalıştırıyoruz
    result = get_server_cpu_usage(**function_args)
    
    # Sonucu konuşmaya ekliyoruz
    messages.append(response_message)
    messages.append({
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": json.dumps(result)
    })
    
    # Model nihai cevabını oluştursun
    final_response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools
    )
    
    print(f"nAsistan: {final_response.choices[0].message.content}")

EOF
python basic_function_calling.py

Bu temel akışı anladıktan sonra daha karmaşık senaryolara geçebiliriz.

Çoklu Fonksiyon Tanımlama

Gerçek dünya uygulamalarında tek bir fonksiyonla iş bitmez. Bir sysadmin asistanı düşünelim: sunucu durumu, disk kullanımı, aktif servisler, log analizi gibi pek çok aracın olması gerekir.

cat > sysadmin_assistant.py << 'EOF'
import os
import json
import subprocess
import psutil
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_disk_usage",
            "description": "Sistemdeki disk kullanım bilgilerini döndürür",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Kontrol edilecek dizin yolu, varsayılan /",
                        "default": "/"
                    }
                },
                "required": []
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "check_service_status",
            "description": "Belirtilen Linux servisinin çalışıp çalışmadığını kontrol eder",
            "parameters": {
                "type": "object",
                "properties": {
                    "service_name": {
                        "type": "string",
                        "description": "Kontrol edilecek servis adı (örn: nginx, mysql, docker)"
                    }
                },
                "required": ["service_name"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_memory_info",
            "description": "Sistem bellek kullanım bilgilerini döndürür",
            "parameters": {
                "type": "object",
                "properties": {},
                "required": []
            }
        }
    }
]

def get_disk_usage(path: str = "/") -> dict:
    usage = psutil.disk_usage(path)
    return {
        "path": path,
        "total_gb": round(usage.total / (1024**3), 2),
        "used_gb": round(usage.used / (1024**3), 2),
        "free_gb": round(usage.free / (1024**3), 2),
        "percent": usage.percent
    }

def check_service_status(service_name: str) -> dict:
    try:
        result = subprocess.run(
            ["systemctl", "is-active", service_name],
            capture_output=True, text=True, timeout=5
        )
        status = result.stdout.strip()
        return {
            "service": service_name,
            "status": status,
            "is_running": status == "active"
        }
    except subprocess.TimeoutExpired:
        return {"service": service_name, "status": "timeout", "is_running": False}
    except Exception as e:
        return {"service": service_name, "status": "error", "error": str(e)}

def get_memory_info() -> dict:
    mem = psutil.virtual_memory()
    return {
        "total_gb": round(mem.total / (1024**3), 2),
        "available_gb": round(mem.available / (1024**3), 2),
        "used_gb": round(mem.used / (1024**3), 2),
        "percent": mem.percent
    }

# Fonksiyon dispatcher
function_map = {
    "get_disk_usage": get_disk_usage,
    "check_service_status": check_service_status,
    "get_memory_info": get_memory_info
}

def run_conversation(user_message: str):
    messages = [
        {
            "role": "system",
            "content": "Sen bir sysadmin asistanısın. Sistem bilgilerini sorgulayabilirsin. Türkçe cevap ver."
        },
        {"role": "user", "content": user_message}
    ]
    
    # Model birden fazla fonksiyon çağırabilir, döngüyle takip ediyoruz
    while True:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )
        
        msg = response.choices[0].message
        messages.append(msg)
        
        # Fonksiyon çağrısı yoksa bitiyoruz
        if not msg.tool_calls:
            return msg.content
        
        # Tüm fonksiyon çağrılarını işliyoruz
        for tool_call in msg.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)
            
            if func_name in function_map:
                result = function_map[func_name](**func_args)
            else:
                result = {"error": f"Bilinmeyen fonksiyon: {func_name}"}
            
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result, ensure_ascii=False)
            })

# Test
print(run_conversation("Nginx servisi çalışıyor mu ve disk doluluk durumu ne?"))
EOF

Paralel Fonksiyon Çağrısı

GPT-4o ve GPT-4 Turbo, birden fazla fonksiyonu aynı anda çağırabiliyor. Bu özelliği kullanmak performansı ciddi ölçüde artırır:

cat > parallel_calls.py << 'EOF'
import os
import json
import concurrent.futures
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def process_parallel_tool_calls(msg, function_map):
    """Birden fazla tool call'u paralel olarak işler"""
    results = {}
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        future_to_call = {}
        
        for tool_call in msg.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)
            
            if func_name in function_map:
                future = executor.submit(function_map[func_name], **func_args)
                future_to_call[future] = tool_call
        
        for future in concurrent.futures.as_completed(future_to_call):
            tool_call = future_to_call[future]
            try:
                results[tool_call.id] = future.result()
            except Exception as e:
                results[tool_call.id] = {"error": str(e)}
    
    return results

# Bu yapı sayesinde model "hem nginx'i hem mysql'i hem de disk durumunu kontrol et"
# dediğinde, üç sorgu sırayla değil paralel yapılır
EOF

Gerçek Dünya Senaryosu: Monitoring Chatbot

Şimdi production’a yakın bir senaryo yazalım. Prometheus veya benzeri bir monitoring sistemine bağlanan, alert sorgulayabilen ve ticket açabilen bir chatbot:

cat > monitoring_chatbot.py << 'EOF'
import os
import json
import requests
from datetime import datetime, timedelta
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

PROMETHEUS_URL = os.getenv("PROMETHEUS_URL", "http://localhost:9090")
JIRA_URL = os.getenv("JIRA_URL", "http://localhost:8080")
JIRA_TOKEN = os.getenv("JIRA_TOKEN", "")

tools = [
    {
        "type": "function",
        "function": {
            "name": "query_prometheus",
            "description": "Prometheus'a PromQL sorgusu gönderir ve metrik verisi döndürür",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "PromQL sorgusu, örn: up{job='node'} veya rate(http_requests_total[5m])"
                    },
                    "time_range_minutes": {
                        "type": "integer",
                        "description": "Kaç dakika geriye bakılacak, varsayılan 30"
                    }
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_active_alerts",
            "description": "Şu anda aktif olan tüm alertleri listeler",
            "parameters": {
                "type": "object",
                "properties": {
                    "severity": {
                        "type": "string",
                        "enum": ["critical", "warning", "info", "all"],
                        "description": "Alert ciddiyet seviyesi filtresi"
                    }
                },
                "required": []
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "create_incident_ticket",
            "description": "Jira'da yeni bir incident ticket oluşturur",
            "parameters": {
                "type": "object",
                "properties": {
                    "title": {
                        "type": "string",
                        "description": "Ticket başlığı"
                    },
                    "description": {
                        "type": "string",
                        "description": "Sorunun detaylı açıklaması"
                    },
                    "priority": {
                        "type": "string",
                        "enum": ["Highest", "High", "Medium", "Low"],
                        "description": "Ticket önceliği"
                    },
                    "affected_service": {
                        "type": "string",
                        "description": "Etkilenen servis adı"
                    }
                },
                "required": ["title", "description", "priority"]
            }
        }
    }
]

def query_prometheus(query: str, time_range_minutes: int = 30) -> dict:
    try:
        end_time = datetime.now()
        start_time = end_time - timedelta(minutes=time_range_minutes)
        
        params = {
            "query": query,
            "start": start_time.timestamp(),
            "end": end_time.timestamp(),
            "step": "60s"
        }
        
        response = requests.get(
            f"{PROMETHEUS_URL}/api/v1/query_range",
            params=params,
            timeout=10
        )
        
        if response.status_code == 200:
            data = response.json()
            return {"status": "success", "data": data.get("data", {})}
        else:
            return {"status": "error", "message": f"HTTP {response.status_code}"}
    except requests.exceptions.ConnectionError:
        # Demo modda sahte veri
        return {
            "status": "success",
            "demo_mode": True,
            "data": {
                "resultType": "matrix",
                "result": [
                    {
                        "metric": {"__name__": "up", "job": "node", "instance": "web-01:9100"},
                        "values": [[datetime.now().timestamp(), "1"]]
                    }
                ]
            }
        }

def get_active_alerts(severity: str = "all") -> dict:
    # Demo veri
    alerts = [
        {
            "name": "HighCPUUsage",
            "severity": "warning",
            "instance": "db-server-01",
            "value": "87.3%",
            "started": "2024-01-15T10:23:00Z"
        },
        {
            "name": "DiskSpaceLow",
            "severity": "critical",
            "instance": "storage-01",
            "value": "94.1% dolu",
            "started": "2024-01-15T09:45:00Z"
        }
    ]
    
    if severity != "all":
        alerts = [a for a in alerts if a["severity"] == severity]
    
    return {
        "alert_count": len(alerts),
        "alerts": alerts
    }

def create_incident_ticket(title: str, description: str, priority: str, 
                           affected_service: str = "unknown") -> dict:
    # Gerçek implementasyonda Jira API çağrısı yapılır
    ticket_id = f"INC-{datetime.now().strftime('%Y%m%d%H%M%S')}"
    
    print(f"n[TICKET OLUŞTURULUYOR]nID: {ticket_id}nBaşlık: {title}nÖncelik: {priority}")
    
    return {
        "ticket_id": ticket_id,
        "status": "created",
        "url": f"{JIRA_URL}/browse/{ticket_id}",
        "title": title,
        "priority": priority,
        "affected_service": affected_service
    }

function_map = {
    "query_prometheus": query_prometheus,
    "get_active_alerts": get_active_alerts,
    "create_incident_ticket": create_incident_ticket
}

def chat(user_input: str, conversation_history: list = None) -> tuple:
    if conversation_history is None:
        conversation_history = [
            {
                "role": "system",
                "content": """Sen bir DevOps/SRE asistanısın. Prometheus metrikleri sorgulayabilir, 
                aktif alertleri görebilir ve Jira'da incident ticket oluşturabilirsin. 
                Türkçe cevap ver. Kritik durumları tespit ettiğinde proaktif olarak ticket açmayı öner."""
            }
        ]
    
    conversation_history.append({"role": "user", "content": user_input})
    
    while True:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=conversation_history,
            tools=tools,
            tool_choice="auto"
        )
        
        msg = response.choices[0].message
        conversation_history.append(msg)
        
        if not msg.tool_calls:
            return msg.content, conversation_history
        
        for tool_call in msg.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)
            result = function_map.get(func_name, lambda **k: {"error": "Fonksiyon bulunamadı"})(**func_args)
            
            conversation_history.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result, ensure_ascii=False)
            })

# Test konuşması
history = None
sorular = [
    "Şu anda kritik alert var mı?",
    "Storage-01 için hemen ticket aç, çünkü disk dolmak üzere"
]

for soru in sorular:
    print(f"nKullanici: {soru}")
    cevap, history = chat(soru, history)
    print(f"Asistan: {cevap}")
EOF

Hata Yönetimi ve Güvenlik

Production’da dikkat edilmesi gereken en önemli konulardan biri güvenlik. Model size kötü parametreler gönderebilir, fonksiyon çağrısı başarısız olabilir:

cat > secure_function_calling.py << 'EOF'
import os
import json
import logging
from functools import wraps
from openai import OpenAI

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# İzin verilen komutlar whitelist
ALLOWED_COMMANDS = {
    "systemctl": ["status", "is-active", "is-enabled"],
    "df": ["-h", "-k"],
    "free": ["-h", "-m"]
}

def validate_function_call(func_name: str, func_args: dict, allowed_functions: list) -> bool:
    """Fonksiyon çağrısının güvenlik kontrolü"""
    if func_name not in allowed_functions:
        logger.warning(f"İzinsiz fonksiyon çağrısı girişimi: {func_name}")
        return False
    
    # Tehlikeli parametre kontrolü
    args_str = json.dumps(func_args)
    dangerous_patterns = [";", "&&", "||", "`", "$(",  "../", "rm ", "dd "]
    
    for pattern in dangerous_patterns:
        if pattern in args_str:
            logger.error(f"Tehlikeli parametre tespit edildi: {pattern}")
            return False
    
    return True

def safe_tool_executor(func):
    """Fonksiyon çağrılarını güvenli şekilde çalıştıran decorator"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            result = func(*args, **kwargs)
            logger.info(f"Fonksiyon başarıyla çalıştı: {func.__name__}")
            return result
        except PermissionError as e:
            logger.error(f"İzin hatası: {e}")
            return {"error": "Bu işlem için yetkiniz yok", "type": "permission_error"}
        except TimeoutError as e:
            logger.error(f"Timeout hatası: {e}")
            return {"error": "İşlem zaman aşımına uğradı", "type": "timeout_error"}
        except Exception as e:
            logger.error(f"Beklenmeyen hata {func.__name__}: {e}")
            return {"error": f"İşlem başarısız: {str(e)}", "type": "general_error"}
    return wrapper

# Rate limiting için basit bir implementasyon
from collections import defaultdict
from time import time

call_counts = defaultdict(list)

def rate_limit_check(func_name: str, max_calls: int = 10, window_seconds: int = 60) -> bool:
    """Dakika başına maksimum çağrı sayısını kontrol eder"""
    now = time()
    window_start = now - window_seconds
    
    # Eski kayıtları temizle
    call_counts[func_name] = [t for t in call_counts[func_name] if t > window_start]
    
    if len(call_counts[func_name]) >= max_calls:
        logger.warning(f"Rate limit aşıldı: {func_name}")
        return False
    
    call_counts[func_name].append(now)
    return True

EOF

Streaming ile Real-Time Yanıtlar

Uzun işlemler için streaming kullanmak kullanıcı deneyimini iyileştirir:

cat > streaming_with_functions.py << 'EOF'
import os
import json
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def stream_with_function_calling(user_message: str, tools: list, function_map: dict):
    messages = [{"role": "user", "content": user_message}]
    
    # İlk streaming yanıt (fonksiyon çağrısını yakala)
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools,
        tool_choice="auto",
        stream=True
    )
    
    tool_calls = []
    current_tool_call = None
    final_content = ""
    
    for chunk in stream:
        delta = chunk.choices[0].delta if chunk.choices else None
        if not delta:
            continue
        
        # Normal metin akışı
        if delta.content:
            print(delta.content, end="", flush=True)
            final_content += delta.content
        
        # Fonksiyon çağrısı akışı
        if delta.tool_calls:
            for tc_chunk in delta.tool_calls:
                if tc_chunk.index is not None:
                    # Yeni tool call başlıyor
                    while len(tool_calls) <= tc_chunk.index:
                        tool_calls.append({
                            "id": "",
                            "type": "function",
                            "function": {"name": "", "arguments": ""}
                        })
                    
                    if tc_chunk.id:
                        tool_calls[tc_chunk.index]["id"] = tc_chunk.id
                    if tc_chunk.function:
                        if tc_chunk.function.name:
                            tool_calls[tc_chunk.index]["function"]["name"] += tc_chunk.function.name
                        if tc_chunk.function.arguments:
                            tool_calls[tc_chunk.index]["function"]["arguments"] += tc_chunk.function.arguments
    
    print()  # Yeni satır
    
    # Fonksiyon çağrıları varsa işle
    if tool_calls:
        messages.append({
            "role": "assistant",
            "content": final_content or None,
            "tool_calls": tool_calls
        })
        
        for tc in tool_calls:
            func_name = tc["function"]["name"]
            func_args = json.loads(tc["function"]["arguments"])
            
            print(f"n[{func_name} çalıştırılıyor...]")
            result = function_map.get(func_name, lambda **k: {})(** func_args)
            
            messages.append({
                "role": "tool",
                "tool_call_id": tc["id"],
                "content": json.dumps(result, ensure_ascii=False)
            })
        
        # Final streaming yanıt
        print("nSonuç: ", end="")
        final_stream = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            stream=True
        )
        
        for chunk in final_stream:
            if chunk.choices and chunk.choices[0].delta.content:
                print(chunk.choices[0].delta.content, end="", flush=True)
        print()

EOF

Production’da Function Calling: İpuçları ve Dikkat Edilecekler

Birkaç ayda bu yapıyı gerçek projelerde kullandıktan sonra öğrendiğim dersleri paylaşayım.

Fonksiyon açıklamalarını ciddiye alın. Model, hangi fonksiyonu ne zaman kullanacağına description’a bakarak karar veriyor. “Disk bilgisi döndürür” yerine “Belirtilen dizin veya tüm mount noktaları için disk kullanım yüzdesi, toplam alan ve boş alan bilgisini döndürür” gibi detaylı yazın.

Tool choice parametresini akıllıca kullanın. Şu seçenekler var:

  • “auto”: Model istediği zaman fonksiyon çağırabilir
  • “required”: Model mutlaka bir fonksiyon çağırmak zorunda
  • {“type”: “function”, “function”: {“name”: “fonksiyon_adi”}}: Belirli bir fonksiyon mutlaka çağrılmalı
  • “none”: Fonksiyon çağrısı yapılamaz, sadece metin yanıtı

Token maliyetine dikkat edin. Her fonksiyon tanımı token tüketiyor. 20 fonksiyon tanımlarsanız ve bunların yarısı nadiren kullanılıyorsa, kullanıcının sorusuna göre dinamik olarak tool listesi oluşturmayı düşünün.

Async yapı kullanın. Production uygulamalarda asyncio ile aiohttp kombinasyonu, özellikle paralel tool call’larda büyük performans farkı yaratır:

cat > async_function_calling.py << 'EOF'
import asyncio
import json
from openai import AsyncOpenAI
from dotenv import load_dotenv
import os

load_dotenv()
async_client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))

async def async_execute_tools(tool_calls: list, async_function_map: dict) -> list:
    """Tüm tool call'ları asenkron ve paralel çalıştırır"""
    tasks = []
    
    for tool_call in tool_calls:
        func_name = tool_call.function.name
        func_args = json.loads(tool_call.function.arguments)
        
        if func_name in async_function_map:
            task = asyncio.create_task(
                async_function_map[func_name](**func_args)
            )
            tasks.append((tool_call.id, task))
    
    results = []
    for call_id, task in tasks:
        try:
            result = await task
            results.append({
                "role": "tool",
                "tool_call_id": call_id,
                "content": json.dumps(result, ensure_ascii=False)
            })
        except Exception as e:
            results.append({
                "role": "tool",
                "tool_call_id": call_id,
                "content": json.dumps({"error": str(e)})
            })
    
    return results

# Örnek async fonksiyon
async def async_check_url_health(url: str) -> dict:
    import aiohttp
    async with aiohttp.ClientSession() as session:
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as response:
                return {
                    "url": url,
                    "status_code": response.status,
                    "is_healthy": response.status == 200,
                    "response_time_ms": 0  # gerçekte time ölçülür
                }
        except Exception as e:
            return {"url": url, "is_healthy": False, "error": str(e)}
EOF

Conversation state’i veritabanında saklayın. Uzun süreli chatbot uygulamalarında, konuşma geçmişini Redis veya PostgreSQL’de tutun. Hem kullanıcıya “geçen haftaki incident neydi?” sorusunu sorma imkanı tanır, hem de token limitine takılmazsınız.

Logging ve audit trail şart. Hangi fonksiyon, hangi parametrelerle, kim tarafından, ne zaman çağrıldığını mutlaka kaydedin. Hem debugging için hem de güvenlik auditi için kritik.

Sonuç

Function Calling, OpenAI API’yi gerçek anlamda işe yarar bir araca dönüştürüyor. Saf bir chatbot ile bir sysadmin asistanı arasındaki fark tam olarak burada: biri size genel bilgi verirken, diğeri sisteminize bakıyor, gerçek verileri okuyor ve aksiyonlar alabiliyor.

Bu yazıda anlattıklarımı özetlersek: temel request-response döngüsü, çoklu fonksiyon yönetimi, paralel çalıştırma, güvenlik kontrolleri, streaming ve async yapı. Bunların hepsini anlamadan production’a çıkmayın derim.

Bir sonraki adım olarak bu yapıyı MCP (Model Context Protocol) ile genişletmeyi, ya da LangChain/LlamaIndex gibi frameworklerin Function Calling’i nasıl soyutladığını inceleyebilirsiniz. Ama temeli sağlam atmadan framework soyutlamalarına geçmek, temeli olmayan binaya kat eklemek gibi. Önce buradaki kavramları özümseyin, sonra üstüne inşa edin.

Sorunuz varsa yorumlarda yazın, elimden geldiğince döneceğim.

Bir yanıt yazın

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