Elixir

Kurulum

mix.exs dosyasına ekleyin. Yanında hiçbir şey getirmez.

defp deps do
  [{:mcpd, "~> 0.1"}]
end

Bağımlılık listesinin tamamı budur ve bu bilerek böyledir. OTP 27 bir JSON kodlayıcı (:json) ve bir HTTP istemcisi (:inets) getirir, yani işi sisteminizi izlemek olan bir kütüphanenin, sisteminizin derlemesinin değişmesine sebep olması gerekmez. Diğer beş SDK da kendi dillerinde aynı sözü verir.

İstemciyi oluşturma

Her seçenek, verilmediğinde kendi ortam değişkenine düşer ve kod üstün gelir. Argümansız Mcpd.start_link(), yerel bir daemon ile konuşur.

# Every option also reads an environment variable (MCPD_ENDPOINT, MCPD_PROJECT,
# MCPD_API_KEY, ...), so `Mcpd.start_link()` on its own works too.
{:ok, mcpd} =
  Mcpd.start_link(
    endpoint: "http://localhost:4317",
    project: "support-desk",
    environment: "development",
    service_name: "support-agent",
    service_version: "1.4.0"
  )

Denetim ağacınıza koyun; başka herhangi bir worker gibi davranır:

children = [
  {Mcpd, project: "support-desk", environment: "production"}
]

Supervisor.start_link(children, strategy: :one_for_one)

Bir çalıştırmayı kaydetme

span bir makrodur, yani bir çalıştırma bir bloktur ve değeri de bloğun değeridir.

question = "where is my order 41?"

# The block's value is the value of `span`. Inside it, this span is the active
# one for the process, so the helpers below find their parent without being
# handed it.
trace_id =
  Mcpd.span mcpd, "support_agent",
    kind: :agent_run,
    input: question,
    session_id: "session-8f21",
    user_id: "customer-4477",
    tags: ["support", "email"] do
    Mcpd.log(mcpd, :info, "handling a question", %{"channel" => "email"})

    plan = plan_answer.(mcpd, question)
    order = look_up_order.(mcpd, 41)
    estimate = estimate_delivery.(mcpd, 41)

    _answer =
      [plan, order, estimate && "It should arrive #{estimate}."]
      |> Enum.reject(&is_nil/1)
      |> Enum.join(" ")

    Mcpd.metric(mcpd, "support.answers", 1, kind: :counter, unit: "{answer}")
    Mcpd.Context.current().trace_id
  end

Makrolar çağrılabilmek için önce require edilmelidir; bu, buradaki bir tercih değil dilin bir kuralıdır. use Mcpd bunu sizin yerinize yapar:

defmodule MyAgent do
  use Mcpd

  def answer(mcpd, question) do
    Mcpd.span mcpd, "agent", kind: :agent_run, input: question do
      do_the_work(question)
    end
  end
end

Gövde bir blok değil de zaten bir fonksiyonsa onun yerine Mcpd.with_span/4 çağırın. Makrosuz hâliyle aynı şeydir.

Ağaç kendi kendine kurulur

Etkin span, süreç sözlüğünde durur; bu bir kestirme yol değil, özellikle Elixir'de doğru olan saklama yeridir. Burada eşzamanlılığın birimi süreçtir, yani aynı anda işlenen iki istek birbirinin span'ini göremez ve ağacı bir arada tutmak için her fonksiyona bir şey geçirmek gerekmez.

Üç çağrı derinde başlatılan bir alt span, üstünü kendi başına bulur:

Mcpd.span mcpd, "agent", kind: :agent_run do
  plan(mcpd)          # söylenmeden "agent" altında bir alt span açar
end

defp plan(mcpd) do
  Mcpd.span mcpd, "plan_answer", kind: :model_generation, model: "gpt-4o" do
    call_the_model()
  end
end

Bilinmesi gereken sonuç şu: Task.async içinde açılan bir span yeni bir trace başlatır, çünkü orası farklı bir süreçtir. Trace'i Mcpd.traceparent/0 ve Mcpd.continue/2 ile karşıya geçirin; bir HTTP sınırında da aynı ikiliyi kullanırsınız.

Hiçbir şey bloklamaz, hiçbir şey hata fırlatmaz

Bir span'i kaydetmek, worker'a bir mesaj gönderir ve döner. Ağ işi o worker'da olur, yani ajanınızın gecikmesi mcpd'nin hızlı, ayakta ya da hiç çalışıyor olmasına bağlı değildir.

Bir blok içinde fırlatılan hata, span üzerine bir hata olarak kaydedilir ve sonra olduğu gibi yeniden fırlatılır. mcpd'nin işi ne olduğunu görmektir, kodunuzun buna ne yapacağına karar vermek değil.

Kuyruk sınırlıdır ve dolduğunda en eski sinyaller atılır. Bu yön bilerek seçilmiştir: daemon'a erişilemediğinde şu an ne olduğunu en yeni sinyaller anlatır ve daemon geri geldiğinde birinin isteyeceği sinyaller de onlardır. Mcpd.stats/1 istemcinin ne yaptığını söyler: queued, dropped, capacity, exported, failed ve rejected.

log_level: ya da MCPD_LOG_LEVEL, Mcpd.log/4 için bir alt sınır koyar. Varsayılan olan debug her şeyi tutar. Sınırın altındaki bir satır, kayıt oluşturulmadan önce atılır; yani üretime alınmış bir dağıtımdaki bir debug satırının bedeli tek bir karşılaştırmadır. Bu sınır yalnızca logları kapsar; span'ler, metrikler, puanlar ve geri bildirimler etkilenmez.

Payload'ları sürecin içinde tutma

capture_payloads: false ya da MCPD_CAPTURE_PAYLOADS=false, prompt'ları ve tamamlamaları bulundukları yerde tutar. Span'ler yine zamanlaması, durumu, token kullanımı, modeli, ilişkilendirmesi ve hatalarıyla çıkar; input:, output: ve bir bloğun yakalanan değeri kuyruğa varmadan atılır, böylece hassas hiçbir şey kuyruğa girmez ve bir flush sırasında hassas hiçbir şey dışarı çıkmaz. Tek bir span, capture_payloads: true ile kendisi için tersini söyleyebilir.

Kaydettiğinizi doğrulama

Sink, boşaltılan bir yığının gittiği yerdir. Varsayılan olan HTTP taşıyıcısıdır; Mcpd.Sink.Recorder her yığını bellekte tutar, böylece bir test ölçümlemenin kaydettiği span'leri, logları, metrikleri, puanları ve geri bildirimleri daemon olmadan geri okur; Mcpd.Sink.Console ise sinyal başına tek bir JSON satırını stderr'e yazar. Kendi sink'iniz Mcpd.Sink davranışını uygular; hata fırlatan bir sink, sayılan ve bildirilen başarısız bir yığındır, çağırana asla fırlatılmaz.

Mcpd.record/3, çoktan bitmiş bir span'i verildiği gibi kuyruğa alır: kendi kimlikleri, üstü, başlangıcı ve bitişiyle, üstü ve zamanları değiştirilmeden ve yeniden örneklenmeden. Başka bir izleme sistemi için yazılmış bir adaptörün ya da toplu bir içe aktarmanın elinde olan da budur. Kimlikleri OpenTelemetry biçiminde olmayan bir kayıt reddedilir ve sayılır, hiçbir zaman hata fırlatmaz. Her span'deki external_parent_span_id:, mcpd olmayan bir izleme sisteminden gelen bir üstü yalnızca ilişkilendirme için taşır.

Kalıcı çalıştırmalar

Daemon'ın kalıcı çalıştırmalarını da aynı istemci karşılar. Mcpd.Workflow tanımı kurar ve adım gövdelerini üç argümanlı fonksiyonlar olarak saklar; bir Mcpd.StepWorker bu tanımı kaydeder, lease'ler için uzun yoklama yapar ve her birini kendi süreci üzerinde karşılar; Mcpd.start_run/3, Mcpd.watch_run/3, Mcpd.publish_event/4 ve diğerleri çalıştırma yüzeyini bire bir yansıtır. Bir gövde, adımın bütçesi içinde yeniden denenmek için class: :retryable ile Mcpd.StepException fırlatır ve Mcpd.StepContext.span/4 üzerinden kendi adımının altına span kaydeder. Gövdenin kapattığı her span kendi kullanımını adımınkine ekler ve en son görülen model geçerli olur, böylece daemon trace'in gösterdiğini fiyatlandırır; set_usage/2 açık geçersiz kılmadır. Worker üzerinde on_error yanında on_lease, on_step_start, on_step_end ve on_conflict durur; StepWorker.stats/1 ise neyi talep ettiğini, tamamladığını, sınıfına göre neyi başarısız kıldığını ve nedenine göre neyi bıraktığını sayar.

support =
  Workflow.new("support_ticket")
  |> Workflow.step("classify", fn _ctx, input, _prior ->
    kind = if String.starts_with?(input["ticket"], "Q"), do: "question", else: "bug"
    %{"kind" => kind}
  end)
  |> Workflow.agent(
    "draft",
    fn ctx, input, _prior ->
      # Your own model call, recorded under this step in the run's trace. Its
      # usage is the step's: the daemon prices what the trace shows.
      opts = [kind: :model_generation, model: "claude-sonnet-4", provider: "anthropic",
        usage: %{"input_tokens" => 812, "output_tokens" => 37}]
      StepContext.span(ctx, "model_call", opts, fn -> :ok end)

      "an answer to a #{input["kind"]}"
    end,
    input: %{"kind" => "${steps.classify.output.kind}"}
  )
  |> Workflow.wait_for("approval",
    event: "approved",
    key: "${input.ticket}",
    timeout_ms: 7 * 24 * 3_600_000,
    when: where("steps.classify.output.kind", :eq, "bug")
  )
  |> Workflow.step(
    "send",
    fn ctx, input, _prior ->
      unless deliver.(input["text"], ctx.idempotency_key) do
        raise Mcpd.StepException,
          message: "mail relay busy", class: :retryable, retry_after_ms: 2_000
      end

      %{"sent" => true}
    end,
    input: %{"text" => "${steps.draft.output}"},
    effect: :at_most_once,
    retry: [max_attempts: 3, backoff: :exponential, initial_ms: 500]
  )
  |> Workflow.output(%{"sent" => "${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

# 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(mcpd, trace_id, "answer_relevance", 0.86,
  min: 0,
  max: 1,
  reason: "answered the question asked, without the delivery estimate"
)

# A person's judgement, from the thumbs-up control in your own product.
Mcpd.feedback(mcpd, trace_id, :thumbs, "up",
  comment: "fast and clear",
  author_id: "customer-4477"
)

# Nothing above touched the network. This is where it goes.
:ok = Mcpd.flush(mcpd)

Geri okuma

{:ok, detail} = Mcpd.get_trace(mcpd, trace_id, include: ["scores", "feedback"])

IO.puts("#{detail.trace.name}: #{length(detail.spans)} spans")

for span <- detail.spans do
  ms = div(span.duration_ns || 0, 1_000_000)
  error = (span.error && span.error.message) || ""
  IO.puts("  #{span.kind} #{span.name} #{ms}ms #{error}")
end

Kısa ömürlü süreçler flush etmelidir

Hızla çıkan bir betik ya da görev, gönderilmemiş olan her şeyi kaybeder. Mcpd.shutdown/1 flush eder, worker'ı durdurur ve iki kez çağrılması güvenlidir.

Sonrası

• Bu SDK'nın diğer beşiyle paylaştığı her davranış: SDK sözleşmesi.
• Bir span'in neler taşıdığı: Veri modeli.
• Sonrasında ilgilendiğiniz çalıştırmayı bulmak: Çalıştırma bulma.