Kotlin

Kurulum

Kotlin 2.4 ile derlenir, JVM 17 veya üstü gerekir. Tek bağımlılıklar kotlinx coroutines ve serialization. HTTP işini JDK'nın kendi istemcisi yapar, yani HTTP konusunda zaten bir tercihi olan bir servise hiçbir şey eklenmez.

dependencies {
    implementation("dev.mcpd:mcpd-kotlin:0.1.0")
}

İstemciyi oluşturma

Süreç başına bir örnek. McpdConfig alanlarının her biri varsayılanını bir ortam değişkeninden okur ve kodda da verilebilir, böylece aynı kod tek bir düzenleme olmadan hem geliştirmede hem üretimde çalışır. Argümansız Mcpd(), yerel bir mcpd ile çalışır.

// Every field also reads an environment variable (MCPD_ENDPOINT,
// MCPD_PROJECT, MCPD_API_KEY, ...), so `Mcpd()` on its own works too.
val mcpd = Mcpd(
    McpdConfig(
        endpoint = "http://localhost:4317",
        project = "support-desk",
        environment = "development",
        serviceName = "support-agent",
        serviceVersion = "1.4.0",
    )
)

Kod ortam değişkenine, ortam değişkeni de varsayılana üstün gelir.

enabled = false, her kayıt çağrısını yine de geçerli span'ler döndüren boş bir işleme çevirir; böylece testlerde sınanan koda dokunmadan mcpd kapatılabilir. Okumalar yine sunucuya gider: bir sorgu, çağıranın yanıtını beklediği bir şeydir ve boş bir yanıt yalan olurdu.

Bir çalıştırmayı kaydetme

span alışılmış biçimdir. Span'i kapatır, bloğun çağırdığı her şey için onu etkin span yapar ve blok ne döndürdüyse onu döndürür. Hata fırlatan bir blok, istisnayı taşıyan bir error span'i bırakır ve istisnayı olduğu gibi yeniden fırlatır.

// Bloğun değeri span'in değeridir, yani ölçümleme sizi kodu onun çevresinde
// yeniden kurmaya zorlamaz.
val answer = mcpd.span("retrieve_context", SpanKind.RETRIEVAL) { span ->
    span.setInput(query)
    val docs = index.search(query)
    span.setOutput(docs.map { it.id })
    span.setAttribute("retrieval.hits", docs.size)
    docs
}

İptal edilen bir blok farklıdır ve farklı kaydedilir. Çağıranı ya da bir zaman aşımı tarafından terk edilen bir coroutine, span'i error yerine cancelled olarak kapatır; çünkü bilerek yapılan bir terk, işlemin başarısızlığı değildir ve trace listesinde öyle okunmamalıdır.

val question = "where is my order 41?"

// `span` starts the operation, ends it, and records anything it throws.
// Inside the block the span is the active one, so the helpers below find
// their parent through the coroutine context rather than being handed it.
val traceId = mcpd.span(
    "support_agent",
    SpanKind.AGENT_RUN,
    input = question,
    sessionId = "session-8f21",
    userId = "customer-4477",
    tags = listOf("support", "email"),
) { run ->
    mcpd.log(LogLevel.INFO, "handling a question", mapOf("channel" to "email"))

    val plan = planAnswer(mcpd, question)
    val order = lookUpOrder(mcpd, 41)
    val estimate = estimateDelivery(mcpd, 41)

    val answer = buildString {
        append(plan).append(' ').append(order)
        if (estimate != null) append(" It should arrive ").append(estimate).append('.')
    }
    run.setOutput(answer)
    mcpd.metric("support.answers", 1.0, MetricKind.COUNTER, unit = "{answer}")

    // The block's value is the value of `span`.
    run.traceId
}

Blok yapısı size uymuyorsa span'i kendiniz başlatıp kendiniz kapatın.

val run = mcpd.trace("nightly_reindex", SpanKind.WORKFLOW_RUN, tags = listOf("batch"))
try {
    reindex()
    run.end()
} catch (e: Exception) {
    run.setError(e)
    run.end()
}

Bağlam

Etkin span, coroutine bağlamının bir ögesidir; yani thread'i değil coroutine'i izler. Birkaç katman aşağıdaki bir suspend fonksiyon, hiçbir çağıran ona bir üst vermeden doğru üst span'in altında bir alt span açar ve aynı dispatcher üzerindeki iki çalıştırma birbirinin span'lerini hiç görmez. Etkin span'i mcpd.currentSpan() ile okuyun, başlığını ise mcpd.traceparent() ile alın.

mcpd.span("support_agent", SpanKind.AGENT_RUN) { run ->
    // draft'a hiçbir şey geçilmez. Kaç suspend çağrı aşağıda olursa olsun,
    // çalıştırmayı coroutine bağlamından bulur.
    run.setOutput(draft(question))
}

suspend fun draft(question: String): String =
    mcpd.span("draft_answer", SpanKind.MODEL_GENERATION) { generation ->
        generation.setInput(question)
        "..."
    }

Servisler arasında bağlam, örnekleme kararını da birlikte taşıyan W3C traceparent başlığında gider; böylece bir trace hiçbir zaman yarım kaydedilmez. Bozuk bir başlık, onu taşıyan isteği başarısız kılmak yerine yeni bir trace başlatır.

// Giden: başlığı bir sonraki servise verin.
request.header("traceparent", span.traceparent())

// Gelen: yeni bir trace başlatmak yerine o trace'i sürdürün.
mcpd.span("checkout", traceparent = headers["traceparent"]) { span ->
    charge(span)
}

Model, kullanım ve hatalar

mcpd alım sırasında token sayılarını maliyete çevirir, dolayısıyla trace'te gördüğünüz rakam size faturalanan rakamdır. Bir hatayı kaydetmek aynı hamlede span'in durumunu error yapar ve her zaman bir mesaj bulunur: "bir şey başarısız oldu" ile kimse bir şey yapamaz.

generation.setModel("gpt-4o", provider = "openai")
generation.setUsage(inputTokens = 1_180, outputTokens = 96)

// setError elinizdeki throwable'ı alır; mesaj, sınıf ve yığın izinin hepsi
// ondan gelir.
tool.setError(e, retryable = true)

// Ya da doğrudan bir mesaj; iki durumda da mesaj zorunludur.
tool.setError("the shipping service returned 503")

Geri kalanlar setInput, setOutput, setMetadata, setAttribute, setAttributes, setTags, setCorrelation ve addLink; ayrıca açık bir alt span için child, bir zaman noktası için event. Hepsi her thread'den güvenle çağrılabilir ve hiçbiri hata fırlatmaz. Başka bir izleme sisteminin ağacına ait bir span, onu externalParentSpanId ile adlandırır; bu değer ilişkilendirme için taşınır ve mcpd'nin kendi ağacında hiçbir zaman üst olarak kullanılmaz.

Payload'lar, sink'ler ve bitmiş span'ler

capturePayloads = false, input ve output alanlarını sürecin içinde tutar: kayıt kuyruğa varmadan atılırlar, böylece bir flush sırasında hassas hiçbir şey dışarı çıkmaz; tek bir span bunu kendisi için geçersiz kılabilir. Exporter her yığını bir Sink nesnesine verir: varsayılan olarak HTTP taşıyıcısına, bir test için yığınları bellekte tutan bir RecordingSink nesnesine ya da sinyal başına tek bir JSON satırı yazan bir ConsoleSink nesnesine. Hata fırlatan bir sink, aralık başına bir kez bildirilen ve çağırana asla iletilmeyen başarısız bir yığın demektir.

// Prompt'ları ve tamamlamaları sürecin içinde tutun: span yine zamanlamasını,
// durumunu, kullanımını, modelini, ilişkilendirmesini ve hatalarını taşır.
val quiet = Mcpd(McpdConfig(capturePayloads = false))
quiet.span("draft_answer", input = question, capturePayloads = true) { }  // bu tutulur

// Başka bir izleyicinin bitirdiği bir span, verildiği gibi kuyruğa girer: üstü,
// zamanları ve örneklemesi değiştirilmez. Bozuk bir kimlik atılır ve sayılır,
// hiç fırlatılmaz.
mcpd.record(finishedSpan)

// Kaydedileni sunucu olmadan doğrulayın.
val recorder = RecordingSink()
val local = Mcpd(McpdConfig(sink = recorder))
local.span("support_agent") { run -> run.setUsage(inputTokens = 12, outputTokens = 3) }
local.flush()
check(recorder.spans.single().usage?.inputTokens == 12L)

// queued, dropped, capacity, exported, failed, rejected.
val stats = local.stats()

Kalıcı çalıştırmalar

Daemon'ın kalıcı çalıştırmalarını da aynı istemci karşılar. Workflow tanımı kurar ve adım gövdelerini suspend lambda olarak saklar; bir Worker bu tanımı kaydeder, lease'ler için uzun yoklama yapar ve her birini bir coroutine üzerinde karşılar; startRun, watchRun, publishEvent ve diğerleri ise istemci üzerinde çalıştırma yüzeyini bire bir yansıtan uzantılardır. Bir gövde, adımın bütçesi içinde yeniden denenmek için RetryableException fırlatır ve StepContext üzerinden kendi adımının altına span kaydeder. Bağlam üzerinden açılan her span, bittikçe kendi kullanımını adımınkine ekler; böylece model çağrılarını kaydeden bir gövde token'ları hiç elle toplamaz. setUsage ise span'lerin topladığını geçersiz kılar. Bir Worker, onError yanında dört isteğe bağlı hook alır (onLease, onStepStart, onStepEnd, onConflict) ve stats() çağrısını yanıtlar: talepler, yolda olan gövdeler, tamamlamalar, sınıfına göre başarısızlıklar, nedenine göre çakışmalar ve karşılanamayan lease'ler.

val support = Workflow("support_ticket")
    .step("classify") { ctx ->
        val ticket = ctx.input<JsonObject>()["ticket"]!!.jsonPrimitive.content
        mapOf("kind" to if (ticket.startsWith("Q")) "question" else "bug")
    }
    .agent("draft", StepOptions(input = mapOf("kind" to "\${steps.classify.output.kind}"))) {
        ctx ->
        // Your own model call, recorded under this step in the run's trace.
        // Its usage is summed into the step, so the daemon prices the run.
        ctx.span("model_call", SpanKind.MODEL_GENERATION) { span ->
            span.setModel("claude-sonnet-4", "anthropic")
            span.setUsage(inputTokens = 812, outputTokens = 37)
        }
        "an answer to a " + ctx.input<JsonObject>()["kind"]!!.jsonPrimitive.content
    }
    .waitFor(
        "approval", "approved", "\${input.ticket}", 7.days,
        StepOptions(guard = predicate("steps.classify.output.kind", "eq", "bug")),
    )
    .step(
        "send",
        StepOptions(
            input = mapOf("text" to "\${steps.draft.output}"),
            effect = Effect.AT_MOST_ONCE,
            retry = RetryPolicy(
                maxAttempts = 3, backoff = Backoff.EXPONENTIAL, initial = 500.milliseconds,
            ),
        ),
    ) { ctx ->
        val text = ctx.input<JsonObject>()["text"]!!.jsonPrimitive.content
        if (!deliver(text, ctx.idempotencyKey)) {
            throw RetryableException("mail relay busy", retryAfter = 2.seconds)
        }
        mapOf("sent" to true)
    }
    .output(mapOf("sent" to "\${steps.send.output.sent}"))

Her yapının ne yaptığı, lease protokolü ve çalıştırma API'si şu sayfalarda: İş akışları, Worker'lar ve Çalıştırmalar.

Puanlar ve geri bildirim

min ve max zorunlu parametrelerdir. Yedi, on üzerinden çok iyidir ama yüz üzerinden kötüdür; yani ölçeği bildirilmemiş bir sayı hiçbir şeyle karşılaştırılamaz. Geri bildirim bir insanın yargısıdır ve puandan ayrı durur; çünkü ikisi farklı sorulara yanıt verir ve asla birlikte ortalanmamalıdır.

// A machine evaluation. `min` and `max` are required: 7 is excellent out of
// 10 and poor out of 100, and a score with no scale cannot be compared.
mcpd.score(
    traceId = traceId,
    scorerId = "answer_relevance",
    value = 0.86,
    min = 0.0,
    max = 1.0,
    reason = "answered the question asked, without the delivery estimate",
)

// A person's judgement, from the thumbs-up control in your own product.
mcpd.feedback(
    traceId = traceId,
    kind = FeedbackKind.THUMBS,
    value = "up",
    comment = "fast and clear",
    authorId = "customer-4477",
)

// Nothing above touched the network. This is where it goes.
mcpd.flush()

Loglar ve metrikler etkin span'e bağlanır, böylece bir grafikteki sıçrama onu doğuran çalıştırmayı açar.

Flush ve kapatma

Kayıt, bellekteki sınırlı bir kuyruğa ekler ve döner; yığınlamayı, sıkıştırmayı ve göndermeyi bir worker coroutine'i yapar. flush() kuyrukta bekleyeni gönderir ve bekler. shutdown() worker'ı durdurur, kalanı gönderir ve iki kez çağrılması güvenlidir. Son dışa aktarım iptal edilemez biçimde çalışır; çünkü kapatmaya genellikle iptal sırasında bir finally bloğundan gelinir ve sürecin neden gittiğini açıklayan yığın da o son yığındır.

Mcpd bir AutoCloseable olduğu için hem use hem de bir kapatma hook'u işe yarar. İkisini de çağırmayan serverless bir fonksiyon ya da kısa ömürlü bir worker son yığınını kaybeder, çünkü runtime handler döner dönmez zamanlamayı durdurur.

// Bir JVM servisi: çıkarken kapatın.
Runtime.getRuntime().addShutdownHook(Thread { mcpd.close() })

// Kısa ömürlü bir worker: use ile kullanın, kapanış süslü parantezinde kapanır.
Mcpd(McpdConfig(project = "batch")).use { mcpd ->
    mcpd.span("reindex") { run -> reindex(run) }
}

Geri aldıklarınız

Okuma metotları HTTP API'yi bire bir yansıtır ve üretilmiş tipleri döndürür. Hata fırlatan tek metotlar bunlardır ve McpdException fırlatırlar: yanıtını beklediğiniz bir sorgu, yutulacak bir şey değildir.

$ gradle run> Task :runsupport_agent: 4 spans, ok  agent_run support_agent 79ms  model_generation plan_answer 41ms  tool_call lookup_order 16ms  tool_call estimate_delivery 11ms shipping service returned 503 for order 41

val detail = mcpd.getTrace(traceId, include = listOf("scores", "feedback"))
println("${detail.trace.name}: ${detail.spans.size} spans, ${detail.trace.status}")
for (span in detail.spans) {
    val ms = (span.durationNs ?: 0) / 1_000_000
    println("  ${span.kind} ${span.name} ${ms}ms ${span.error?.message ?: ""}")
}

val summary = mcpd.analyticsSummary(
    from = Clock.System.now() - 1.hours,
    to = Clock.System.now(),
)
println(
    "last hour: ${summary.traceCount} runs, " +
        "${summary.errorCount} failed, ${summary.totalCost} USD",
)

Güvenebilecekleriniz

• Ölçümleme hiçbir zaman hata fırlatmaz. Başarısız bir dışa aktarım, erişilemeyen bir sunucu ya da serileştirilemeyen bir değer, java.util.logging üzerinden 30 saniyede en çok bir kez bildirilir, geri kalanında yutulur. Gözlemlenebilirlik, gözlediği şeyi asla devirmemelidir.
• Kayıt bloklamaz, yani servisinizin gecikmesi mcpd'nin hızlı ya da ayakta olmasına hiç bağlı değildir.
• Bellek sınırlıdır. Kuyruk maxQueue kadar sinyal tutar; bunun ötesinde en eskiler atılır ve sayılır; bu sayı droppedSignals alanından ve stats() içinden okunur. Sınırsız bir tampon, bir ağ kesintisini bellek yetersizliğinden ölüme çevirir.
• Örnekleme kararı bir kez, kökte verilir ve aşağıya kalıtılır.
• Bir span hem başladığında hem bittiğinde dışa aktarılır, yani yolun ortasında çökmüş bir çalıştırma yine de görünür olur. Bir span, başlangıç kaydı gönderilmeden önce biterse kayıt yerinde güncellenir ve yalnızca bir tanesi gönderilir.
• Yığınlar 1 KiB üzerinde gzip ile sıkıştırılır ve yeniden denemeyi atlatan bir idempotency anahtarı taşır; böylece sunucu kaydettikten sonra zaman aşımına uğrayan bir istek iki kez sayılamaz.
• Süreler monotonik bir saatten ölçülür, yani bir NTP sıçraması negatif bir süre üretemez.

Çalıştırma bulma filtreleri eksiksiz anlatır. Veri modeli bu tiplerin taşıdığı her alanı listeler.