Ollama REST API ile Python Script Yazma
Ollama’yı komut satırından kullanmak harika, ama asıl güç REST API üzerinden gelir. Python ile Ollama API’sini kullanmaya başladığınızda, yerel LLM’leri kendi araçlarınıza, otomasyonlarınıza ve iş akışlarınıza entegre etmenin ne kadar kolay olduğunu göreceksiniz. Bu yazıda sıfırdan başlayıp production’a yakın senaryolara kadar her şeyi ele alacağız.
Ollama REST API’sine Genel Bakış
Ollama, varsayılan olarak http://localhost:11434 adresinde bir HTTP sunucusu başlatır. Bu sunucu üzerinden model yükleme, metin üretme, embedding alma gibi işlemleri yapabilirsiniz. Herhangi bir özel kütüphane kurmadan sadece requests ile her şeyi halledebilirsiniz, bu da bağımlılık yönetimini kolaylaştırır.
Temel endpoint’ler şunlardır:
- POST /api/generate: Tek seferlik metin üretimi
- POST /api/chat: Sohbet geçmişiyle konuşma
- POST /api/embeddings: Vektör embedding üretimi
- GET /api/tags: Yüklü modelleri listele
- POST /api/pull: Model indir
- DELETE /api/delete: Model sil
- GET /api/ps: Çalışan modelleri göster
API dokümantasyonu oldukça sade ve RESTful prensiplere uygun. Cevaplar JSON formatında gelir, streaming desteklenir.
Ortam Hazırlığı
Önce Python ortamımızı hazırlayalım. Sanal ortam kullanmayı alışkanlık haline getirmenizi öneririm:
# Sanal ortam oluştur ve aktif et
python3 -m venv ollama-env
source ollama-env/bin/activate # Linux/Mac
# Windows için: ollama-envScriptsactivate
# Gerekli paketleri kur
pip install requests python-dotenv rich
# Ollama'nın çalıştığını doğrula
curl http://localhost:11434/api/tags
rich kütüphanesini terminal çıktılarını güzel formatlamak için kullanacağız. python-dotenv ise konfigürasyon yönetimi için lazım olacak.
Ollama’nın çalışıp çalışmadığını Python’dan kontrol etmek için basit bir health check fonksiyonu yazalım:
import requests
import sys
def check_ollama_health(base_url="http://localhost:11434"):
"""Ollama servisinin ayakta olup olmadığını kontrol eder."""
try:
response = requests.get(f"{base_url}/api/tags", timeout=5)
if response.status_code == 200:
models = response.json().get("models", [])
print(f"Ollama calisiyor. Yuklu model sayisi: {len(models)}")
for model in models:
size_gb = model.get("size", 0) / (1024**3)
print(f" - {model['name']} ({size_gb:.1f} GB)")
return True
else:
print(f"Ollama yanit verdi ama hata kodu: {response.status_code}")
return False
except requests.exceptions.ConnectionError:
print("HATA: Ollama'ya baglanılamadı. Servis calisiyor mu?")
print("Calistirmak icin: ollama serve")
return False
except requests.exceptions.Timeout:
print("HATA: Ollama zaman asimina ugradi.")
return False
if __name__ == "__main__":
if not check_ollama_health():
sys.exit(1)
Temel Generate API Kullanımı
En basit kullanım senaryosuyla başlayalım. generate endpoint’i, bir prompt alır ve yanıt döner:
import requests
import json
def simple_generate(prompt, model="llama3.2", base_url="http://localhost:11434"):
"""
Basit metin uretimi yapar.
stream=False ile tum yaniti bekler.
"""
payload = {
"model": model,
"prompt": prompt,
"stream": False,
"options": {
"temperature": 0.7,
"top_p": 0.9,
"num_predict": 500 # Maksimum token sayisi
}
}
response = requests.post(
f"{base_url}/api/generate",
json=payload,
timeout=120
)
if response.status_code == 200:
result = response.json()
return result["response"]
else:
raise Exception(f"API hatasi: {response.status_code} - {response.text}")
# Kullanim ornegi
if __name__ == "__main__":
yanit = simple_generate(
"Python'da bir log parser nasil yazilir? Kisa ozet ver.",
model="llama3.2"
)
print(yanit)
Bu yaklaşım küçük sorgular için işe yarar ama büyük modellerde veya uzun çıktılarda kullanıcı deneyimi kötü olur çünkü tüm cevap gelene kadar ekranda hiçbir şey göremezsiniz.
Streaming ile Gerçek Zamanlı Çıktı
Production senaryolarında streaming kullanmak neredeyse zorunlu. Büyük dil modelleri token token üretir ve streaming’i etkinleştirdiğinizde her token geldiğinde işleyebilirsiniz:
import requests
import json
import sys
def streaming_generate(prompt, model="llama3.2", base_url="http://localhost:11434"):
"""
Streaming modunda metin uretir.
Her token geldikce ekrana yazar.
"""
payload = {
"model": model,
"prompt": prompt,
"stream": True,
"options": {
"temperature": 0.8,
"num_predict": 1000
}
}
full_response = ""
with requests.post(
f"{base_url}/api/generate",
json=payload,
stream=True,
timeout=300
) as response:
response.raise_for_status()
for line in response.iter_lines():
if line:
try:
chunk = json.loads(line.decode("utf-8"))
token = chunk.get("response", "")
full_response += token
# Her tokeni aninda yazdir, flush ile tamponu bosalt
print(token, end="", flush=True)
# done=True gelince uretim bitti demektir
if chunk.get("done", False):
print() # Son satir sonu
# Istatistikleri goster
total_duration = chunk.get("total_duration", 0) / 1e9
eval_count = chunk.get("eval_count", 0)
print(f"n--- Istatistikler ---")
print(f"Toplam sure: {total_duration:.2f} saniye")
print(f"Uretilen token: {eval_count}")
if total_duration > 0:
print(f"Hiz: {eval_count/total_duration:.1f} token/saniye")
break
except json.JSONDecodeError:
continue
return full_response
if __name__ == "__main__":
print("Yanit uretiliyor...n")
streaming_generate(
"Linux sistemde yuksek CPU kullanan prosesleri bulmak icin hangi komutlari kullanmali?"
)
Streaming ile token başına hız hesabı yapabilirsiniz. Bir llama3.2 modeli tipik olarak 20-50 token/saniye üretir, llama3.1:70b gibi büyük modeller ise 3-8 token/saniye civarında çalışır.
Chat API ile Konuşma Geçmişi Yönetimi
Chat API, sohbet geçmişini sizin yönetmenizi gerektirir. Stateless bir API olduğu için her istekte tüm konuşma geçmişini göndermeniz gerekir:
import requests
import json
from datetime import datetime
class OllamaChat:
"""
Ollama Chat API icin basit bir wrapper sinifi.
Konusma gecmisini yonetir ve oturum loglaması yapar.
"""
def __init__(self, model="llama3.2", base_url="http://localhost:11434",
system_prompt=None):
self.model = model
self.base_url = base_url
self.messages = []
self.session_start = datetime.now()
if system_prompt:
self.messages.append({
"role": "system",
"content": system_prompt
})
def chat(self, user_message, stream=True):
"""Kullanicidan mesaj alir ve modelden yanit ister."""
self.messages.append({
"role": "user",
"content": user_message
})
payload = {
"model": self.model,
"messages": self.messages,
"stream": stream,
"options": {
"temperature": 0.7,
"num_ctx": 4096 # Context window boyutu
}
}
assistant_message = ""
with requests.post(
f"{self.base_url}/api/chat",
json=payload,
stream=True,
timeout=300
) as response:
response.raise_for_status()
for line in response.iter_lines():
if line:
chunk = json.loads(line.decode("utf-8"))
if "message" in chunk:
token = chunk["message"].get("content", "")
assistant_message += token
print(token, end="", flush=True)
if chunk.get("done", False):
print()
break
# Asistan mesajini gecmise ekle
self.messages.append({
"role": "assistant",
"content": assistant_message
})
return assistant_message
def clear_history(self):
"""Konusma gecmisini temizler, sistem mesajini korur."""
system_messages = [m for m in self.messages if m["role"] == "system"]
self.messages = system_messages
print("Konusma gecmisi temizlendi.")
def get_token_estimate(self):
"""Yaklasik token kullanimi hesaplar (4 karakter = 1 token)."""
total_chars = sum(len(m["content"]) for m in self.messages)
return total_chars // 4
def save_session(self, filename=None):
"""Konusma oturumunu JSON dosyasina kaydeder."""
if not filename:
timestamp = self.session_start.strftime("%Y%m%d_%H%M%S")
filename = f"chat_session_{timestamp}.json"
session_data = {
"model": self.model,
"start_time": self.session_start.isoformat(),
"messages": self.messages
}
with open(filename, "w", encoding="utf-8") as f:
json.dump(session_data, f, ensure_ascii=False, indent=2)
print(f"Oturum kaydedildi: {filename}")
return filename
# Kullanim ornegi: Sysadmin asistani
if __name__ == "__main__":
asistan = OllamaChat(
model="llama3.2",
system_prompt="""Sen deneyimli bir Linux sistem yoneticisisin.
Kisa, net ve uygulanabilir yanıtlar ver.
Komutlari her zaman orneklerle goster."""
)
# Simule edilmis konusma
sorular = [
"Diskimdeki en buyuk dosyalari nasil bulabilirim?",
"Peki bu dosyalari otomatik olarak silebilir miyim?"
]
for soru in sorular:
print(f"nKullanici: {soru}n")
print("Asistan: ", end="")
asistan.chat(soru)
print(f"n[Tahmini token kullanimi: {asistan.get_token_estimate()}]")
asistan.save_session()
Bağlam penceresini dikkatli yönetin. num_ctx değerini model limitinin altında tutun. Llama 3.2 için 4096-8192 arası genellikle güvenli bir seçimdir. Uzun konuşmalarda token sayısını izleyin ve gerekirse eski mesajları budayın.
Embedding API ile Semantik Arama
Embedding’ler metinleri vektörlere dönüştürür. Bu vektörler üzerinden benzerlik araması yapabilirsiniz. Basit bir döküman arama sistemi kuralım:
import requests
import json
import numpy as np
from pathlib import Path
def get_embedding(text, model="nomic-embed-text", base_url="http://localhost:11434"):
"""Metin icin embedding vektoru uretir."""
payload = {
"model": model,
"prompt": text
}
response = requests.post(
f"{base_url}/api/embeddings",
json=payload,
timeout=60
)
response.raise_for_status()
return response.json()["embedding"]
def cosine_similarity(vec1, vec2):
"""Iki vektor arasindaki kosinus benzerligini hesaplar."""
vec1 = np.array(vec1)
vec2 = np.array(vec2)
dot_product = np.dot(vec1, vec2)
norm1 = np.linalg.norm(vec1)
norm2 = np.linalg.norm(vec2)
if norm1 == 0 or norm2 == 0:
return 0.0
return dot_product / (norm1 * norm2)
class SimpleDocumentSearch:
"""
Embedding tabanli basit dokuman arama sistemi.
Buyuk projeler icin ChromaDB veya Qdrant kullanmak daha mantiklidir.
"""
def __init__(self, embed_model="nomic-embed-text"):
self.embed_model = embed_model
self.documents = []
self.embeddings = []
def add_document(self, doc_id, content, metadata=None):
"""Sisteme yeni dokuman ekler ve embedding uretir."""
print(f"Embedding uretiliyor: {doc_id}")
embedding = get_embedding(content, self.embed_model)
self.documents.append({
"id": doc_id,
"content": content,
"metadata": metadata or {}
})
self.embeddings.append(embedding)
print(f" Tamamlandi. Vektor boyutu: {len(embedding)}")
def search(self, query, top_k=3):
"""Sorguya en yakin dokumanlari bulur."""
query_embedding = get_embedding(query, self.embed_model)
similarities = []
for i, doc_embedding in enumerate(self.embeddings):
sim = cosine_similarity(query_embedding, doc_embedding)
similarities.append((i, sim))
similarities.sort(key=lambda x: x[1], reverse=True)
results = []
for idx, score in similarities[:top_k]:
results.append({
"document": self.documents[idx],
"similarity_score": score
})
return results
# Gercek dunya senaryosu: Sysadmin runbook arama
if __name__ == "__main__":
searcher = SimpleDocumentSearch()
# Runbook dokumanlari ekle
runbooks = [
("runbook_001",
"Disk dolulugu alarmi: df -h komutuyla disk kullanimi kontrol edilir. "
"du -sh /* ile en buyuk dizinler bulunur. /var/log altindaki eski loglar temizlenir.",
{"kategori": "disk", "oncelik": "yuksek"}),
("runbook_002",
"Yuksek RAM kullanimi: free -m ile mevcut bellek durumu gorulur. "
"ps aux --sort=-%mem | head -20 ile en fazla RAM kullanan prosesler tespit edilir.",
{"kategori": "memory", "oncelik": "orta"}),
("runbook_003",
"SSH baglantilar kesilebiliyorsa: /etc/ssh/sshd_config dosyasinda ClientAliveInterval "
"ve ClientAliveCountMax parametreleri kontrol edilir. systemctl restart sshd ile servis yeniden baslatilir.",
{"kategori": "ssh", "oncelik": "orta"}),
]
for doc_id, content, meta in runbooks:
searcher.add_document(doc_id, content, meta)
print("n--- Arama Sonuclari ---")
query = "sunucu doluyor, yer kalmadi"
print(f"Sorgu: '{query}'n")
results = searcher.search(query, top_k=2)
for i, result in enumerate(results, 1):
doc = result["document"]
score = result["similarity_score"]
print(f"{i}. {doc['id']} (Benzerlik: {score:.3f})")
print(f" {doc['content'][:100]}...")
print(f" Metadata: {doc['metadata']}n")
Embedding için nomic-embed-text modelini kurmayı unutmayın: ollama pull nomic-embed-text
Pratik Senaryo: Log Analiz Scripti
Şimdiye kadar öğrendiklerimizi gerçek bir sysadmin senaryosunda kullanalım. Sistem loglarını analiz edip özet çıkaran bir script yazalım:
import requests
import json
import subprocess
from pathlib import Path
from datetime import datetime, timedelta
def get_recent_logs(log_file="/var/log/syslog", lines=100):
"""Son N satir sistem logunu okur."""
try:
result = subprocess.run(
["tail", "-n", str(lines), log_file],
capture_output=True,
text=True,
timeout=10
)
return result.stdout
except (subprocess.TimeoutExpired, FileNotFoundError) as e:
# Test icin ornek log uret
return generate_sample_logs()
def generate_sample_logs():
"""Test amacli ornek log uretir."""
return """
Jan 15 10:23:14 server01 kernel: Out of memory: Kill process 1234 (python3)
Jan 15 10:23:15 server01 systemd: web-app.service: Main process exited
Jan 15 10:24:01 server01 sshd: Failed password for invalid user admin from 192.168.1.100
Jan 15 10:24:05 server01 sshd: Failed password for invalid user root from 192.168.1.100
Jan 15 10:24:09 server01 sshd: Failed password for invalid user root from 192.168.1.100
Jan 15 10:25:00 server01 cron: disk_cleanup.sh completed successfully
Jan 15 10:26:33 server01 kernel: EXT4-fs error (device sda1): ext4_find_entry
Jan 15 10:30:00 server01 systemd: web-app.service: Start request repeated too quickly
"""
def analyze_logs_with_llm(log_content, model="llama3.2", base_url="http://localhost:11434"):
"""LLM kullanarak log analizi yapar."""
prompt = f"""Asagidaki sistem loglarini analiz et ve su basliklar altinda ozet cikar:
1. KRITIK SORUNLAR: Hemen mudahale gerektiren durumlar
2. UYARILAR: Dikkat edilmesi gereken ancak acil olmayan durumlar
3. GUVENLIK OLAYLARI: Yetkisiz erisim girisimleri, brute force vb.
4. ONERILEN AKSIYONLAR: Her sorun icin spesifik cozum adımlari
LOG ICERIK:
{log_content}
Teknik ve ozlü yaz. Komut ornekleri ver."""
payload = {
"model": model,
"prompt": prompt,
"stream": True,
"options": {
"temperature": 0.3, # Analiz icin dusuk temperature daha tutarli
"num_predict": 1000
}
}
print(f"n{'='*60}")
print(f"LOG ANALIZ RAPORU - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print(f"{'='*60}n")
with requests.post(
f"{base_url}/api/generate",
json=payload,
stream=True,
timeout=300
) as response:
response.raise_for_status()
full_response = ""
for line in response.iter_lines():
if line:
chunk = json.loads(line.decode("utf-8"))
token = chunk.get("response", "")
full_response += token
print(token, end="", flush=True)
if chunk.get("done", False):
print("n")
break
return full_response
def save_report(analysis, output_dir="/tmp/log_reports"):
"""Analiz raporunu dosyaya kaydeder."""
Path(output_dir).mkdir(exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
report_file = f"{output_dir}/log_analysis_{timestamp}.txt"
with open(report_file, "w", encoding="utf-8") as f:
f.write(f"Log Analiz Raporun")
f.write(f"Tarih: {datetime.now().isoformat()}n")
f.write("="*60 + "nn")
f.write(analysis)
print(f"Rapor kaydedildi: {report_file}")
return report_file
if __name__ == "__main__":
# Logları oku
print("Loglar okunuyor...")
logs = get_recent_logs(lines=50)
# LLM ile analiz et
analysis = analyze_logs_with_llm(logs)
# Raporu kaydet
save_report(analysis)
Hata Yönetimi ve Retry Mekanizması
Production scriptlerinde hata yönetimi kritik önem taşır. Özellikle büyük modeller yüklenirken veya sistem yoğunken timeout’lar yaşanabilir:
import requests
import time
import logging
from functools import wraps
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
def retry_on_failure(max_retries=3, delay=5, backoff=2):
"""
Decorator: Hata durumunda exponential backoff ile tekrar dener.
max_retries: Maksimum deneme sayisi
delay: Ilk bekleme suresi (saniye)
backoff: Her denemede bekleme suresini carp
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
current_delay = delay
for attempt in range(max_retries + 1):
try:
return func(*args, **kwargs)
except requests.exceptions.Timeout:
if attempt == max_retries:
logger.error(f"{func.__name__}: {max_retries} denemede zaman asimi")
raise
logger.warning(
f"{func.__name__}: Timeout (deneme {attempt+1}/{max_retries}). "
f"{current_delay}s beklenecek..."
)
except requests.exceptions.ConnectionError:
if attempt == max_retries:
logger.error(f"{func.__name__}: Baglanti kurulamadi")
raise
logger.warning(
f"Ollama'ya baglanılamadı (deneme {attempt+1}/{max_retries}). "
f"Servis calisiyor mu?"
)
except requests.exceptions.HTTPError as e:
# 404 veya 400 gibi hatalarda retry etme
if e.response.status_code in [400, 404]:
logger.error(f"Kurtarilamaz HTTP hatasi: {e.response.status_code}")
raise
logger.warning(f"HTTP hatasi {e.response.status_code}, yeniden denenecek")
time.sleep(current_delay)
current_delay *= backoff
return wrapper
return decorator
class RobustOllamaClient:
"""Hata yonetimi olan guvenilir Ollama istemcisi."""
def __init__(self, base_url="http://localhost:11434", model="llama3.2"):
self.base_url = base_url
self.model = model
self.session = requests.Session()
# Connection pooling icin adapter ayarla
adapter = requests.adapters.HTTPAdapter(
pool_connections=5,
pool_maxsize=10,
max_retries=0 # Manuel retry yapiyoruz
)
self.session.mount("http://", adapter)
@retry_on_failure(max_retries=3, delay=3, backoff=2)
def generate(self, prompt, temperature=0.7, max_tokens=500):
"""Metin uretir, hata durumunda yeniden dener."""
payload = {
"model": self.model,
"prompt": prompt,
"stream": False,
"options": {
"temperature": temperature,
"num_predict": max_tokens
}
}
response = self.session.post(
f"{self.base_url}/api/generate",
json=payload,
timeout=120
)
response.raise_for_status()
result = response.json()
logger.info(
f"Token uretildi: {result.get('eval_count', 0)}, "
f"Sure: {result.get('total_duration', 0)/1e9:.2f}s"
)
return result["response"]
@retry_on_failure(max_retries=2, delay=2)
def check_model_available(self):
"""Modelin yuklu olup olmadigini kontrol eder."""
response = self.session.get(f"{self.base_url}/api/tags", timeout=10)
response.raise_for_status()
models = response.json().get("models", [])
model_names = [m["name"] for m in models]
model_base = self.model.split(":")[0]
available = any(model_base in name for name in model_names)
if not available:
logger.error(
f"Model '{self.model}' yuklu degil. "
f"Yuklemek icin: ollama pull {self.model}"
)
return available
if __name__ == "__main__":
client = RobustOllamaClient(model="llama3.2")
if client.check_model_available():
try:
yanit = client.generate(
"Bir Linux sunucuda memory leak nasil tespit edilir?",
temperature=0.5,
max_tokens=300
)
print(yanit)
except Exception as e:
logger.error(f"Uretim basarisiz: {e}")
Performans Optimizasyonu İpuçları
API’yi verimli kullanmak için bazı pratik noktalar:
- num_ctx: Context window boyutunu ihtiyacınız kadar tutun. Gereksiz yere büyük değer verirseniz bellek boşa harcanır.
- num_predict: Maksimum token sayısını sınırlayın. Özellikle toplu işlemlerde kontrolsüz token üretimi sistemi yavaşlatır.
- temperature=0: Deterministik çıktı istiyorsanız (kod üretimi, veri çıkarma gibi) temperature’ı sıfıra yakın tutun.
- keep_alive parametresi:
generateendpoint’ine"keep_alive": "10m"eklerseniz model 10 dakika boyunca RAM’de tutulur, sonraki istekler çok daha hızlı gelir. - Batch işlemler: Çok sayıda küçük istek yerine prompt’ları birleştirmeye çalışın. Her istek model yükleme/hazırlama overhead’i taşır.
- Connection pooling: Yukarıdaki
RobustOllamaClientörneğindeki gibirequests.Sessionkullanın. Her istekte yeni TCP bağlantısı açmak gereksiz latency ekler.
Sonuç
Ollama REST API, yerel LLM’leri Python uygulamalarınıza entegre etmenin en temiz yoludur. Herhangi bir vendor bağımlılığı olmadan, verileriniz tamamen kendi altyapınızda kalarak güçlü LLM uygulamaları geliştirebilirsiniz.
Bu yazıda temel generate ve chat kullanımından başlayıp embedding tabanlı semantik aramaya, log analiz scriptine ve production kalitesinde hata yönetimine kadar geniş bir spektrumu ele aldık. Gerçek ortamlarda streaming kesinlikle kullanın, büyük context window’lardan kaçının ve retry mekanizmasını ihmal etmeyin.
Bir sonraki adım olarak ChromaDB veya Qdrant gibi vektör veritabanlarını Ollama ile birleştirerek RAG (Retrieval-Augmented Generation) sistemi kurmayı araştırabilirsiniz. Ya da bu scriptleri cron job’lara bağlayarak günlük log analizi, anomali tespiti gibi tamamen otomatik iş akışları oluşturabilirsiniz. Yerel LLM’lerin güzelliği de zaten bu: denemekten çekinmiyorsunuz çünkü API ücreti yok, rate limit yok, gizlilik kaygısı yok.
