Elixir

Installazione

Aggiungilo a mix.exs. Non si porta dietro niente.

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

L’elenco delle dipendenze è tutto qui, ed è voluto. OTP 27 porta un codec JSON (:json) e un client HTTP (:inets), quindi una libreria il cui mestiere è guardare il tuo sistema non deve diventare il motivo per cui la build del tuo sistema cambia. Gli altri cinque SDK fanno la stessa promessa, ognuno nel proprio linguaggio.

Costruire il client

Ogni opzione ricade sulla propria variabile d’ambiente, e il codice vince. Mcpd.start_link() senza argomenti parla con un daemon locale.

# 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"
  )

Mettilo nel tuo albero di supervisione e si comporta come qualunque altro worker:

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

Supervisor.start_link(children, strategy: :one_for_one)

Registrare un’esecuzione

span è una macro, quindi un’esecuzione è un blocco e il suo valore è il valore del blocco.

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

Le macro vanno richieste prima di poter essere chiamate, ed è una regola del linguaggio, non una scelta presa qui. use Mcpd lo fa per te:

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

Se il corpo è già una funzione anziché un blocco, chiama Mcpd.with_span/4. È la stessa cosa senza la macro.

L’albero si costruisce da solo

Lo span attivo vive nel process dictionary, e in Elixir nello specifico quello è il posto giusto dove tenerlo, non una scorciatoia. Qui l’unità di concorrenza è il processo, quindi due richieste servite nello stesso momento non possono vedere lo span dell’altra, e non c’è niente da far passare per ogni funzione per tenere insieme l’albero.

Un figlio avviato tre chiamate più in basso trova il proprio padre da solo:

Mcpd.span mcpd, "agent", kind: :agent_run do
  plan(mcpd)          # apre un figlio di "agent" senza che glielo si dica
end

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

La conseguenza che vale la pena conoscere: uno span aperto dentro Task.async fa partire una traccia nuova, perché quello è un altro processo. Passa la traccia di là con Mcpd.traceparent/0 e Mcpd.continue/2, la stessa coppia che useresti attraverso un confine HTTP.

Niente blocca, e niente solleva eccezioni

Registrare uno span manda un messaggio al worker e ritorna. La rete avviene su quel worker, quindi la latenza del tuo agente non dipende mai da quanto mcpd sia veloce, né dal fatto che sia raggiungibile, né dal fatto che stia girando.

Un’eccezione sollevata dentro un blocco viene registrata sullo span come errore e poi risollevata identica. Il mestiere di mcpd è vedere che cosa è successo, mai decidere che cosa ne fa il tuo codice.

La coda ha un tetto, e quando si riempie i segnali più vecchi vengono scartati. Quella direzione è scelta: quando il daemon è irraggiungibile, i segnali più recenti raccontano che cosa sta succedendo adesso, e sono quelli che qualcuno vorrà quando il daemon torna. Mcpd.stats/1 risponde che cosa ha fatto il client: queued, dropped, capacity, exported, failed e rejected.

log_level:, oppure MCPD_LOG_LEVEL, impone una soglia minima a Mcpd.log/4. L’impostazione predefinita è debug, che tiene tutto. Una riga sotto la soglia viene scartata prima che il record venga costruito, quindi una riga di debug in un deploy di produzione costa un confronto. La soglia vale solo per i log: span, metriche, punteggi e feedback non ne risentono.

Tenere i payload dentro il processo

capture_payloads: false, oppure MCPD_CAPTURE_PAYLOADS=false, tiene i prompt e le completion dove sono. Gli span escono lo stesso con i loro tempi, lo stato, il consumo, il modello, la correlazione e gli errori; input:, output: e il valore catturato di un blocco vengono scartati prima di arrivare alla coda, così niente di sensibile va in coda e niente di sensibile esce con un flush. Un singolo span può dire il contrario per sé con capture_payloads: true.

Verificare quello che hai registrato

Un sink è il posto dove finisce un batch svuotato. Il trasporto HTTP è quello predefinito; Mcpd.Sink.Recorder tiene in memoria ogni batch, così un test rilegge gli span, i log, le metriche, i punteggi e il feedback che la strumentazione ha registrato senza un daemon, e Mcpd.Sink.Console stampa una riga JSON per segnale su stderr. Un sink tuo implementa Mcpd.Sink; uno che solleva un’eccezione è un batch fallito, contato e riferito, mai sollevato verso chi ha chiamato.

Mcpd.record/3 mette in coda uno span già concluso, così com’è: con i suoi identificatori, il suo padre, il suo inizio e la sua fine, senza che gli si cambi mai il padre, i tempi o il campionamento. È quello che ha sottomano un adattatore per un altro sistema di tracing o un’importazione in blocco. Un record i cui identificatori non hanno la forma di OpenTelemetry viene rifiutato e contato, e non solleva mai un’eccezione. external_parent_span_id: su qualunque span porta un padre che viene da un sistema di tracing che non è mcpd, solo per la correlazione.

Esecuzioni durevoli

Lo stesso client serve le esecuzioni durevoli del daemon. Mcpd.Workflow costruisce la definizione e conserva i corpi dei passi come funzioni a tre argomenti; un Mcpd.StepWorker la registra, fa long polling per ottenere i lease e serve ognuno di essi su un processo suo, e Mcpd.start_run/3, Mcpd.watch_run/3, Mcpd.publish_event/4 e gli altri rispecchiano uno a uno la superficie delle esecuzioni. Un corpo solleva Mcpd.StepException con class: :retryable per essere ritentato entro il budget del suo passo, e registra span sotto il proprio passo attraverso Mcpd.StepContext.span/4. Ogni span che il corpo chiude somma il proprio consumo a quello del passo, e vince l’ultimo modello visto, così il daemon valorizza quello che la traccia mostra; set_usage/2 è l’override esplicito. on_lease, on_step_start, on_step_end e on_conflict stanno accanto a on_error sul worker, e StepWorker.stats/1 conta quello che ha preso in carico, quello che ha completato, quello che è fallito per classe e quello che ha abbandonato per motivo.

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}"})

Che cosa fa ogni costrutto, il protocollo dei lease e l’API delle esecuzioni stanno su Workflow, Worker ed Esecuzioni.

Punteggi e feedback

# 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)

Rileggerla

{: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

I processi di breve durata devono fare il flush

Uno script o un task che esce in fretta perde tutto quello che non era ancora partito. Mcpd.shutdown/1 fa il flush e ferma il worker, e si può chiamare due volte senza rischi.

Poi

• Ogni comportamento che questo SDK ha in comune con gli altri cinque: il contratto degli SDK.
• Che cosa porta uno span: Modello dei dati.
• Trovare più tardi l’esecuzione che ti interessa: Trovare un’esecuzione.