LangChain ve Ollama ile profesyonel AI uygulamaları geliştirmek için ileri düzey teknikleri keşfedelim.
1. Custom Output Parser (Özel Çıktı Ayrıştırıcı):
from langchain_core.output_parsers import BaseOutputParser
from langchain_core.exceptions import OutputParserException
import json
import re
class StructuredResponseParser(BaseOutputParser):
"""JSON formatında yapılandırılmış yanıt ayrıştırıcısı"""
def parse(self, text: str) -> dict:
try:
json_match = re.search(r'\{.*\}', text, re.DOTALL)
if json_match:
json_str = json_match.group()
return json.loads(json_str)
else:
lines = text.strip().split('\n')
result = {}
for line in lines:
if ':' in line:
key, value = line.split(':', 1)
result[key.strip().lower()] = value.strip()
return result
except Exception as e:
raise OutputParserException(f"Parsing hatası: {e}")
def get_format_instructions(self) -> str:
return """Yanıtını şu JSON formatında ver:
{
"konu": "ana konu",
"özet": "kısa özet",
"anahtar_kelimeler": ["kelime1", "kelime2"],
"güvenilirlik": 0.9
}"""
parser = StructuredResponseParser()
structured_prompt = PromptTemplate(
template="""Verilen metni analiz et ve yapılandırılmış bilgi çıkar.
Metin: {text}
{format_instructions}
Analiz:""",
input_variables=["text"],
partial_variables={"format_instructions": parser.get_format_instructions()}
)
llm = OllamaLLM(model="llama3.2", temperature=0.3)
structured_chain = structured_prompt | llm | parser
sample_text = """
LangChain, Python ve JavaScript ile AI uygulamaları geliştirmek için
kullanılan açık kaynak bir framework'tür. Büyük dil modellerini
uygulamalara entegre etmeyi kolaylaştırır ve modüler yapısı sayesinde
karmaşık AI iş akışları oluşturmanıza olanak tanır.
"""
result = structured_chain.invoke({"text": sample_text})
print("Yapılandırılmış Analiz:")
print(json.dumps(result, ensure_ascii=False, indent=2))
2. Callback System (Geri Çağırım Sistemi):
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
import logging
class DetailedCallbackHandler(BaseCallbackHandler):
"""Detaylı callback handler - logging ve monitoring için"""
def __init__(self):
self.start_time = None
self.token_count = 0
self.call_count = 0
def on_llm_start(self, serialized, prompts, **kwargs):
import time
self.start_time = time.time()
self.call_count += 1
print(f"🤖 LLM Başladı (Çağrı #{self.call_count})")
print(f" Prompt: {prompts[0][:100]}...")
def on_llm_end(self, response: LLMResult, **kwargs):
import time
end_time = time.time()
duration = end_time - self.start_time
if response.generations:
output = response.generations[0][0].text
self.token_count += len(output.split())
print(f"✅ LLM Tamamlandı")
print(f" Süre: {duration:.2f}s")
print(f" Token: ~{len(output.split())} kelime")
print(f" Hız: {len(output.split())/duration:.1f} kelime/sn")
def on_llm_error(self, error, **kwargs):
print(f"❌ LLM Hatası: {error}")
callback_handler = DetailedCallbackHandler()
monitored_llm = OllamaLLM(
model="llama3.2",
temperature=0.7,
callbacks=[callback_handler]
)
print("📊 Callback Monitoring Testi:\n")
questions = [
"Python'da exception handling nasıl yapılır?",
"REST API nedir?",
"Database indexing nasıl çalışır?"
]
for i, question in enumerate(questions, 1):
print(f"\n--- Soru {i} ---")
response = monitored_llm.invoke(question)
print(f"Cevap: {response[:100]}...\n")
print(f"📈 Toplam Statistics:")
print(f" Toplam Çağrı: {callback_handler.call_count}")
print(f" Toplam Token: ~{callback_handler.token_count}")
3. Async Processing (Asenkron İşleme):
import asyncio
from langchain_ollama import OllamaLLM
async def async_llm_call(llm, prompt, task_id):
"""Asenkron LLM çağrısı"""
print(f"🚀 Task {task_id} başladı")
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(None, llm.invoke, prompt)
print(f"✅ Task {task_id} tamamlandı")
return {'task_id': task_id, 'result': result}
async def batch_process_questions():
"""Çoklu soruları paralel işle"""
llm = OllamaLLM(model="phi3", temperature=0.3)
questions = [
"Python'da decorators nedir?",
"Git merge vs rebase farkı?",
"Docker container nedir?",
"API rate limiting nasıl yapılır?"
]
tasks = []
for i, question in enumerate(questions):
task = async_llm_call(llm, question, i+1)
tasks.append(task)
start_time = asyncio.get_event_loop().time()
results = await asyncio.gather(*tasks)
end_time = asyncio.get_event_loop().time()
print(f"\n⏱️ Toplam Süre: {end_time - start_time:.2f} saniye")
print(f"📊 İşlenen Soru: {len(results)}")
return results
if __name__ == "__main__":
print("🔄 Asenkron Batch Processing:\n")
results = asyncio.run(batch_process_questions())
print("\n📋 Sonuçlar:")
for result in results:
print(f"Task {result['task_id']}: {result['result'][:80]}...")
İleri Düzey Teknikler:
- Custom Chains: Özel işlem zincirleri
- Streaming: Real-time yanıt akışı
- Caching: Yanıt önbellekleme
- Error Handling: Gelişmiş hata yönetimi
- Model Switching: Dinamik model değişimi
Production Hazırlığı:
• Logging ve monitoring sistemi kurun
• Rate limiting ve quota yönetimi
• Error recovery ve fallback mekanizmaları
• Model versiyonlama ve deployment
• Güvenlik ve kullanıcı doğrulama
Öğrenmeye Devam: LangGraph, AutoGen
UI Framework: Streamlit, Gradio, FastAPI
Deployment: Docker, Kubernetes, Cloud
Monitoring: LangSmith, Weights & Biases