Elixir

Instalación

Añádelo a mix.exs. No trae nada consigo.

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

Esa es toda la lista de dependencias, y es a propósito. OTP 27 trae un códec JSON (:json) y un cliente HTTP (:inets), así que una biblioteca cuyo trabajo es vigilar tu sistema no tiene por qué convertirse en un motivo para que cambie la compilación de tu sistema. Los otros cinco SDK hacen la misma promesa en sus propios lenguajes.

Construir el cliente

Cada opción recurre a su variable de entorno, y el código gana. Mcpd.start_link() sin argumentos habla con 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"
  )

Ponlo en tu árbol de supervisión y se comporta como cualquier otro worker:

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

Supervisor.start_link(children, strategy: :one_for_one)

Registrar una ejecución

span es una macro, así que una ejecución es un bloque y su valor es el valor del bloque.

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

Las macros se tienen que requerir antes de poder llamarlas, que es una regla del lenguaje y no una decisión de aquí. use Mcpd lo hace por ti:

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 el cuerpo ya es una función en vez de un bloque, llama a Mcpd.with_span/4. Es lo mismo sin la macro.

El árbol se arma solo

El span activo vive en el diccionario de proceso, y en Elixir en concreto ese es el almacén correcto, no un atajo. Aquí el proceso es la unidad de concurrencia, así que dos solicitudes atendidas a la vez no pueden ver el span de la otra, y no hay que ir pasando nada por cada función para mantener el árbol unido.

Un hijo que se abre tres llamadas más abajo encuentra su padre por su cuenta:

Mcpd.span mcpd, "agent", kind: :agent_run do
  plan(mcpd)          # opens a child of "agent" without being told
end

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

La consecuencia que conviene saber: un span abierto dentro de Task.async inicia una traza nueva, porque ese es otro proceso. Pasa la traza al otro lado con Mcpd.traceparent/0 y Mcpd.continue/2, el mismo par que usarías a través de un límite HTTP.

Nada bloquea, y nada lanza

Registrar un span envía un mensaje al worker y vuelve. La red ocurre en ese worker, así que la latencia de tu agente nunca depende de que mcpd sea rápido, ni de que esté en pie, ni de que esté corriendo siquiera.

Una excepción lanzada dentro de un bloque se registra en el span como error y después se relanza sin cambios. El trabajo de mcpd es ver qué pasó, nunca decidir qué hace tu código al respecto.

La cola está acotada, y cuando se llena se descartan las señales más antiguas. Esa dirección está elegida: cuando el daemon está inalcanzable, las señales más nuevas describen lo que está pasando ahora, y esas son las que alguien va a querer cuando vuelva. Mcpd.stats/1 responde qué ha hecho el cliente: queued, dropped, capacity, exported, failed y rejected.

log_level:, o MCPD_LOG_LEVEL, pone un suelo bajo Mcpd.log/4. El valor por defecto es debug, que lo guarda todo. Una línea por debajo del suelo se descarta antes de construir el registro, así que una línea de debug en un despliegue de producción cuesta una comparación. El suelo cubre solo los logs; los spans, las métricas, las puntuaciones y el feedback no se ven afectados.

Mantener los payloads dentro del proceso

capture_payloads: false, o MCPD_CAPTURE_PAYLOADS=false, mantiene los prompts y las completions donde están. Los spans siguen saliendo con sus tiempos, su estado, su uso, su modelo, su correlación y sus errores; input:, output: y el valor capturado de un bloque se descartan antes de llegar a la cola, así que no se encola nada sensible y nada sensible sale en un flush. Un span puede decir lo contrario para sí mismo con capture_payloads: true.

Comprobar lo que registraste

Un sink es adonde va un lote vaciado. El transporte HTTP es el valor por defecto; Mcpd.Sink.Recorder guarda cada lote en memoria, así que una prueba vuelve a leer los spans, los logs, las métricas, las puntuaciones y el feedback que registró la instrumentación sin ningún daemon, y Mcpd.Sink.Console imprime una línea JSON por señal en stderr. Un sink propio implementa Mcpd.Sink; uno que lanza es un lote fallido, contado e informado, nunca lanzado hacia quien llama.

Mcpd.record/3 encola un span que ya terminó, tal cual: con sus propios ids, su padre, su inicio y su fin, sin que se le cambie el padre, ni los tiempos, ni el muestreo. Eso es lo que tiene en la mano un adaptador de otro sistema de trazas o una importación por lotes. Un registro cuyos ids no tienen la forma de OpenTelemetry se rechaza y se cuenta, y nunca lanza. external_parent_span_id: en cualquier span lleva un padre de un sistema de trazas que no es mcpd, solo para correlacionar.

Ejecuciones duraderas

El mismo cliente sirve las ejecuciones duraderas del daemon. Mcpd.Workflow construye la definición y guarda los cuerpos de los pasos como funciones de aridad tres; un Mcpd.StepWorker la registra, hace long polling para obtener leases y atiende cada uno en un proceso propio, y Mcpd.start_run/3, Mcpd.watch_run/3, Mcpd.publish_event/4 y el resto reflejan la superficie de ejecuciones una a una. Un cuerpo lanza Mcpd.StepException con class: :retryable para que se reintente dentro del presupuesto del paso, y registra spans bajo su paso a través de Mcpd.StepContext.span/4. Cada span que cierra el cuerpo suma su uso al del paso, y gana el último modelo visto, de modo que el daemon cobra lo que muestra la traza; set_usage/2 es la anulación explícita. on_lease, on_step_start, on_step_end y on_conflict están junto a on_error en el worker, y StepWorker.stats/1 cuenta lo que ha tomado, completado, fallado por clase y abandonado por 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}"})

Qué hace cada construcción, el protocolo de leases y la API de ejecuciones están en Workflows, Workers y Ejecuciones.

Puntuaciones y 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)

Volver a leerlo

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

Los procesos de vida corta tienen que hacer flush

Un script o una tarea que sale rápido pierde todo lo que no se hubiera enviado. Mcpd.shutdown/1 hace flush y detiene el worker, y se puede llamar dos veces sin problema.

A continuación

• Cada comportamiento que este SDK comparte con los otros cinco: el contrato de los SDK.
• Qué lleva un span: Modelo de datos.
• Encontrar después la ejecución que te importa: Encontrar una ejecución.