Elixir

Installation

Ajoutez-le à mix.exs. Il n’amène rien avec lui.

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

C’est là toute la liste des dépendances, et c’est délibéré. OTP 27 fournit un codec JSON (:json) et un client HTTP (:inets) : une bibliothèque dont le travail est de surveiller votre système n’a donc pas à devenir une raison pour laquelle la compilation de ce système change. Les cinq autres SDK font la même promesse dans leur propre langage.

Construire le client

Chaque option retombe sur sa variable d’environnement, et le code l’emporte. Mcpd.start_link() sans argument s’adresse à un daemon local.

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

Placez-le dans votre arbre de supervision et il se comporte comme n’importe quel autre worker :

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

Supervisor.start_link(children, strategy: :one_for_one)

Enregistrer une exécution

span est une macro : une exécution est donc un bloc et sa valeur est la valeur du bloc.

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

Les macros doivent être requises avant de pouvoir être appelées, ce qui est une règle du langage et non un choix fait ici. use Mcpd s’en charge pour vous :

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

Si le corps est déjà une fonction plutôt qu’un bloc, appelez Mcpd.with_span/4 à la place. C’est la même chose sans la macro.

L’arbre s’assemble tout seul

Le span actif vit dans le dictionnaire de processus, et c’est le bon endroit en Elixir précisément, non un raccourci. Le processus est ici l’unité de concurrence : deux requêtes traitées en même temps ne peuvent donc pas voir le span l’une de l’autre, et rien n’a besoin d’être passé de fonction en fonction pour tenir l’arbre ensemble.

Un enfant démarré trois appels plus bas trouve son parent tout seul :

Mcpd.span mcpd, "agent", kind: :agent_run do
  plan(mcpd)          # ouvre un enfant de "agent" sans qu’on le lui dise
end

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

La conséquence à connaître : un span ouvert à l’intérieur de Task.async démarre une nouvelle trace, parce que c’est un autre processus. Faites passer la trace avec Mcpd.traceparent/0 et Mcpd.continue/2, la même paire que vous emploieriez pour franchir une frontière HTTP.

Rien ne bloque, et rien ne lève d’exception

Enregistrer un span envoie un message au worker et rend la main. Le réseau se passe sur ce worker : la latence de votre agent ne dépend donc jamais de la vitesse de mcpd, ni de sa disponibilité, ni même du fait qu’il tourne.

Une exception levée dans un bloc est enregistrée sur le span comme une erreur, puis relevée telle quelle. Le travail de mcpd est de voir ce qui s’est passé, jamais de décider ce que votre code en fait.

La file d’attente est bornée, et quand elle se remplit les signaux les plus anciens sont écartés. Ce sens est un choix : quand le daemon est injoignable, ce sont les signaux les plus récents qui décrivent ce qui se passe maintenant, et ce sont ceux-là que quelqu’un voudra à son retour. Mcpd.stats/1 répond ce que le client a fait : queued, dropped, capacity, exported, failed et rejected.

log_level:, ou MCPD_LOG_LEVEL, pose un plancher sous Mcpd.log/4. La valeur par défaut est debug, qui garde tout. Une ligne sous le plancher est écartée avant que l’enregistrement ne soit construit : une ligne de debug dans un déploiement de production coûte donc une comparaison. Le plancher ne concerne que les logs ; les spans, les métriques, les scores et le feedback ne sont pas touchés.

Garder les payloads dans le processus

capture_payloads: false, ou MCPD_CAPTURE_PAYLOADS=false, garde les prompts et les complétions là où ils sont. Les spans sortent toujours avec leurs temps, leur statut, leur consommation, leur modèle, leur corrélation et leurs erreurs ; input:, output: et la valeur capturée d’un bloc sont écartés avant d’atteindre la file d’attente, si bien que rien de sensible n’est mis en file et que rien de sensible ne sort lors d’un flush. Un span peut décider autrement pour lui-même avec capture_payloads: true.

Vérifier ce que vous avez enregistré

Un sink est l’endroit où part un lot vidé. Le transport HTTP est le sink par défaut ; Mcpd.Sink.Recorder garde chaque lot en mémoire, si bien qu’un test relit les spans, les logs, les métriques, les scores et le feedback que l’instrumentation a enregistrés sans daemon, et Mcpd.Sink.Console imprime une ligne JSON par signal sur stderr. Un sink à vous implémente Mcpd.Sink ; un sink qui lève une exception donne un lot en échec, compté et signalé, jamais relevé chez l’appelant.

Mcpd.record/3 met en file un span déjà terminé, tel quel : ses propres identifiants, son parent, son début et sa fin, jamais re-parenté, re-daté ni ré-échantillonné. C’est ce que détient un adaptateur pour un autre système de tracing, ou un import par lots. Un enregistrement dont les identifiants n’ont pas la forme OpenTelemetry est refusé et compté, et il ne lève jamais d’exception. external_parent_span_id:, sur n’importe quel span, porte un parent venu d’un système de tracing autre que mcpd, pour la seule corrélation.

Exécutions durables

Le même client dessert les exécutions durables du daemon. Mcpd.Workflow construit la définition et retient le corps des étapes sous forme de fonctions d’arité trois ; un Mcpd.StepWorker l’enregistre, attend des leases en long polling et sert chacun d’eux sur un processus à lui, et Mcpd.start_run/3, Mcpd.watch_run/3, Mcpd.publish_event/4 et les autres reprennent un pour un la surface des exécutions. Un corps d’étape lève Mcpd.StepException avec class: :retryable pour être réessayé dans le budget de l’étape, et enregistre des spans sous son étape via Mcpd.StepContext.span/4. Chaque span que le corps referme ajoute sa consommation à celle de l’étape, et le dernier modèle vu l’emporte : le daemon tarife donc ce que montre la trace ; set_usage/2 est la substitution explicite. on_lease, on_step_start, on_step_end et on_conflict accompagnent on_error sur le worker, et StepWorker.stats/1 compte ce qu’il a pris, achevé, échoué par classe et abandonné par motif.

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

Ce que fait chaque construction, le protocole de lease et l’API des exécutions sont décrits dans Workflows, Workers et Exécutions.

Scores et 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)

Relire ce qui a été enregistré

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

Les processus de courte durée doivent vider la file

Un script ou une tâche qui se termine vite perd tout ce qui n’avait pas été envoyé. Mcpd.shutdown/1 vide la file et arrête le worker, et peut être appelé deux fois sans risque.

La suite

• Tous les comportements que ce SDK partage avec les cinq autres : le contrat des SDK.
• Ce que porte un span : Modèle de données.
• Mettre ensuite la main sur l’exécution qui vous intéresse : Retrouver une exécution.