Blog AI Agents
SISTEM AKSES KOMPUTER

Debugging dan Observabilitas dalam Sistem Agen Otonom

agentsdebuggingobservabilitytutorialarchitecture

Agen otonom yang gagal secara diam-diam lebih buruk daripada tidak ada agen sama sekali. Ketika fungsi tradisional melempar pengecualian, Anda mendapatkan stack trace. Ketika agen mengambil jalan yang salah melalui dua puluh pemanggilan tool dan tiga invokasi model, Anda mendapatkan jawaban yang salah — tanpa penjelasan yang jelas.

Men-debug agen membutuhkan model mental yang berbeda. Sistem tidak menjalankan jalur yang deterministik; sistem membuat serangkaian keputusan. Observabilitas berarti menangkap keputusan-keputusan tersebut — bukan hanya input dan output, tetapi penalaran yang menghubungkannya.

Mengapa Debugging Tradisional Gagal untuk Agen

Logging standar menangkap apa yang terjadi. Observabilitas agen membutuhkan penangkapan mengapa — apa yang disimpulkan model, tool mana yang dipilih dan mengapa, dan dari state menengah mana ia bekerja.

Mode kegagalannya juga berbeda:

  • Halusinasi diam: Agen dengan percaya diri menghasilkan jawaban yang salah tanpa memberi sinyal ketidakpastian.
  • Pergeseran keputusan: Setiap langkah terlihat masuk akal secara lokal, tetapi urutannya menyimpang dari tujuan.
  • Penggunaan tool yang salah: Agen memanggil tool yang tepat dengan parameter yang sedikit salah.
  • Loop tak terbatas: Agen terjebak mencoba ulang pendekatan yang gagal.
  • Pencemaran konteks: Output buruk dari langkah awal merusak semua penalaran berikutnya.

Tidak ada satupun dari ini yang menghasilkan pengecualian. Mereka menghasilkan perilaku yang salah yang hanya terlihat ketika Anda merekonstruksi trace eksekusi lengkap.

Logging Terstruktur untuk Keputusan Agen

Langkah pertama adalah membungkus setiap interaksi agen dalam log terstruktur. Jangan log respons API mentah — log event semantik.

import json
import time
import uuid
from dataclasses import dataclass, asdict
from typing import Any
import anthropic


client = anthropic.Anthropic()


@dataclass
class AgentEvent:
    trace_id: str
    step: int
    event_type: str      # "llm_call", "tool_call", "tool_result", "decision", "error"
    model: str | None
    input_tokens: int | None
    output_tokens: int | None
    latency_ms: float | None
    content: dict[str, Any]
    timestamp: float


def log_event(event: AgentEvent):
    print(json.dumps(asdict(event)))  # Ganti dengan tujuan log Anda


class TracedAgent:
    def __init__(self, trace_id: str | None = None):
        self.trace_id = trace_id or str(uuid.uuid4())
        self.step = 0
        self.tools = []


    def add_tool(self, name: str, description: str, input_schema: dict):
        self.tools.append({
            "name": name,
            "description": description,
            "input_schema": input_schema
        })


    def call(self, messages: list[dict], system: str = "") -> str:
        self.step += 1
        start = time.monotonic()


        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=4096,
            system=system,
            tools=self.tools,
            messages=messages
        )


        latency_ms = (time.monotonic() - start) * 1000


        log_event(AgentEvent(
            trace_id=self.trace_id,
            step=self.step,
            event_type="llm_call",
            model="claude-opus-4-6",
            input_tokens=response.usage.input_tokens,
            output_tokens=response.usage.output_tokens,
            latency_ms=latency_ms,
            content={
                "stop_reason": response.stop_reason,
                "text_blocks": [b.text for b in response.content if b.type == "text"],
                "tool_calls": [
                    {"name": b.name, "input": b.input}
                    for b in response.content if b.type == "tool_use"
                ]
            },
            timestamp=time.time()
        ))


        return response

Membangun Trace Lengkap

Satu baris log tidak cukup — Anda membutuhkan trace eksekusi lengkap yang menghubungkan setiap keputusan dengan hasilnya:

from typing import Callable


def run_traced_agent(
    task: str,
    tools: dict[str, Callable],
    tool_schemas: list[dict],
    system: str,
    max_steps: int = 20,
) -> dict:
    agent = TracedAgent()
    for schema in tool_schemas:
        agent.add_tool(**schema)


    messages = [{"role": "user", "content": task}]
    trace = {"trace_id": agent.trace_id, "task": task, "steps": []}
    step_count = 0


    while step_count < max_steps:
        step_count += 1
        response = agent.call(messages, system=system)


        step_record = {
            "step": step_count,
            "stop_reason": response.stop_reason,
            "model_output": [],
            "tool_results": []
        }


        if response.stop_reason == "end_turn":
            for block in response.content:
                if block.type == "text":
                    step_record["model_output"].append(block.text)
            trace["steps"].append(step_record)
            trace["final_answer"] = step_record["model_output"][-1] if step_record["model_output"] else ""
            break


        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                step_record["model_output"].append({
                    "tool": block.name,
                    "input": block.input
                })


                tool_fn = tools.get(block.name)
                if not tool_fn:
                    result = f"Error: tool tidak dikenal '{block.name}'"
                else:
                    try:
                        result = tool_fn(**block.input)
                    except Exception as e:
                        result = f"Error tool: {e}"


                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(result)
                })
                step_record["tool_results"].append({
                    "tool": block.name,
                    "result_preview": str(result)[:200]
                })


        trace["steps"].append(step_record)
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})


    else:
        trace["error"] = f"melebihi max_steps ({max_steps})"


    return trace

Deteksi Loop

Loop tak terbatas adalah mode kegagalan yang umum. Deteksi dengan mengambil sidik jari pola invokasi tool dari setiap panggilan LLM:

def detect_loop(trace: dict, window: int = 4) -> bool:
    steps = trace["steps"]
    if len(steps) < window:
        return False


    def step_signature(step: dict) -> str:
        tools_called = sorted(
            t["tool"] if isinstance(t, dict) else t
            for t in step.get("model_output", [])
            if isinstance(t, dict) and "tool" in t
        )
        return "|".join(tools_called)


    recent = [step_signature(s) for s in steps[-window:]]
    if len(set(recent)) == 1 and recent[0]:
        return True


    if len(steps) >= 4:
        pattern = [step_signature(s) for s in steps[-4:]]
        if pattern[0] == pattern[2] and pattern[1] == pattern[3]:
            return True


    return False

Metrik untuk Produksi

from collections import Counter


def compute_trace_metrics(trace: dict) -> dict:
    steps = trace["steps"]
    errors = [s for s in steps if "error" in s]


    tool_calls_by_name: Counter = Counter()
    for step in steps:
        for output in step.get("model_output", []):
            if isinstance(output, dict) and "tool" in output:
                tool_calls_by_name[output["tool"]] += 1


    return {
        "trace_id": trace["trace_id"],
        "total_steps": len(steps),
        "error_steps": len(errors),
        "tool_call_distribution": dict(tool_calls_by_name),
        "completed": "final_answer" in trace,
        "loop_detected": detect_loop(trace),
    }

Sinyal utama untuk alert:

  • Tingkat loop > 5% run — agen terjebak
  • Tingkat error per tool > ambang batas — sebuah tool rusak
  • Rata-rata langkah tren naik — tugas semakin sulit
  • Latensi p99 spike — endpoint model lambat

Integrasi OpenTelemetry

Untuk tim yang sudah menggunakan OpenTelemetry, emit trace agen sebagai span:

from opentelemetry import trace as otel_trace


tracer = otel_trace.get_tracer("agent")


def run_with_otel(task: str, tools: dict, tool_schemas: list, system: str):
    with tracer.start_as_current_span("agent.run") as root_span:
        root_span.set_attribute("agent.task", task[:200])


        agent = TracedAgent()
        for schema in tool_schemas:
            agent.add_tool(**schema)


        messages = [{"role": "user", "content": task}]


        for step in range(20):
            with tracer.start_as_current_span(f"agent.step.{step}") as step_span:
                response = agent.call(messages, system=system)
                step_span.set_attribute("llm.stop_reason", response.stop_reason)
                step_span.set_attribute("llm.input_tokens", response.usage.input_tokens)


                if response.stop_reason == "end_turn":
                    break

Redaksi PII dalam Log

Log agen sering berisi data sensitif. Sebelum mengirim ke sistem eksternal, lakukan redaksi:

import re


PII_PATTERNS = [
    (re.compile(r'\b[\w.+-]+@[\w-]+\.[a-z]{2,}\b'), '[EMAIL]'),
    (re.compile(r'\b0\d{2,3}[-\s]?\d{6,8}\b'), '[TELEPON]'),
    (re.compile(r'\bsk-[a-zA-Z0-9]{20,}\b'), '[API_KEY]'),
]


def redact(text: str) -> str:
    for pattern, replacement in PII_PATTERNS:
        text = pattern.sub(replacement, text)
    return text

Tiga Metrik Terpenting

Tingkat penyelesaian tugas — berapa fraksi run yang mencapai final_answer vs. mencapai max_steps atau error. Buat baseline per jenis tugas.

Biaya token per tugas — jumlah input_tokens + output_tokens di semua langkah. Kenaikan biaya 20% tanpa perubahan tingkat penyelesaian biasanya menandakan degradasi prompt.

Tingkat error tool — error_steps / total_steps. Lonjakan metrik ini menunjuk langsung ke tool atau API yang rusak.

Observabilitas dalam sistem agen bukan opsional — itu adalah perbedaan antara sistem yang bisa Anda iterasi dan yang hanya bisa Anda restart saat rusak. Mulai dengan event terstruktur dan trace ID. Tambahkan deteksi loop. Push metrik. Investasi ini terbayar saat pertama kali Anda mengalami kegagalan produksi dan bisa merekonstruksi tepat apa yang terjadi alih-alih menebak-nebak.

Artikel Terkait