Belajar LangChain - Output Parsers & Structured Output

#Pendahuluan

Di episode 5 kalian sudah merakit chain pertama dengan LCEL: chain = prompt | model | parser. Rantai itu bekerja karena setiap komponennya adalah Runnable, dan output parser adalah salah satu anggota penting dalam rantai tersebut. Kalian juga sudah mencicipi RunnableParallel dan RunnablePassthrough untuk memanipulasi aliran data antar komponen.

Episode ini fokus ke satu elemen yang sering diremehkan tapi menentukan kualitas aplikasi: output parser. Model bahasa mengembalikan teks bebas, sedangkan aplikasi kalian sering butuh data terstruktur. Kita akan membedah empat strategi: StrOutputParser untuk teks biasa, JsonOutputParser untuk JSON, PydanticOutputParser untuk objek yang divalidasi, dan with_structured_output yang menyerahkan tugas format langsung ke model. Terakhir, kita pelajari cara mem-parsing aliran streamed chunks.

#Kenapa Output Parser Penting

Tanpa parser, setiap respons model adalah string bebas yang harus kalian parsing sendiri — dan parsing string bebas itu rapuh. Output parser menambahkan lapisan kontrak: chain selalu mengembalikan tipe yang sama, sehingga kode pemanggil tidak perlu menebak-nebak.

Tanpa parser, output berupa string mentah

from langchain_openai import ChatOpenAI
 
model = ChatOpenAI(model="gpt-4o-mini")
hasil = model.invoke("Sebutkan tiga bahasa pemrograman populer")
print(hasil.content)

Ujung rantai inilah yang menjadi pembeda utama setiap parser. Ada empat keluarga yang akan kita pakai terus-menerus: StrOutputParser mengubah AIMessage menjadi str, JsonOutputParser mengubah string JSON menjadi dict, PydanticOutputParser mengubah JSON menjadi instance BaseModel, dan with_structured_output membuat model memproduksi JSON sesuai skema yang diminta.

#StrOutputParser

Parser paling sederhana dan paling banyak dipakai. Ia mengambil konten dari output model — biasanya AIMessage — lalu mengembalikannya sebagai string murni.

Chain dengan StrOutputParser

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "Jawab dalam bahasa Indonesia, maksimal dua kalimat."),
    ("human", "{pertanyaan}"),
])
model = ChatOpenAI(model="gpt-4o-mini")
 
chain = prompt | model | StrOutputParser()
jawaban = chain.invoke({"pertanyaan": "Apa itu vector database?"})
print(jawaban)

Perhatikan bahwa chain.invoke() kini mengembalikan str, bukan AIMessage. Ini menyederhanakan kode pemanggil: fungsi lain bisa langsung mengonsumsi jawaban tanpa tahu-menahu soal struktur pesan. StrOutputParser juga bisa diletakkan di ujung chain yang lebih kompleks, misalnya setelah RunnableParallel yang hasilnya ingin digabung jadi satu teks.

#JsonOutputParser

Saat kalian butuh data terstruktur seperti key-value pairs, JsonOutputParser adalah pilihan praktis. Keunggulannya: ia tahu cara menangani JSON parsial yang keluar saat streaming, karena parser ini menjalankan logika per-karakter menggunakan parser internal yang toleran terhadap token yang belum lengkap.

JsonOutputParser untuk data terstruktur

from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate
 
parser = JsonOutputParser()
prompt = PromptTemplate(
    template="Jawab dalam JSON. {format_instructions}\nPertanyaan: {pertanyaan}",
    input_variables=["pertanyaan"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)
model = ChatOpenAI(model="gpt-4o-mini")
 
chain = prompt | model | parser
data = chain.invoke({"pertanyaan": "Data ibu kota Indonesia"})
print(data["ibu_kota"])

get_format_instructions() menyuntikkan instruksi format ke dalam prompt sehingga model tahu persis struktur JSON yang diharapkan. Dalam contoh di atas, hasilnya adalah dict Python sehingga kalian bisa langsung mengakses data["ibu_kota"] tanpa regex atau string slicing.

#PydanticOutputParser

Untuk kontrak yang lebih ketat, gunakan PydanticOutputParser. Ia menggabungkan JSON parsing dengan validasi Pydantic: skema output diturunkan dari definisi BaseModel, dan field yang salah tipe langsung menimbulkan ValidationError.

Output terstruktur dengan Pydantic

from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import PromptTemplate
from pydantic import BaseModel, Field
 
class Ringkasan(BaseModel):
    judul: str = Field(description="Judul artikel")
    poin_utama: list[str] = Field(description="Poin-poin penting")
 
parser = PydanticOutputParser(pydantic_object=Ringkasan)
prompt = PromptTemplate(
    template="Ringkas teks berikut dalam JSON. {format_instructions}\nTeks: {teks}",
    input_variables=["teks"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)
model = ChatOpenAI(model="gpt-4o-mini")
 
chain = prompt | model | parser
hasil = chain.invoke({"teks": "LangChain adalah framework untuk membangun aplikasi LLM..."})
print(hasil.judul)
print(hasil.poin_utama)

Kelebihan utama di sini adalah type safety: hasil.judul sudah terjamin bertipe str dan hasil.poin_utama bertipe list[str]. Kesalahan ketik di JSON output langsung ketahuan saat validasi, bukan di tengah-tengah logika aplikasi.

#with_structured_output

Pendekatan paling modern: bukannya mem-parsing setelah respons tiba, kita minta model mengembalikan format tertentu sejak awal. with_structured_output memetakan skema ke tool calling atau response format bawaan provider, sehingga hasilnya lebih konsisten daripada mengandalkan instruksi teks di prompt.

with_structured_output dengan dataclass

from dataclasses import dataclass
from langchain_openai import ChatOpenAI
 
@dataclass
class Klasifikasi:
    label: str
    skor: float
 
model = ChatOpenAI(model="gpt-4o-mini")
model_terstruktur = model.with_structured_output(Klasifikasi)
 
hasil = model_terstruktur.invoke(
    "Teks ini tentang database vector: skor sentimen dan label temanya"
)
print(hasil.label, hasil.skor)

with_structured_output menerima kelas Pydantic, dataclass, maupun TypedDict. Bila diberi dataclass, LangChain melakukan koersi otomatis ke tipe Python yang kalian definisikan. Hasilnya bukan JSON mentah, melainkan objek yang sudah siap dipakai di kode domain.

#Streaming Parsing

Saat mode streaming, output datang per bagian. Parser berbasis teks seperti StrOutputParser bisa langsung dipakai; sedangkan untuk JSON, JsonOutputParser menghasilkan partial object yang masih valid saat diurai sebagian. Ini penting untuk pengalaman pengguna yang responsif.

Parsing streamed chunks

chain = prompt | model | StrOutputParser()
 
async for potongan in chain.astream({"pertanyaan": "Jelaskan RAG secara singkat"}):
    print(potongan, end="", flush=True)

Karena StrOutputParser hanya meneruskan teks, setiap potongan di atas adalah potongan string yang bisa langsung dirender. Untuk JsonOutputParser, chain.astream() akan menghasilkan dictionary parsial yang semakin lengkap seiring aliran data berjalan — cukup render field yang sudah tersedia. Untuk parser berbasis instruksi seperti PydanticOutputParser, streaming penuh belum didukung semua versi, jadi lebih aman invoke biasa atau stream_events untuk memantau tahapan tanpa mengorbankan validasi.

#Penutup

Episode ini mengubah output model dari teks bebas menjadi kontrak terstruktur: StrOutputParser untuk string, JsonOutputParser untuk dict, PydanticOutputParser untuk objek tervalidasi, dan with_structured_output untuk koersi langsung ke tipe Python. Kalian juga memahami cara menangani streamed chunks tanpa merusak pipeline.

Inti yang harus dibawa pulang:

• StrOutputParser mengubah AIMessage menjadi str — titik akhir paling umum dalam chain.
• JsonOutputParser cocok untuk JSON cepat dan toleran terhadap partial output saat streaming.
• PydanticOutputParser memberikan validasi ketat lewat skema BaseModel dan instruksi format otomatis.
• with_structured_output menyerahkan pembentukan JSON ke model lewat tool calling atau response format, paling konsisten untuk produksi.
• Streaming parsing berjalan mulus untuk parser berbasis teks dan JSON; parser berbasis instruksi lebih aman dengan invoke.

Di episode 7 berikutnya kita berurusan dengan ingatan: Memory & Conversation State — menyimpan riwayat percakapan, memakai thread_id, dan menggunakan LangGraph checkpoints agar agent kalian ingat siapa yang sedang diajak bicara. Sampai jumpa!