Membangun Agen LLM yang Type-Safe dengan PydanticAI

PydanticAI adalah framework agen dari tim di balik Pydantic, dirancang untuk membawa pengalaman pengembang yang sama seperti yang dibawa FastAPI ke dunia web API ke dalam ranah AI generatif. Tutorial ini membahas konsep intinya: agen yang type-safe, output terstruktur, dependency injection, dan tools, diakhiri dengan agen customer-support lengkap yang bisa Anda sesuaikan untuk proyek Anda sendiri.

Apa Itu PydanticAI dan Mengapa Dibuat

Tim Pydantic membangun PydanticAI karena sebagian besar ekosistem AI Python entah menemukan ulang validasi dengan buruk atau membungkus panggilan LLM dalam dictionary yang minim tipe. PydanticAI mengambil sikap berbeda: bersandar pada sistem tipe, memvalidasi segalanya, dan menjaga API tetap cukup ringkas untuk dipahami.

Beberapa prinsip desain membentuk framework ini:

• Type-safe secara bawaan. Agen mendeklarasikan tipe dependensi dan tipe output sebagai parameter generik. Pemeriksa statis seperti mypy dan Pyright memahaminya, dan editor Anda memberikan autocomplete yang sesuai.
• Model-agnostic. OpenAI, Anthropic, Google (Gemini), Groq, Mistral, Ollama, dan lainnya didukung di balik satu antarmuka Agent. Berpindah model biasanya cukup perubahan satu baris.
• Ergonomi ala FastAPI. Dependency injection, dekorator untuk tools dan validator, serta pemisahan yang rapi antara mendefinisikan agen dan menjalankannya akan terasa akrab jika Anda pernah memakai FastAPI.
• Dibangun di atas Pydantic. Output terstruktur adalah model Pydantic biasa, sehingga Anda mendapatkan validasi, pembuatan JSON schema, dan serialisasi secara gratis.
• Observability untuk produksi. Integrasi kelas satu dengan Logfire memberi Anda tracing dan debugging tanpa perlu memasang framework terpisah.

PydanticAI sengaja bukan platform orkestrasi serba bisa. Ia fokus pada loop agen, sehingga Anda bebas mengomposisikan agen dengan kode aplikasi Anda sendiri.

Instalasi

PydanticAI membutuhkan Python 3.9 atau lebih baru. Instal dengan pip:

pip install pydantic-ai

Paket dasar menyertakan dukungan untuk penyedia model utama. Jika menginginkan instalasi yang lebih ramping, Anda bisa memilih hanya penyedia yang Anda butuhkan:

pip install "pydantic-ai-slim[openai]"
pip install "pydantic-ai-slim[anthropic]"
pip install "pydantic-ai-slim[google]"

Jika Anda berencana memakai observability, instal juga ekstra Logfire:

pip install "pydantic-ai[logfire]"

Sebagian besar penyedia membaca API key-nya dari variabel lingkungan:

export OPENAIAPIKEY="sk-..."
export ANTHROPICAPIKEY="sk-ant-..."
export GEMINIAPIKEY="..."

Agen Pertama Anda

from pydanticai import Agent

agent = Agent(
    "openai:gpt-4o",
    systemprompt="Anda asisten yang ringkas. Jawab dalam satu atau dua kalimat.",
)

result = agent.runsync("Apa ibu kota Indonesia?")

print(result.output)
Jakarta adalah ibu kota Indonesia.

agent = Agent("anthropic:claude-3-5-sonnet-latest")
agent = Agent("google-gla:gemini-1.5-flash")

import asyncio

Sinkron - praktis di skrip dan notebook
result = agent.runsync("Halo")
print(result.output)

Asinkron - untuk aplikasi async dan konkurensi
async def main():
    result = await agent.run("Halo")
    print(result.output)

asyncio.run(main())

Streaming - menerima output bertahap saat dihasilkan
async def streamexample():

    async with agent.runstream("Ceritakan kisah pendek") as response:
        async for chunk in response.streamtext(delta=True):

            print(chunk, end="", flush=True)

asyncio.run(streamexample())

Ketiganya mengembalikan (atau menghasilkan) objek result yang kaya dan sama, yang membawa output, riwayat pesan, dan penggunaan token.

Output Terstruktur dan Bertipe

Teks bebas cocok untuk chat, tetapi aplikasi biasanya membutuhkan data terstruktur. PydanticAI memungkinkan Anda mendeklarasikan outputtype sebagai model Pydantic. Framework menginstruksikan model untuk menghasilkan data yang sesuai dan memvalidasi respons sebelum mengembalikannya.

from pydantic import BaseModel, Field
from pydanticai import Agent


class CityInfo(BaseModel):
    name: str
    country: str
    population: int = Field(description="Perkiraan populasi")
    iscapital: bool



agent = Agent(
    "openai:gpt-4o",
    outputtype=CityInfo,
    systemprompt="Ekstrak informasi terstruktur tentang kota yang disebut pengguna.",

)

result = agent.runsync("Ceritakan tentang Bandung")
print(result.output)
name='Bandung' country='Indonesia' population=2500000 iscapital=False

print(type(result.output))
main.CityInfo'>

from typing import Union

class Success(BaseModel):
    answer: str

class Clarification(BaseModel):
    question: str

agent = Agent(
    "openai:gpt-4o",
    outputtype=Union[Success, Clarification],
)

Model memilih bentuk yang sesuai, dan hasilnya bertipe union sehingga Anda bisa bercabang dengan aman.

Dependency Injection

Agen nyata membutuhkan akses ke basis data, klien HTTP, konfigurasi, atau identitas pengguna saat ini. Alih-alih menggunakan variabel global, PydanticAI memakai dependency injection. Anda mendeklarasikan depstype, biasanya sebuah dataclass, dan mengoper instance saat menjalankan agen. Tools dan system prompt kemudian menerimanya melalui RunContext.

from dataclasses import dataclass
import httpx
from pydanticai import Agent, RunContext


@dataclass
class SupportDeps:
    customerid: int

    db: "DatabaseConn"
    httpclient: httpx.AsyncClient


agent = Agent(
    "openai:gpt-4o",
    depstype=SupportDeps,

    systemprompt="Anda agen dukungan untuk toko online.",
)

• @agent.tool untuk tools yang membutuhkan akses ke dependensi. Parameter pertama berupa RunContext.
• @agent.toolplain untuk tools yang tidak membutuhkan konteks.

PydanticAI membangun JSON schema untuk setiap tool dari type hint dan docstring-nya, sehingga model tahu bagaimana dan kapan harus memanggilnya.

from pydanticai import Agent, RunContext


@agent.tool
async def getcustomername(ctx: RunContext[SupportDeps]) -> str:

    """Mengembalikan nama lengkap pelanggan saat ini."""
    return await ctx.deps.db.customername(id=ctx.deps.customerid)



@agent.tool
async def getbalance(ctx: RunContext[SupportDeps], includepending: bool) -> float:

    """Mengembalikan saldo akun pelanggan.

    Args:
        includepending: Apakah menyertakan transaksi yang masih tertunda.
    """
    return await ctx.deps.db.customerbalance(

        id=ctx.deps.customerid,
        includepending=includepending,
    )


@agent.toolplain

def currentexchangerate(base: str, quote: str) -> float:

    """Mengembalikan kurs demo statis antara dua mata uang."""
    rates = {("USD", "IDR"): 16250.0, ("EUR", "IDR"): 17600.0}
    return rates.get((base, quote), 1.0)

Deskripsi docstring dan bagian Args diekstrak secara otomatis. Tipe parameter divalidasi saat model memanggil tool, sehingga tool yang dideklarasikan dengan includepending: bool tidak akan pernah menerima string.

System Prompt Dinamis

System prompt statis ditetapkan saat pembuatan agen. Ketika prompt bergantung pada data runtime, daftarkan prompt dinamis dengan @agent.systemprompt. Ia berjalan pada setiap eksekusi dan dapat membaca dependensi.

@agent.systemprompt
async def addcustomercontext(ctx: RunContext[SupportDeps]) -> str:
    name = await ctx.deps.db.customername(id=ctx.deps.customerid)
    return f"Nama pelanggan adalah {name}. Sapa dengan sopan menggunakan namanya."

System prompt statis dan dinamis digabungkan, dengan yang dinamis dievaluasi saat run. Anda bisa mendaftarkan beberapa; semuanya digabung sesuai urutan pendaftaran.

Output Validator

Terkadang memvalidasi bentuk output saja tidak cukup; Anda perlu memvalidasi isinya terhadap aturan bisnis atau sistem eksternal. Dekorator @agent.outputvalidator berjalan setelah model menghasilkan output terstruktur. Jika Anda memunculkan ModelRetry, PydanticAI mengirim balik error tersebut ke model dan memintanya mencoba lagi.

from pydanticai import Agent, ModelRetry, RunContext


@agent.outputvalidator

async def validatesupportoutput(

    ctx: RunContext[SupportDeps], output: "SupportResult"
) -> "SupportResult":
    if output.blockcard and output.risk < 5:
        raise ModelRetry(
            "Anda menetapkan blockcard=True padahal risiko rendah. "

            "Blokir kartu hanya ketika risiko 5 atau lebih tinggi."
        )
    return output

Setiap panggilan ke run, runsync, atau runstream bersifat stateless secara bawaan. Untuk melanjutkan percakapan, oper pesan sebelumnya melalui messagehistory.

agent = Agent("openai:gpt-4o", systemprompt="Anda tutor yang membantu.")

first = agent.runsync("Nama saya Ruby dan saya sedang belajar Python.")
print(first.output)

second = agent.runsync(

    "Tadi nama saya siapa?",
    messagehistory=first.newmessages(),

)
print(second.output)
Nama Anda Ruby.

Gunakan result.allmessages() untuk mendapatkan riwayat lengkap termasuk system prompt, atau result.newmessages() untuk hanya pesan yang dihasilkan run terakhir. Pesan dapat diserialkan ke JSON untuk disimpan antar permintaan dalam aplikasi web.

from pydantic import BaseModel
from pydanticai import Agent


class Report(BaseModel):
    title: str
    summary: str
    bulletpoints: list[str]



agent = Agent("openai:gpt-4o", outputtype=Report)


async def streamreport():

    async with agent.runstream("Ringkas data penjualan kuartalan") as response:
        async for partial in response.stream():
            print(partial)
        final = await response.getoutput()

        print("Final:", final)

from pydanticai import models
from pydanticai.models.test import TestModel


Cegah permintaan nyata yang tidak disengaja di seluruh suite tes
models.ALLOWMODELREQUESTS = False



def testsupportagent():

    deps = SupportDeps(customerid=1, db=FakeDB(), httpclient=FakeClient())

    with agent.override(model=TestModel()):
        result = agent.runsync("Apakah kartu saya diblokir?", deps=deps)
    assert isinstance(result.output, SupportResult)

Untuk kendali presisi, FunctionModel menerima riwayat pesan dan tools yang tersedia, lalu Anda mengembalikan respons model yang persis:

from pydanticai.messages import ModelResponse, TextPart
from pydanticai.models.function import FunctionModel, AgentInfo


def mymodellogic(messages, info: AgentInfo) -> ModelResponse:
    return ModelResponse(parts=[TextPart("Balasan yang ditulis manual")])


def testwithfunctionmodel():

    with agent.override(model=FunctionModel(mymodellogic)):

        result = agent.runsync("Apa saja")
    assert result.output == "Balasan yang ditulis manual"

Pendekatan ini memberikan pengujian logika agen, tools, dan validator yang deterministik, cepat, dan tanpa jaringan.

Observability dengan Logfire

Men-debug agen yang melakukan beberapa panggilan model dan invokasi tool per permintaan akan terbantu oleh tracing. PydanticAI berintegrasi dengan Logfire, juga dari tim Pydantic, untuk menangkap setiap langkah.

import logfire

logfire.configure()
logfire.instrumentpydanticai()

Semua run agen setelah ini di-trace secara otomatis
result = agent.runsync("Halo")

from dataclasses import dataclass
from pydantic import BaseModel, Field
from pydanticai import Agent, ModelRetry, RunContext


Basis data tiruan kecil agar contoh tetap mandiri
class DatabaseConn:
    names = {1: "Ruby Abdullah"}

    balances = {1: 1250.0}

    async def customername(self, id: int) -> str:

        return self.names[id]

    async def customerbalance(self, id: int, includepending: bool) -> float:
        balance = self.balances[id]

        return balance - 50.0 if includepending else balance


@dataclass
class SupportDeps:
    customerid: int

    db: DatabaseConn


class SupportResult(BaseModel):
    advice: str = Field(description="Saran yang diberikan kepada pelanggan")
    blockcard: bool = Field(description="Apakah perlu memblokir kartu pelanggan")
    risk: int = Field(ge=0, le=10, description="Tingkat risiko dari 0 hingga 10")


supportagent = Agent(

    "openai:gpt-4o",
    depstype=SupportDeps,
    outputtype=SupportResult,

    systemprompt=(
        "Anda agen dukungan untuk sebuah bank. Bantu pelanggan dengan pertanyaannya "
        "dan nilai tingkat risiko dari permintaannya."
    ),
)


@supportagent.systemprompt
async def addcustomername(ctx: RunContext[SupportDeps]) -> str:
    name = await ctx.deps.db.customername(id=ctx.deps.customerid)
    return f"Nama pelanggan adalah {name!r}."


@supportagent.tool

async def customerbalance(
    ctx: RunContext[SupportDeps], includepending: bool

) -> str:
    """Mengembalikan saldo akun pelanggan saat ini.

    Args:
        includepending: Apakah menyertakan transaksi tertunda dalam angka.
    """
    balance = await ctx.deps.db.customerbalance(

        id=ctx.deps.customerid,
        includepending=includepending,
    )
    return f"Rp{balance:.2f}"


@supportagent.outputvalidator
async def validaterisk(

    ctx: RunContext[SupportDeps], output: SupportResult
) -> SupportResult:
    if output.blockcard and output.risk < 5:
        raise ModelRetry(
            "Jangan blokir kartu kecuali tingkat risiko 5 atau lebih tinggi."
        )
    return output


async def main():
    deps = SupportDeps(customerid=1, db=DatabaseConn())


    result = await supportagent.run("Berapa saldo saya?", deps=deps)
    print(result.output)
    # advice='Saldo Anda saat ini Rp1250.00.' blockcard=False risk=1


    followup = await supportagent.run(
        "Sepertinya kartu saya dicuri, tolong bantu.",
        deps=deps,
        messagehistory=result.newmessages(),
    )
    print(followup.output)
    # advice='Saya akan segera memblokir kartu Anda...' blockcard=True risk=8



import asyncio
asyncio.run(main())

Agen tunggal ini memperlihatkan gambaran utuh: dependency injection menyediakan basis data dan pelanggan, system prompt dinamis mempersonalisasi respons, tool mengambil data langsung, output validator menegakkan aturan bisnis, dan messagehistory menjaga percakapan tetap koheren antar run.

Praktik Terbaik

• Biarkan sistem tipe bekerja untuk Anda. Selalu beri anotasi depstype dan outputtype. Jaminan statis menangkap kesalahan sebelum runtime dan meningkatkan dukungan editor.
• Utamakan dependency injection ketimbang variabel global. Mengoper dataclass berisi dependensi menjaga agen tetap mudah diuji dan membuat aliran data eksplisit.
• Tulis docstring tool yang jelas. Model mengandalkan docstring dan type hint untuk memutuskan kapan dan bagaimana memanggil tool. Perlakukan keduanya sebagai bagian dari prompt.
• Gunakan output validator untuk aturan bisnis. Validasi skema memeriksa bentuk; output validator memeriksa makna. Memunculkan ModelRetry memungkinkan model pulih dengan baik.
• Tetapkan batas percobaan ulang yang masuk akal. Percobaan ulang otomatis bermanfaat tetapi dapat melipatgandakan biaya. Konfigurasikan retries pada agen atau per tool untuk membatasinya.
• Uji dengan TestModel dan FunctionModel. Setel ALLOWMODELREQUESTS = False di suite tes Anda untuk menjamin tidak ada panggilan jaringan yang tidak disengaja.
• Pasang instrumentasi sejak awal. Menambahkan Logfire lebih awal membuat diagnosis perilaku model yang aneh jauh lebih mudah daripada memasang logging belakangan.
• Jaga agen tetap fokus. Komposisikan beberapa agen kecil bertujuan tunggal ketimbang membangun satu agen monolitik dengan lusinan tool.

Kesimpulan dan Poin Penting

PydanticAI membawa pendekatan yang disiplin dan mengutamakan tipe untuk membangun agen LLM. Dengan bersandar pada Pydantic untuk validasi dan mengadopsi dependency injection bergaya FastAPI, ia memungkinkan Anda membangun agen yang dapat diprediksi, mudah diuji, dan mudah dipahami.

Poin penting:

• Sebuah Agent menggabungkan model, system prompt, tipe dependensi, dan tipe output menjadi satu objek bertipe.
• Output terstruktur dengan outputtype memberi Anda model Pydantic tervalidasi alih-alih teks mentah.
• Dependency injection melalui depstype dan RunContext menjaga agen tetap bersih dan mudah diuji.
• Tools (@agent.tool, @agent.toolplain), system prompt dinamis, dan output validator (ModelRetry) mencakup pola umum aplikasi nyata.
• messagehistory, streaming, TestModel/FunctionModel, dan Logfire melengkapi jalur dari prototipe hingga produksi.

Mulailah dengan satu agen dan satu output bertipe, lalu tambahkan tools dan dependensi seiring aplikasi Anda berkembang. Framework ini menghargai pendekatan bertahap yang dipandu oleh tipe.