LangChain Tracing ile Uygulama Hata Ayıklama

Yazar:komut.io ekibi
0 Yorum

Prodüksiyonda bir LangChain uygulaması patladığında ve stack trace sana sadece “Chain raised an exception” gibi işe yaramaz bir şey söylediğinde ne yaparsın? Ben bu durumu defalarca yaşadım ve her seferinde aynı sonuca vardım: tracing olmadan LLM uygulamaları debug etmek, ışık olmadan araba tamir etmeye benziyor. Neye baktığını göremiyorsun.

LangChain’in tracing altyapısı, özellikle LangSmith entegrasyonuyla birlikte, bu karanlığı ortadan kaldırıyor. Bu yazıda gerçek prodüksiyon senaryolarından örnekler vererek nasıl kurulur, nasıl kullanılır ve en önemlisi nasıl yorumlanır, bunları aktaracağım.

LangSmith Nedir ve Neden Gerekli?

LangChain’in kendi debug mekanizmaları başlangıç için yeterli olabilir ama bir chain birden fazla LLM çağrısı yapıyorsa, retrieval adımları varsa veya agent döngüleri söz konusuysa, standart logging yetersiz kalıyor. LangSmith bu noktada devreye giriyor.

LangSmith’in temel olarak yaptığı şu: Her LLM çağrısını, her araç kullanımını, her vektör veritabanı sorgusunu bir “run” olarak kaydediyor. Bu run’ları hiyerarşik olarak görüntüleyebiliyorsun. Bir agent’ın kaç adım attığını, her adımda ne kadar token harcadığını, hangi ara çıktıların üretildiğini görebiliyorsun.

Ücretli bir servis ama ücretsiz katmanı geliştirme ve test için gayet yeterli. Alternatif olarak self-hosted seçenek de var, bunu da ileride ele alacağım.

Ortam Kurulumu

Önce temel kurulumu yapalım. Python ortamınızın hazır olduğunu varsayıyorum.

pip install langchain langchain-openai langsmith python-dotenv

LangSmith hesabı açtıktan sonra API anahtarınızı alıyorsunuz. Sonra ortam değişkenlerini ayarlıyorsunuz:

export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_ENDPOINT="https://api.smith.langchain.com"
export LANGCHAIN_API_KEY="ls__xxxxxxxxxxxxx"
export LANGCHAIN_PROJECT="production-debug"
export OPENAI_API_KEY="sk-xxxxxxxxxxxxx"

Bu değişkenleri .env dosyasına koyup python-dotenv ile yüklemek daha temiz bir yaklaşım. Özellikle CI/CD pipeline’larında environment variable yönetimi kritik hale geliyor.

from dotenv import load_dotenv
import os

load_dotenv()

# Tracing aktif mi kontrol et
print(f"Tracing aktif: {os.getenv('LANGCHAIN_TRACING_V2')}")
print(f"Proje: {os.getenv('LANGCHAIN_PROJECT')}")

LANGCHAIN_TRACING_V2=true ayarlandığı anda LangChain otomatik olarak tüm chain ve LLM çağrılarını LangSmith’e göndermeye başlıyor. Ek kod yazmanıza gerek yok, bu önemli bir nokta.

İlk Trace: Basit Bir Örnekle Başlayalım

Kompleks bir örnekle başlamak yerine önce temel akışı görelim. Sonra üzerine katman katman ekleyeceğiz.

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "Sen yardımcı bir asistansın. Türkçe yanıt ver."),
    ("human", "{soru}")
])

chain = prompt | llm | StrOutputParser()

yanit = chain.invoke({"soru": "Python'da list comprehension nasıl kullanılır?"})
print(yanit)

Bu kodu çalıştırdıktan sonra LangSmith dashboard’una girdiğinizde bu çağrıyı göreceksiniz. Prompt’un tam içeriği, model parametreleri, kullanılan token sayısı, yanıt süresi. Hepsi kayıt altında.

Gerçek Dünya Senaryosu: RAG Pipeline Debug

Şimdi daha ilginç bir senaryoya geçelim. Bir müşteri destek botu kuruyorsunuz. Sorular geliyor, vektör veritabanından ilgili dokümanlar çekiliyor, LLM yanıt üretiyor. Kulağa basit geliyor ama prodüksiyonda şu sorunlarla karşılaştım:

  • Retrieval adımı yanlış dokümanları getiriyor
  • LLM aldığı context’i görmezden geliyor
  • Bazı sorgular için yanıt kalitesi tutarsız

Tracing olmadan bu sorunların hangisinde takıldığınızı bulmak saatler alıyor. Tracing ile saniyeler.

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langsmith import traceable

# Örnek dokümanlar (gerçekte bunlar veritabanından gelir)
from langchain_core.documents import Document

docs = [
    Document(page_content="Ürünümüzün iade süresi 30 gündür.", metadata={"source": "iade_politikasi"}),
    Document(page_content="Kargo süresi 2-5 iş günüdür.", metadata={"source": "kargo_bilgisi"}),
    Document(page_content="Teknik destek hattımız 7/24 hizmet vermektedir.", metadata={"source": "destek"}),
]

embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(docs, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

prompt = ChatPromptTemplate.from_template("""
Aşağıdaki bağlamı kullanarak soruyu yanıtla.
Bağlamda yanıt yoksa 'Bu konuda bilgim yok' de.

Bağlam:
{context}

Soru: {soru}
""")

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def format_docs(docs):
    return "nn".join(doc.page_content for doc in docs)

rag_chain = (
    {"context": retriever | format_docs, "soru": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

yanit = rag_chain.invoke("İade politikanız nedir?")
print(yanit)

Bu chain’i LangSmith’te incelediğinizde şunları görebiliyorsunuz: retriever’ın tam olarak hangi dokümanları getirdiğini, bu dokümanların prompt’a nasıl eklendiğini ve LLM’e giden son prompt’un içeriğini. Retrieval kalitesi düşükse hemen anlıyorsunuz, LLM prompt’u yanlış mı yorumluyor da anlıyorsunuz.

@traceable Dekoratörü ile Custom Fonksiyon İzleme

Bazen LangChain bileşenleri dışında kendi yazdığınız fonksiyonları da trace etmek istiyorsunuz. @traceable dekoratörü tam bu iş için.

from langsmith import traceable
import json

@traceable(name="Veri Ön İşleme", tags=["preprocessing", "production"])
def kullanici_sorusunu_isle(ham_soru: str) -> dict:
    """
    Kullanıcı sorusunu temizler ve kategorize eder.
    Bu fonksiyon artık LangSmith'te görünecek.
    """
    temiz_soru = ham_soru.strip().lower()
    
    # Basit kategorizasyon
    kategori = "genel"
    if any(kelime in temiz_soru for kelime in ["iade", "iptal", "geri"]):
        kategori = "iade_islemleri"
    elif any(kelime in temiz_soru for kelime in ["kargo", "teslimat", "ne zaman"]):
        kategori = "kargo"
    elif any(kelime in temiz_soru for kelime in ["teknik", "sorun", "çalışmıyor"]):
        kategori = "teknik_destek"
    
    return {
        "temiz_soru": temiz_soru,
        "kategori": kategori,
        "uzunluk": len(temiz_soru)
    }

@traceable(name="Yanıt Kalite Kontrolü", tags=["quality-check"])
def yanit_kalite_kontrol(yanit: str, min_uzunluk: int = 50) -> dict:
    """
    Üretilen yanıtın kalite kriterlerini kontrol eder.
    """
    problemler = []
    
    if len(yanit) < min_uzunluk:
        problemler.append("Yanıt çok kısa")
    
    if "bilmiyorum" in yanit.lower() and "özür" not in yanit.lower():
        problemler.append("Kibarsız ret yanıtı")
    
    return {
        "kaliteli": len(problemler) == 0,
        "problemler": problemler,
        "yanit_uzunlugu": len(yanit)
    }

# Kullanım
soru_verisi = kullanici_sorusunu_isle("İade işlemi nasıl yapılır?")
print(f"İşlenmiş veri: {soru_verisi}")

@traceable ile işaretlediğiniz fonksiyonlar LangSmith’te ayrı birer run olarak görünüyor. Fonksiyona giren parametreler ve çıkan değerler kayıt altına alınıyor. Bir pipeline’daki her adımın ne ürettiğini görmek, hataların kaynağını bulmayı inanılmaz kolaylaştırıyor.

Run Metadata ve Tags Kullanımı

Prodüksiyonda farklı kullanıcılardan gelen istekleri, farklı A/B test gruplarını veya farklı model versiyonlarını ayırt etmek için metadata ve tag’leri etkin kullanmak gerekiyor.

from langchain_core.callbacks import BaseCallbackHandler
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# RunnableConfig ile metadata ekleme
from langchain_core.runnables import RunnableConfig

llm = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_messages([
    ("system", "Kısa ve öz yanıtlar ver."),
    ("human", "{soru}")
])
chain = prompt | llm | StrOutputParser()

# Her çağrıya özel metadata ekleyebilirsiniz
config = RunnableConfig(
    tags=["production", "v2.1", "ab-test-group-b"],
    metadata={
        "kullanici_id": "usr_12345",
        "oturum_id": "sess_abc789",
        "uygulama_versiyonu": "2.1.0",
        "ortam": "production"
    },
    run_name="Müşteri Destek - Ürün Sorusu"
)

yanit = chain.invoke(
    {"soru": "Python list ile tuple arasındaki fark nedir?"},
    config=config
)
print(yanit)

Bu metadata LangSmith’te filtreleme için kullanılabiliyor. “Sadece A/B test grubunun çağrılarını göster” veya “Belirli bir kullanıcının oturumundaki tüm çağrıları listele” gibi sorgular yapabiliyorsunuz. Prodüksiyonda hangi kullanıcının sorunlu deneyim yaşadığını tespit etmek çok kolaylaşıyor.

Hata Ayıklama: Gerçek Bir Sorun Senaryosu

Geçen ay bir agent uygulamasında şu sorunla karşılaştım: Agent bazen döngüye giriyor, aynı aracı defalarca çağırıyordu. Log’lara baktığımda sadece “Max iterations reached” hatasını görüyordum. LangSmith olmadan ne olduğunu anlamam çok zaman alırdı.

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
import random

@tool
def veritabani_sorgula(sorgu: str) -> str:
    """Veritabanında kayıt ara. Sorguyu string olarak al."""
    # Simüle edilmiş veritabanı - kasıtlı olarak bazen boş döndürüyor
    if random.random() > 0.7:
        return f"'{sorgu}' için sonuç bulunamadı."
    return f"Sonuç: {sorgu} ile ilgili 3 kayıt bulundu."

@tool  
def rapor_olustur(veri: str) -> str:
    """Verilen veriden rapor oluştur."""
    return f"Rapor oluşturuldu: {veri[:50]}..."

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", """Sen bir veri analisti asistanısın. 
    Görevleri adım adım tamamla. 
    Eğer veritabanında sonuç bulunamazsa farklı bir sorgu dene, 
    ama aynı sorguyu 2 defadan fazla tekrarlama."""),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

tools = [veritabani_sorgula, rapor_olustur]
agent = create_tool_calling_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10,
    return_intermediate_steps=True
)

from langchain_core.runnables import RunnableConfig

config = RunnableConfig(
    tags=["agent", "debug-session"],
    metadata={"test_senaryosu": "dongu_testi"},
    run_name="Agent Döngü Debug Testi"
)

try:
    sonuc = agent_executor.invoke(
        {"input": "Satış verilerini sorgula ve rapor oluştur"},
        config=config
    )
    print("Sonuç:", sonuc["output"])
except Exception as e:
    print(f"Hata: {e}")

Bu kodu çalıştırıp LangSmith’te incelediğinizde her agent adımını net olarak görebiliyorsunuz. Hangi araç kaç kez çağrıldı, her çağrıda ne döndü, agent bunlara nasıl tepki verdi. Döngü problemini tespit etmek için artık tahminde bulunmak zorunda değilsiniz.

LangSmith SDK ile Programatik Erişim

Dashboard güzel ama bazen verilere programatik erişmek gerekiyor. Özellikle kendi monitoring dashboard’unuzu kurmak veya otomasyon yapmak istediğinizde.

from langsmith import Client
from datetime import datetime, timedelta

client = Client()

# Son 24 saatteki başarısız run'ları listele
son_24_saat = datetime.utcnow() - timedelta(hours=24)

basarisiz_runlar = client.list_runs(
    project_name="production-debug",
    error=True,
    start_time=son_24_saat,
    limit=50
)

print("Son 24 saatteki hatalar:")
for run in basarisiz_runlar:
    print(f"  Run ID: {run.id}")
    print(f"  Hata: {run.error[:100] if run.error else 'Bilinmiyor'}")
    print(f"  Süre: {run.end_time - run.start_time if run.end_time else 'Devam ediyor'}")
    print(f"  Tags: {run.tags}")
    print("  ---")

# Belirli bir tag'e sahip run'ların ortalama süresini hesapla
production_runlar = list(client.list_runs(
    project_name="production-debug",
    filter='has(tags, "production")',
    start_time=son_24_saat,
    limit=100
))

if production_runlar:
    sureler = [
        (r.end_time - r.start_time).total_seconds()
        for r in production_runlar
        if r.end_time and r.start_time
    ]
    if sureler:
        ortalama_sure = sum(sureler) / len(sureler)
        print(f"nOrtalama yanıt süresi: {ortalama_sure:.2f} saniye")
        print(f"Toplam sorgulan run sayısı: {len(production_runlar)}")

Bu yaklaşım özellikle uyarı sistemi kurarken işe yarıyor. Hata oranı belirli bir eşiği aştığında Slack’e mesaj atmak, ortalama yanıt süresi yükseldiğinde PagerDuty’ye alert göndermek gibi senaryolar için bu SDK’yı kullanabilirsiniz.

Self-Hosted Alternatif: Yerel LangSmith

Veri gizliliği gerektiren ortamlar için LangSmith’i self-hosted çalıştırabiliyorsunuz. Bu özellikle bankacılık, sağlık veya kamu projelerinde önemli.

# LangSmith self-hosted kurulumu
mkdir langsmith-local && cd langsmith-local

# Docker Compose dosyasını indir
curl -o docker-compose.yml 
  https://raw.githubusercontent.com/langchain-ai/langsmith-sdk/main/python/langsmith/docker-compose.yaml

# Ortam değişkenlerini ayarla
cat > .env << EOF
LANGSMITH_LICENSE_KEY=your-license-key-here
POSTGRES_DB=langsmith
POSTGRES_USER=langsmith
POSTGRES_PASSWORD=guclu-bir-sifre-koy
EOF

# Servisleri başlat
docker-compose up -d

# Durumu kontrol et
docker-compose ps
docker-compose logs -f langchain-backend

Self-hosted kurulumda endpoint’i değiştirmeniz yeterli:

export LANGCHAIN_ENDPOINT="http://localhost:1984"
export LANGCHAIN_API_KEY="local-api-key"

Tüm trace verileriniz artık kendi sunucunuzda. Dışarıya çıkan tek şey LangChain kütüphanesi kendisi, veriler değil.

Prompt Versiyonlama ve A/B Testing

LangSmith’in en az bilinen ama en değerli özelliklerinden biri prompt hub entegrasyonu. Farklı prompt versiyonlarını test etmek ve karşılaştırmak için kullanabiliyorsunuz.

from langsmith import Client
from langchain import hub

client = Client()

# Prompt'u hub'a yükle (ilk sefer)
# client.push_prompt("musteri-destek-v1", ...)

# Hub'dan prompt çek
try:
    prompt_v1 = hub.pull("musteri-destek-v1")
    prompt_v2 = hub.pull("musteri-destek-v2")
except Exception:
    # Hub'da yoksa yerel kullan
    from langchain_core.prompts import ChatPromptTemplate
    prompt_v1 = ChatPromptTemplate.from_messages([
        ("system", "Sen yardımcı bir asistansın."),
        ("human", "{soru}")
    ])

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableConfig

llm = ChatOpenAI(model="gpt-4o-mini")

# A/B test - iki farklı prompt ile test
test_sorulari = [
    "Ürünümü iade etmek istiyorum",
    "Kargom nerede?",
    "Teknik destek lazım"
]

for soru in test_sorulari:
    # V1 prompt ile test
    chain_v1 = prompt_v1 | llm | StrOutputParser()
    yanit_v1 = chain_v1.invoke(
        {"soru": soru},
        config=RunnableConfig(
            tags=["ab-test", "prompt-v1"],
            metadata={"prompt_versiyonu": "v1", "test_sorusu": soru}
        )
    )
    
    print(f"Soru: {soru}")
    print(f"V1 Yanıt: {yanit_v1[:100]}...")
    print()

LangSmith’te bu iki grubun metriklerini yan yana karşılaştırabiliyorsunuz: hangi prompt versiyonu daha az token kullandı, hangi versiyonun yanıtları daha uzundu, hata oranları nasıl karşılaştırılıyor.

Callback Handler ile Özel İzleme

Kendi özel izleme mantığınızı eklemek için callback handler yazabilirsiniz. Bu özellikle LangSmith’e gitmesini istemediğiniz hassas veriler için ya da özel metrik toplamak istediğinizde gerekiyor.

from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from typing import Any, Union
import time
import logging

logger = logging.getLogger(__name__)

class PerformansCallback(BaseCallbackHandler):
    """LLM çağrılarının performansını izler ve loglar."""
    
    def __init__(self):
        self.baslangic_zamanlari = {}
        self.toplam_token = 0
        self.cagri_sayisi = 0
    
    def on_llm_start(self, serialized: dict, prompts: list, **kwargs):
        run_id = str(kwargs.get("run_id", "bilinmiyor"))
        self.baslangic_zamanlari[run_id] = time.time()
        self.cagri_sayisi += 1
        logger.info(f"LLM çağrısı başladı - Run ID: {run_id[:8]}")
    
    def on_llm_end(self, response: LLMResult, **kwargs):
        run_id = str(kwargs.get("run_id", "bilinmiyor"))
        baslangic = self.baslangic_zamanlari.pop(run_id, None)
        
        if baslangic:
            sure = time.time() - baslangic
            logger.info(f"LLM çağrısı tamamlandı - Süre: {sure:.2f}s")
        
        if response.llm_output and "token_usage" in response.llm_output:
            kullanilan = response.llm_output["token_usage"].get("total_tokens", 0)
            self.toplam_token += kullanilan
            logger.info(f"Kullanılan token: {kullanilan}, Toplam: {self.toplam_token}")
    
    def on_llm_error(self, error: Union[Exception, KeyboardInterrupt], **kwargs):
        logger.error(f"LLM hatası: {str(error)}")
    
    def on_chain_error(self, error: Union[Exception, KeyboardInterrupt], **kwargs):
        logger.error(f"Chain hatası: {str(error)}")

# Callback'i kullan
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

perf_callback = PerformansCallback()
llm = ChatOpenAI(
    model="gpt-4o-mini",
    callbacks=[perf_callback]
)

prompt = ChatPromptTemplate.from_messages([
    ("human", "{soru}")
])

chain = prompt | llm | StrOutputParser()
yanit = chain.invoke({"soru": "Kubernetes pod nedir?"})

print(f"nToplam çağrı sayısı: {perf_callback.cagri_sayisi}")
print(f"Toplam kullanılan token: {perf_callback.toplam_token}")

Bu callback’i LangSmith ile birlikte kullanabilirsiniz. LangSmith genel tracing’i yaparken bu callback kendi özel metriklerinizi topluyor. İkisi birbirini tamamlıyor.

Prodüksiyon İpuçları

Tracing’i prodüksiyona aldığınızda dikkat etmeniz gereken birkaç şey var:

Sampling kullanın: Her isteği trace etmek pahalı olabilir. Yüksek trafikli uygulamalarda yüzde 10 veya 20 sampling yeterli olabiliyor.

# Sampling için ortam değişkeni
export LANGCHAIN_TRACING_SAMPLING_RATE=0.1

Hassas veri maskeleyin: Kullanıcı kişisel verilerini trace’e koymayın. @traceable dekoratöründe hide_inputs=True veya hide_outputs=True parametrelerini kullanabilirsiniz.

Proje isimlerini anlamlı tutun: production-v2-customer-support gibi isimler, test123 gibi isimlerden çok daha yararlı. Aylarca geçmiş data’ya baktığınızda hangi projenin ne olduğunu anlıyorsunuz.

Alert kurun: LangSmith’in webhook’larını veya kendi script’lerinizle hata oranı belirli bir eşiği geçtiğinde alarm kurun. Prodüksiyonda sessiz hatalar en tehlikeli olanlardır.

Token maliyetlerini takip edin: LangSmith metadata’sında token kullanımını görüyorsunuz. Hangi query pattern’leri en çok token tüketiyor, bunu analiz etmek maliyet optimizasyonu için kritik.

Sonuç

LangChain uygulamaları belirli bir kompleksliğin üzerine çıktığında tracing bir lüks olmaktan çıkıp zorunluluk haline geliyor. “Bu agent neden döngüye giriyor?”, “RAG pipeline’ım neden alakasız cevaplar üretiyor?”, “Hangi prompt versiyonu daha iyi performans gösteriyor?” gibi soruları cevaplamak için tracing şart.

LangSmith, doğrudan LangChain ekosistemiyle entegre geldiği için başlangıç maliyeti düşük. LANGCHAIN_TRACING_V2=true ortam değişkenini set ettiğiniz anda temel tracing başlıyor. Üzerine @traceable dekoratörleri, metadata ve callback handler’lar ekledikçe giderek daha detaylı görünürlük elde ediyorsunuz.

Veri gizliliği endişesi varsa self-hosted seçenek var. Maliyet endişesi varsa sampling var. Özel entegrasyona ihtiyaç varsa SDK ve callback’ler var. Geliştirme aşamasından başlayıp prodüksiyona taşıyabileceğiniz, ölçeklenebilir bir observability altyapısı kurabilirsiniz.

LLM uygulamalarında hata ayıklamak, geleneksel yazılım debug’ından farklı. Determinizm yok, her çağrı potansiyel olarak farklı sonuç üretebiliyor. Bu belirsizliği yönetmenin tek yolu iyi bir tracing altyapısı kurmaktan geçiyor.

Bir yanıt yazın

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