Integrasi AI Agents ke Django dengan PydanticAI: 4 Pola Praktis

Empat pola integrasi AI agents ke Django pakai PydanticAI: streaming endpoint, persistent memory, service-layer tools, dan model-aware agents - plus pertimbangan keamanan untuk production.

·8 min read·1,598 words
pythondjangopydantic-aiai-agentstutorial

Kamu udah baca pengantar PydanticAI dan tertarik untuk integrasi ke Django. Sekarang muncul pertanyaan: gimana caranya biar agent LLM bisa jadi bagian natural dari Django app kamu - bukan cuma API wrapper yang dipanggil dari view?

Post ini bukan tentang “cara chat dengan GPT dari Django.” Ada banyak tutorial untuk itu. Ini tentang empat pola integrasi yang sudah teruji di production: dari streaming chat endpoint sampai background agent loop yang jalan tanpa blocking request-response cycle.

Semua pola pake PydanticAI sebagai framework agent. Kenapa? Karena type safety-nya cocok sama filosofi Django - explicit lebih baik dari implicit, validasi di boundary, dan schema yang bisa di-compose.

Sebelum lanjut, pastikan kamu familiar dengan Django async support - karena semua pola di sini asynchronous.

Pattern 1: Streaming Chat Endpoint dengan django-bolt

Ini pola paling sederhana: user ngirim prompt, agent ngasih response lewat HTTP streaming. Cocok buat chatbot, AI assistant di dashboard, atau live Q&A.

Setup

pip install django-bolt pydantic-ai

Tambah django_bolt ke INSTALLED_APPS:

INSTALLED_APPS = [
    ...,
    "django_bolt",
]

Endpoint

from django_bolt import BoltAPI
from django_bolt.responses import StreamingResponse
from pydantic_ai import Agent

api = BoltAPI()
agent = Agent(
    "anthropic:claude-haiku-4-5",
    instructions="Be concise, reply with one sentence."
)


@api.get("/chat")
async def chat(prompt: str = "Where does hello world come from?"):
    async def content():
        async with agent.run_stream(prompt) as result:
            async for chunk in result.stream_text(delta=True):
                yield chunk

    return StreamingResponse(content(), media_type="text/plain")

Yang terjadi di balik layar: agent.run_stream membuka koneksi ke Anthropic, dan setiap chunk teks langsung di-forward ke client sebagai SSE (Server-Sent Events). StreamingResponse dari django-bolt handle koneksi tetap terbuka tanpa blocking event loop Django.

Kenapa ini penting? Kalau kamu pake response biasa, user nunggu seluruh response selesai - bisa 10-30 detik buat output panjang - baru bisa lihat hasil. Dengan streaming, user lihat teks muncul kata per kata, dan UX-nya jauh lebih responsif.

Production considerations

  • Tambah rate limiting per user - jangan sampai satu user bisa trigger 100 stream concurrent
  • Set timeout per stream (biasanya 60 detik cukup)
  • Simpan conversation_id di session biar ada konteks percakapan

Pattern 2: Agent Memory di Django dengan django-pydantic-ai

Streaming endpoint keren, tapi agent tanpa memory itu kayak customer service yang lupa omongan kamu 5 detik lalu. Setiap chat dimulai dari awal.

django-pydantic-ai (v6.1.0a1) punya DjangoModelStore - persistent memory yang disimpan di database Django. Agent bisa ingat percakapan sebelumnya, preferensi user, bahkan state dari interaksi multi-step.

Setup

pip install django-pydantic-ai
from pydantic_ai import Agent
from pydantic_ai_harness.memory import Memory
from django_pydantic_ai.memory.models import Directory
from django_pydantic_ai.memory.store import DjangoModelStore


async def create_agent(model: str, user_id: int) -> Agent:
    directory, _ = await Directory.objects.aget_or_create(
        name=f"user:{user_id}"
    )
    store = DjangoModelStore(directory)
    return Agent(
        model,
        capabilities=[Memory(store=store)]
    )

Setiap user punya Directory sendiri di database - memory-nya terisolasi. DjangoModelStore otomatis serialize/deserialize conversation history ke model Django. Jadi kalau server restart, memory tetap ada.

Pola ini cocok buat:

  • Customer support agent yang perlu ingat riwayat tiket user
  • Onboarding wizard multi-step yang state-nya disimpan per user
  • AI tutor yang tracking progress belajar

Yang perlu diperhatikan: memory punya context window. PydanticAI handle truncation otomatis - history lama di-summarize atau di-drop kalau udah exceed limit model. Tapi kamu tetap harus punya strategi buat jangka panjang: archive percakapan lama ke tabel terpisah setelah N messages.

Pattern 3: Django Service Layer sebagai Agent Tools

Ini pola yang paling sering dipake di production Django apps. Agent tidak cuma ngobrol - dia bisa melakukan sesuatu: baca data, buat record, kirim notifikasi.

Ada dua cara: pake djangorestframework-pydantic-ai yang otomatis expose DRF ServiceSpec/SelectorSpec sebagai tools, atau pake @tool decorator biasa.

Cara A: DRF ServiceSpec sebagai tools

from pydantic_ai import Agent
from rest_framework_pydantic_ai import AgentDeps, SpecToolset

toolset = SpecToolset({
    "list_orders": orders_selector_spec,
    "create_order": create_order_spec,
})

agent = Agent(
    "anthropic:claude-opus-4-8",
    deps_type=AgentDeps,
    toolsets=[toolset],
)

Dengan cara ini, setiap ServiceSpec atau SelectorSpec dari DRF kamu otomatis jadi tool yang bisa dipanggil agent. Schema-nya di-generate dari type hints DRF - jadi agent tahu parameter apa yang dibutuhkan dan return type-nya.

Cara B: Plain @tool dengan Django ORM

Kadang kamu perlu kontrol lebih granular. Pake @tool decorator:

from pydantic_ai import Agent, RunContext
from dataclasses import dataclass
from django.contrib.auth.models import User
from support.models import Ticket


@dataclass
class AgentDeps:
    user_id: int


agent = Agent(
    "anthropic:claude-haiku-4-5",
    deps_type=AgentDeps,
)


@agent.tool
async def get_open_tickets(
    ctx: RunContext[AgentDeps], limit: int = 20
) -> list[dict]:
    return list(
        Ticket.objects
        .filter(assigned_to_id=ctx.deps.user_id, status="open")
        .values("id", "subject", "created_at")[:limit]
    )

Ini krusial: setiap tool harus filter berdasarkan user yang sedang login. Jangan sampai agent bisa baca tiket user lain cuma karena tool-nya query tanpa filter. AgentDeps adalah mekanisme PydanticAI untuk inject dependency - di sini kita pass user_id dari request user, dan setiap tool pake itu sebagai basis filter.

Contoh penggunaan di view

from django.http import JsonResponse
from pydantic_ai import Agent


async def agent_view(request):
    deps = AgentDeps(user_id=request.user.id)
    result = await agent.run(
        "Apa tiket open yang paling lama belum direspon?",
        deps=deps,
    )
    return JsonResponse({"response": result.data})

Kapan pake yang mana?

Situasi Rekomendasi
Udah punya DRF ServiceSpec/SelectorSpec DRF toolset - zero extra code
Tool-nya sederhana (1-2 query) @tool decorator
Butuh validasi input kompleks DRF toolset (dapet validasi gratis)
Tool panggil API eksternal @tool - lebih fleksibel

Pattern 4: Model-Aware Agents dengan django-model-agent

Kadang kamu butuh agent yang melekat pada satu instance model. Misalnya: review setiap Place listing yang statusnya draft, generate summary otomatis, atau validasi kelengkapan data.

django-model-agent memungkinkan kamu bikin agent per model row:

from django_model_agent import ModelAgent
from listings.models import Place
from listings.schemas import PlaceReview


class PlaceAgent(ModelAgent):
    model = Place
    fields = ["name", "address", "description"]
    output_type = PlaceReview
    _system_prompts = "Review this place listing for completeness."


async def review_draft_places():
    async for place in Place.objects.filter(state="draft"):
        result = await PlaceAgent(place).run("Summarise this place.")
        place.summary = result.output
        await place.asave(update_fields=["summary"])

Setiap PlaceAgent instance terikat ke satu Place row. Agent bisa akses field-field yang di-defined di fields, dan output-nya diverifikasi sama output_type (Pydantic model) - jadi kamu jamin place.summary selalu valid.

Pola ini berguna untuk:

  • Data enrichment: generate meta description, tags, atau SEO fields
  • Content moderation: scan user-generated content
  • Bulk validation: jalanin agent buat ribuan record di background
  • Auto-categorization: classify items berdasarkan deskripsi

Tapi perhatikan: ini N LLM calls buat N records. Kalau kamu punya 10.000 draft places, jangan jalanin sync di request thread. Pake background task.

Bonus: Background Agent Loop

Semua pola di atas bisa dipicu dari HTTP request. Tapi agent sering perlu jalan tanpa menunggu response - processing antrian, ngecek data baru, atau ngejalanin workflow multi-step.

Django punya task framework built-in sejak Django 5.x. Lihat panduan background tasks tanpa Celery untuk detailnya.

from django.tasks import task
from agent_runs.models import AgentRun


@task
def run_agent_task(run_id: str):
    run = AgentRun.objects.get(id=run_id)
    result = run_agent(run.prompt, user=run.user)
    run.output = result
    run.status = "completed"
    run.save()

Trigger dari view:

def trigger_agent(request):
    run = AgentRun.objects.create(
        user=request.user,
        prompt=request.POST["prompt"],
        status="pending",
    )
    run_agent_task.delay(str(run.id))
    return JsonResponse({"run_id": str(run.id), "status": "pending"})

User bisa polling status, atau kamu pake WebSocket buat notify pas selesai. Kombinasi ini cocok buat agent yang butuh waktu lama - document generation, data analysis, atau multi-step research.

Security Checklist

Integrasi AI agent ke Django membuka surface area baru. Ini checklist minimal:

User isolation. Setiap tool yang akses database harus filter berdasarkan request.user. Jangan sampai agent user A bisa query data user B. Inject user_id lewat AgentDeps dan pastikan setiap tool menggunakannya.

Write tools perlu confirmation. Tool yang nulis atau ngedelete data - kayak create_order atau delete_ticket - harus minta konfirmasi eksplisit sebelum eksekusi. Bisa pake two-step pattern: tool pertama return ringkasan, user confirm, tool kedua eksekusi.

Step budget. Agent jangan bisa looping forever. Set batas maksimum tool calls per run:

agent = Agent(model, max_tool_calls=15)

Kalau agent butuh lebih dari 15 langkah, force complete dan return partial result.

Timeout. Set timeout buat setiap agent run. Di task background, timeout 5 menit biasanya cukup:

@task(timeout=300)
def run_agent_task(run_id: str):
    ...

Audit trail. Simpan log setiap tool call: siapa yang trigger, tool apa, parameter apa, output apa, berapa lama. Ini penting buat debugging dan compliance.

class AgentAuditLog(models.Model):
    user = models.ForeignKey(User, on_delete=models.SET_NULL, null=True)
    agent_name = models.CharField(max_length=100)
    tool_name = models.CharField(max_length=100)
    input_data = models.JSONField()
    output_data = models.JSONField(blank=True, null=True)
    duration_ms = models.IntegerField()
    created_at = models.DateTimeField(auto_now_add=True)

Perbandingan Pola

Pola Use Case Kelebihan Kekurangan
Streaming endpoint Chat, live assistant Real-time, UX responsif Butuh koneksi persistent
Persistent memory Multi-turn conversation State aman, konteks lintas session Overhead storage, context window
Service-layer tools Agent yang action Integrasi langsung dgn existing code Risk: data leakage tanpa filter
Model-aware agents Bulk processing per row Terstruktur, type-safe N calls = N LLM requests
Background loop Long-running agent Non-blocking, retry-able Butuh polling / WebSocket

Kesimpulan

Integrasi AI agents ke Django bukan magic - ini soal milih pola yang tepat buat problem yang kamu hadapi.

Mulai dari streaming endpoint kalau kamu cuma butuh chat interface. Tambah persistent memory begitu butuh konteks lintas session. Ekspos service layer sebagai tools saat agent perlu action. Pake model-aware agents untuk bulk processing. Dan bungkus semuanya di background task kalau agent-nya berat.

Satu aturan yang nempel di semua pola: setiap tool harus explicit tentang siapa yang boleh akses apa. Django ngajarin kamu prinsip ini sejak awal - request.user, permissions, row-level filtering. AI agents bukan alasan buat ninggalin prinsip itu.

Pilih satu pola yang paling relevan dengan project kamu sekarang. Coba implementasiin di view sederhana. Dari situ kamu bakal lihat sendiri mana yang perlu di-scale dan mana yang cukup seadanya.

↑