Tutorial 8: LangSmith - Observabilitas dan Debugging LLM
Daftar Isi
Pendahuluan
Membangun aplikasi berbasis LLM hanyalah separuh dari tantangan. Memahami bagaimana model berperilaku di produksi, mengapa model gagal, berapa biayanya, dan apakah prompt Anda benar-benar membaik dari waktu ke waktu adalah separuh yang sama pentingnya. LangSmith adalah platform observabilitas dan evaluasi yang dibangun oleh LangChain khusus untuk mengatasi tantangan-tantangan ini.
LangSmith menyediakan pelacakan menyeluruh (end-to-end tracing) dari setiap panggilan LLM, eksekusi chain, dan langkah agent. Platform ini memungkinkan Anda membuat dataset evaluasi, menjalankan evaluasi otomatis, mengelola versi prompt, memantau kinerja produksi, dan men-debug alur kerja multi-langkah yang kompleks. Baik Anda sedang membuat prototipe chatbot sederhana maupun menjalankan pipeline RAG di produksi, LangSmith memberikan visibilitas yang Anda butuhkan untuk menghasilkan aplikasi AI yang andal.
Tutorial ini membahas setiap fitur utama LangSmith dengan contoh kode praktis yang siap digunakan di produksi.
Prasyarat
- Python 3.9 atau lebih tinggi
- Akun LangSmith aktif (daftar di smith.langchain.com)
- API key OpenAI (atau penyedia LLM lain yang didukung)
- Pemahaman dasar tentang konsep LangChain
Instal paket yang diperlukan:
pip install langsmith langchain langchain-openai openai
Menyiapkan LangSmith
Langkah 1: Konfigurasi Variabel Lingkungan
LangSmith memerlukan beberapa variabel lingkungan untuk menghubungkan aplikasi Anda ke platform.
import os
os.environ["LANGCHAINTRACINGV2"] = "true"
os.environ["LANGCHAINAPIKEY"] = "ls_apikeyandadisini"
os.environ["LANGCHAINPROJECT"] = "proyek-pertama-saya"
os.environ["LANGCHAINENDPOINT"] = "https://api.smith.langchain.com"
Langkah 2: Verifikasi Koneksi
from langsmith import Client
client = Client()
Daftar proyek yang ada untuk memverifikasi konektivitas
projects = list(client.listprojects())
print(f"Terhubung ke LangSmith. Ditemukan {len(projects)} proyek.")
for project in projects:
print(f" - {project.name} (dibuat: {project.createdat})")
Langkah 3: Membuat Proyek Khusus
# Buat proyek baru untuk pelacakan yang terorganisir
namaproyek = "tutorial-langsmith-demo"
try:
project = client.createproject(namaproyek)
print(f"Proyek '{project.name}' berhasil dibuat.")
except Exception as e:
print(f"Proyek mungkin sudah ada: {e}")
Tetapkan proyek aktif
os.environ["LANGCHAINPROJECT"] = namaproyek
Melacak Panggilan LLM
Pelacakan Dasar dengan LangChain
Ketika pelacakan diaktifkan, setiap panggilan LangChain secara otomatis ditangkap.
from langchainopenai import ChatOpenAI
from langchain
core.messages import HumanMessage, SystemMessage
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
messages = [
SystemMessage(content="Anda adalah asisten coding yang membantu."),
HumanMessage(content="Jelaskan decorator Python dalam 3 kalimat.")
]
response = llm.invoke(messages)
print(response.content)
Trace secara otomatis dikirim ke LangSmith
Pelacakan Manual dengan Dekorator @traceable
Untuk fungsi kustom yang bukan komponen LangChain, gunakan dekorator @traceable.
from langsmith import traceable
@traceable(name="prosespertanyaanpengguna")
def prosespertanyaanpengguna(pertanyaan: str, konteks: str = "") -> dict:
"""Memproses pertanyaan pengguna dengan konteks opsional."""
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = f"Konteks: {konteks}\n\nPertanyaan: {pertanyaan}" if konteks else pertanyaan
response = llm.invoke([HumanMessage(content=prompt)])
return {
"pertanyaan": pertanyaan,
"jawaban": response.content,
"model": "gpt-4o",
"memilikikonteks": bool(konteks)
}
Setiap panggilan dilacak dengan input, output, dan metadata
hasil = prosespertanyaanpengguna(
pertanyaan="Apa itu dependency injection?",
konteks="Kita sedang membahas pola desain perangkat lunak."
)
print(hasil["jawaban"])
Pelacakan Bersarang (Nested Tracing)
Panggilan fungsi bersarang membuat pohon trace, memudahkan melihat alur eksekusi lengkap.
@traceable(name="ambildokumen")
def ambildokumen(query: str) -> list:
"""Simulasi pengambilan dokumen."""
return [
{"id": 1, "teks": "Python adalah bahasa pemrograman tingkat tinggi."},
{"id": 2, "teks": "Python mendukung berbagai paradigma pemrograman."},
]
@traceable(name="buatjawaban")
def buatjawaban(query: str, dokumen: list) -> str:
"""Menghasilkan jawaban dari dokumen yang diambil."""
llm = ChatOpenAI(model="gpt-4o", temperature=0)
konteks = "\n".join([doc["teks"] for doc in dokumen])
prompt = f"Berdasarkan konteks berikut:\n{konteks}\n\nJawab: {query}"
response = llm.invoke([HumanMessage(content=prompt)])
return response.content
@traceable(name="pipelinerag")
def pipelinerag(query: str) -> str:
"""Pipeline RAG lengkap dengan pelacakan bersarang."""
docs = ambildokumen(query)
jawaban = buatjawaban(query, docs)
return jawaban
jawaban = pipelinerag("Paradigma pemrograman apa yang didukung Python?")
print(jawaban)
Dataset Evaluasi
Membuat Dataset
from langsmith import Client
client = Client()
Buat dataset untuk evaluasi
namadataset = "set-evaluasi-qa"
dataset = client.createdataset(
namadataset,
description="Pasangan pertanyaan-jawaban untuk mengevaluasi pipeline RAG kami."
)
Tambahkan contoh ke dataset
contohdata = [
{
"inputs": {"query": "Apa itu Python?"},
"outputs": {"answer": "Python adalah bahasa pemrograman tingkat tinggi yang diinterpretasi."}
},
{
"inputs": {"query": "Apa itu machine learning?"},
"outputs": {"answer": "Machine learning adalah subset dari AI yang memungkinkan sistem belajar dari data."}
},
{
"inputs": {"query": "Apa itu neural network?"},
"outputs": {"answer": "Neural network adalah model komputasi yang terinspirasi dari otak manusia."}
},
{
"inputs": {"query": "Apa itu natural language processing?"},
"outputs": {"answer": "NLP adalah bidang AI yang berfokus pada interaksi antara komputer dan bahasa manusia."}
},
]
for contoh in contohdata:
client.createexample(
inputs=contoh["inputs"],
outputs=contoh["outputs"],
datasetid=dataset.id
)
print(f"Dataset '{namadataset}' dibuat dengan {len(contohdata)} contoh.")
Menjalankan Evaluasi Terhadap Dataset
from langsmith.evaluation import evaluate
def prediksijawaban(inputs: dict) -> dict:
"""Fungsi yang akan dievaluasi."""
llm = ChatOpenAI(model="gpt-4o", temperature=0)
response = llm.invoke([HumanMessage(content=inputs["query"])])
return {"answer": response.content}
results = evaluate(
prediksijawaban,
data=namadataset,
experimentprefix="gpt4o-baseline",
maxconcurrency=2,
)
print(f"Evaluasi selesai. Lihat hasil di: {results.experimenturl}")
Evaluator Kustom
Membangun Evaluator Kemiripan Semantik
from langsmith.schemas import Example, Run
def evaluatorkemiripansemantik(run: Run, example: Example) -> dict:
"""Evaluasi kemiripan semantik antara jawaban prediksi dan yang diharapkan."""
from langchainopenai import OpenAIEmbeddings
import numpy as np
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
prediksi = run.outputs.get("answer", "")
ekspektasi = example.outputs.get("answer", "")
embprediksi = embeddings.embedquery(prediksi)
embekspektasi = embeddings.embedquery(ekspektasi)
kemiripan = np.dot(embprediksi, embekspektasi) / (
np.linalg.norm(embprediksi) np.linalg.norm(embekspektasi)
)
return {
"key": "kemiripansemantik",
"score": float(kemiripan),
"comment": f"Cosine similarity: {kemiripan:.4f}"
}
def evaluatorrelevansijawaban(run: Run, example: Example) -> dict:
"""Periksa apakah jawaban relevan dengan pertanyaan."""
llm = ChatOpenAI(model="gpt-4o", temperature=0)
query = example.inputs.get("query", "")
prediksi = run.outputs.get("answer", "")
evalprompt = f"""Nilai seberapa relevan jawaban berikut terhadap pertanyaan.
Pertanyaan: {query}
Jawaban: {prediksi}
Kembalikan hanya angka antara 0 dan 1, di mana 1 berarti sangat relevan."""
response = llm.invoke([HumanMessage(content=evalprompt)])
try:
skor = float(response.content.strip())
except ValueError:
skor = 0.0
return {
"key": "relevansijawaban",
"score": skor,
}
Jalankan evaluasi dengan evaluator kustom
results = evaluate(
prediksijawaban,
data=namadataset,
evaluators=[evaluatorkemiripansemantik, evaluatorrelevansijawaban],
experimentprefix="gpt4o-eval-kustom",
maxconcurrency=2,
)
Versioning Prompt
Mengelola Prompt di LangSmith Hub
from langchain import hub
from langchaincore.prompts import ChatPromptTemplate
Buat template prompt
qaprompt = ChatPromptTemplate.frommessages([
("system", "Anda adalah asisten ahli. Jawab pertanyaan dengan akurat dan ringkas. "
"Jika Anda tidak tahu jawabannya, katakan demikian."),
("human", "Konteks: {context}\n\nPertanyaan: {question}")
])
Unggah prompt ke LangSmith Hub
hub.push("org-saya/qa-prompt", qaprompt, newrepoispublic=False)
print("Prompt berhasil diunggah ke LangSmith Hub.")
Tarik versi tertentu
promptv1 = hub.pull("org-saya/qa-prompt")
print(f"Prompt ditarik dengan {len(promptv1.messages)} pesan.")
Perbarui prompt (otomatis membuat versi baru)
qapromptv2 = ChatPromptTemplate.frommessages([
("system", "Anda adalah asisten ahli yang berspesialisasi dalam topik teknologi. "
"Berikan jawaban yang detail dan terstruktur dengan baik beserta contoh bila memungkinkan. "
"Jika tidak yakin, nyatakan tingkat keyakinan Anda."),
("human", "Konteks: {context}\n\nPertanyaan: {question}")
])
hub.push("org-saya/qa-prompt", qapromptv2)
print("Prompt v2 diunggah. Versi sebelumnya tersimpan dalam riwayat.")
Monitoring Produksi
Menyiapkan Aturan Monitoring
from langsmith import Client
client = Client()
Query run terbaru untuk monitoring
runs = client.listruns(
projectname="tutorial-langsmith-demo",
executionorder=1, # Hanya run tingkat atas
error=False,
limit=100,
)
Analisis latensi dan penggunaan token
latensilist = []
totaltokenlist = []
for run in runs:
if run.endtime and run.starttime:
latensi = (run.endtime - run.starttime).totalseconds()
latensilist.append(latensi)
if run.totaltokens:
totaltokenlist.append(run.totaltokens)
if latensilist:
rataratalatensi = sum(latensilist) / len(latensilist)
makslatensi = max(latensilist)
p95latensi = sorted(latensilist)[int(len(latensilist) 0.95)]
print(f"Rata-rata latensi: {rataratalatensi:.2f} detik")
print(f"Latensi P95: {p95latensi:.2f} detik")
print(f"Latensi maksimum: {makslatensi:.2f} detik")
if totaltokenlist:
rataratatoken = sum(totaltokenlist) / len(totaltokenlist)
print(f"Rata-rata token per panggilan: {rataratatoken:.0f}")
Peringatan Otomatis
@traceable(name="pipelineterpantau")
def pipeline
terpantau(pertanyaan: str) -> dict:
"""Pipeline dengan monitoring bawaan."""
import time
mulai = time.time()
llm = ChatOpenAI(model="gpt-4o", temperature=0)
response = llm.invoke([HumanMessage(content=pertanyaan)])
waktueksekusi = time.time() - mulai
hasil = {
"jawaban": response.content,
"latensidetik": waktueksekusi,
"jumlahtoken": response.usagemetadata.get("totaltokens", 0)
if hasattr(response, "usagemetadata") and response.usagemetadata
else 0,
}
# Peringatan jika latensi melebihi batas
if waktueksekusi > 10.0:
print(f"[PERINGATAN] Latensi tinggi terdeteksi: {waktueksekusi:.2f}s "
f"untuk pertanyaan: {pertanyaan[:50]}...")
return hasil
Debugging Chain dan Agent
Memeriksa Run yang Gagal
# Ambil run yang menghasilkan error
runerror = client.listruns(
projectname="tutorial-langsmith-demo",
error=True,
limit=20,
)
for run in runerror:
print(f"ID Run: {run.id}")
print(f" Nama: {run.name}")
print(f" Error: {run.error}")
print(f" Input: {run.inputs}")
print(f" Mulai: {run.starttime}")
print("---")
Debugging Eksekusi Agent Langkah demi Langkah
from langchain.agents import AgentExecutor, createopenaitoolsagent
from langchaincore.tools import tool
from langchaincore.prompts import ChatPromptTemplate, MessagesPlaceholder
@tool
def hitung(ekspresi: str) -> str:
"""Mengevaluasi ekspresi matematika."""
try:
hasil = eval(ekspresi, {"builtins": {}}, {})
return str(hasil)
except Exception as e:
return f"Error: {e}"
@tool
def caripengetahuan(query: str) -> str:
"""Mencari informasi di basis pengetahuan."""
pengetahuan = {
"python": "Python adalah bahasa pemrograman serbaguna yang dibuat oleh Guido van Rossum.",
"javascript": "JavaScript adalah bahasa web yang berjalan di browser dan Node.js.",
}
for kunci, nilai in pengetahuan.items():
if kunci in query.lower():
return nilai
return "Tidak ditemukan informasi yang relevan."
prompt = ChatPromptTemplate.frommessages([
("system", "Anda adalah asisten yang membantu dengan akses ke alat-alat."),
MessagesPlaceholder(variablename="chathistory", optional=True),
("human", "{input}"),
MessagesPlaceholder(variablename="agentscratchpad"),
])
llm = ChatOpenAI(model="gpt-4o", temperature=0)
agent = createopenaitoolsagent(llm, [hitung, caripengetahuan], prompt)
agentexecutor = AgentExecutor(
agent=agent, tools=[hitung, caripengetahuan], verbose=True
)
Seluruh eksekusi dilacak - setiap panggilan tool, panggilan LLM, dan keputusan
hasil = agentexecutor.invoke(
{"input": "Berapa 15 23 + 7? Dan ceritakan tentang Python."}
)
print(hasil["output"])
Setelah eksekusi, periksa trace di UI LangSmith untuk melihat:
1. Penalaran agent di setiap langkah
2. Tool mana yang dipanggil beserta input/output-nya
3. Penggunaan token per langkah
4. Rincian latensi total
Pelacakan Biaya
Menghitung Biaya dari Run yang Dilacak
# Harga token (contoh tarif, sesuaikan dengan harga terkini)
HARGA = {
"gpt-4o": {"input": 2.50 / 1000000, "output": 10.00 / 1000000},
"gpt-4o-mini": {"input": 0.15 / 1000000, "output": 0.60 / 1000000},
"gpt-3.5-turbo": {"input": 0.50 / 1000000, "output": 1.50 / 1000000},
}
def hitungbiayaproyek(namaproyek: str, hari: int = 7) -> dict:
"""Menghitung biaya proyek selama jumlah hari yang ditentukan."""
from datetime import datetime, timedelta
client = Client()
tanggalmulai = datetime.now() - timedelta(days=hari)
runs = client.listruns(
projectname=namaproyek,
runtype="llm",
starttime=tanggalmulai,
)
totalbiaya = 0.0
biayapermodel = {}
for run in runs:
model = (run.extra.get("metadata", {}).get("lsmodelname", "tidak diketahui")
if run.extra else "tidak diketahui")
tokeninput = run.prompttokens or 0
tokenoutput = run.completiontokens or 0
if model in HARGA:
biaya = (tokeninput HARGA[model]["input"] +
tokenoutput * HARGA[model]["output"])
else:
biaya = 0.0
totalbiaya += biaya
if model not in biayapermodel:
biayapermodel[model] = {"biaya": 0.0, "panggilan": 0, "token": 0}
biayapermodel[model]["biaya"] += biaya
biayapermodel[model]["panggilan"] += 1
biayapermodel[model]["token"] += tokeninput + tokenoutput
print(f"Laporan biaya untuk '{namaproyek}' ({hari} hari terakhir):")
print(f" Total biaya: ${totalbiaya:.4f}")
for model, data in biayapermodel.items():
print(f" {model}: ${data['biaya']:.4f} "
f"({data['panggilan']} panggilan, {data['token']} token)")
return {"totalbiaya": totalbiaya, "biayapermodel": biayapermodel}
biaya = hitungbiayaproyek("tutorial-langsmith-demo", hari=30)
Integrasi dengan LangChain
Pipeline RAG Lengkap dengan Pelacakan LangSmith
from langchainopenai import ChatOpenAI, OpenAIEmbeddings
from langchaincore.prompts import ChatPromptTemplate
from langchaincore.outputparsers import StrOutputParser
Semua komponen LangChain secara otomatis dilacak
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
llm = ChatOpenAI(model="gpt-4o", temperature=0)
Simulasi retriever vector store
@traceable(name="pengambilvektor")
def ambil(query: str) -> str:
"""Simulasi pengambilan dari vector store."""
docs = {
"ai": "Kecerdasan buatan mencakup machine learning, deep learning, dan NLP.",
"ml": "Algoritma machine learning mempelajari pola dari data tanpa pemrograman eksplisit.",
}
hasil = []
for kunci, nilai in docs.items():
if kunci in query.lower():
hasil.append(nilai)
return "\n".join(hasil) if hasil else "Tidak ditemukan dokumen yang relevan."
prompt = ChatPromptTemplate.fromtemplate(
"Jawab pertanyaan berdasarkan konteks.\n\n"
"Konteks: {context}\n\n"
"Pertanyaan: {question}\n\n"
"Jawaban:"
)
Bangun chain - setiap langkah dilacak
chain = (
{"context": lambda x: ambil(x["question"]), "question": lambda x: x["question"]}
| prompt
| llm
| StrOutputParser()
)
Eksekusi dan lacak
jawaban = chain.invoke({"question": "Apa itu machine learning?"})
print(jawaban)
A/B Testing Prompt
Membandingkan Varian Prompt
from langsmith.evaluation import evaluate
Definisikan varian prompt
PROMPTA = "Jawab pertanyaan berikut secara ringkas: {query}"
PROMPTB = """Anda adalah asisten ahli. Berikan jawaban yang jelas dan terstruktur
untuk pertanyaan berikut. Sertakan contoh yang relevan bila membantu.
Pertanyaan: {query}
Jawaban:"""
def buatprediktor(templateprompt: str):
"""Membuat fungsi prediktor untuk template prompt tertentu."""
def prediksi(inputs: dict) -> dict:
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = templateprompt.format(query=inputs["query"])
response = llm.invoke([HumanMessage(content=prompt)])
return {"answer": response.content}
return prediksi
def evaluatorkualitas(run: Run, example: Example) -> dict:
"""Evaluasi kualitas jawaban menggunakan LLM sebagai juri."""
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prediksi = run.outputs.get("answer", "")
ekspektasi = example.outputs.get("answer", "")
evalprompt = f"""Bandingkan jawaban prediksi dengan jawaban referensi.
Nilai kualitasnya dari 0 sampai 1.
Referensi: {ekspektasi}
Prediksi: {prediksi}
Kembalikan hanya angka desimal antara 0 dan 1."""
response = llm.invoke([HumanMessage(content=evalprompt)])
try:
skor = float(response.content.strip())
except ValueError:
skor = 0.0
return {"key": "skorkualitas", "score": skor}
Jalankan A/B test
hasila = evaluate(
buatprediktor(PROMPTA),
data=namadataset,
evaluators=[evaluatorkualitas],
experimentprefix="varian-prompt-A",
)
hasilb = evaluate(
buatprediktor(PROMPTB),
data=namadataset,
evaluators=[evaluatorkualitas],
experimentprefix="varian-prompt-B",
)
print("A/B test selesai. Bandingkan hasil di UI LangSmith.")
print(f"Varian A: {hasila.experimenturl}")
print(f"Varian B: {hasilb.experimenturl}")
Praktik Terbaik
LANGCHAINPROJECT untuk mengorganisir trace berdasarkan aplikasi, lingkungan, atau tim. Hindari memasukkan semuanya ke proyek default.@traceable untuk logika kustom. Setiap fungsi yang melakukan pekerjaan penting (panggilan API, transformasi data, logika bisnis) harus dilacak untuk memberikan visibilitas penuh.tierpengguna:premium, fitur:pencarian) untuk memungkinkan analisis yang difilter.LANGCHAIN_PROJECT yang berbeda untuk dev, staging, dan produksi agar data tetap bersih.Kesimpulan
LangSmith mengubah pengembangan aplikasi LLM dari tebak-tebakan menjadi proses berbasis data. Dengan melacak setiap panggilan, mengevaluasi terhadap dataset yang dikurasi, mengelola versi prompt, dan memantau kinerja produksi, Anda mendapatkan kepercayaan diri yang dibutuhkan untuk mengirimkan dan memelihara aplikasi AI yang andal.
Poin utama yang perlu diingat adalah memperlakukan observabilitas sebagai perhatian utama sejak awal proyek Anda, bukan sebagai tambahan belakangan. Siapkan pelacakan di hari pertama, bangun dataset evaluasi saat Anda mengembangkan, dan tetapkan monitoring biaya sebelum Anda melakukan scaling. LangSmith menyediakan semua alat yang Anda butuhkan untuk membuat alur kerja ini berjalan lancar.
Mulailah dengan mengintegrasikan LangSmith ke proyek LangChain Anda yang sudah ada, bangun dataset evaluasi kecil dari pertanyaan pengguna nyata, dan iterasikan prompt Anda dengan data yang mendukung setiap keputusan.