본문으로 건너뛰기
mcpd
데몬 없음
소스

Elixir

워커는 GenServer가, 컨텍스트는 프로세스가 맡고, 의존성 트리에는 아무것도 더하지 않습니다.

설치

mix.exs에 추가하십시오. 딸려 오는 것은 없습니다. 

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

그것이 의존성 목록의 전부이고, 그렇게 한 것은 의도입니다. OTP 27에는 JSON 코덱(:json)과 HTTP 클라이언트(:inets)가 들어 있으므로, 시스템을 지켜보는 것이 일인 라이브러리가 그 시스템의 빌드를 바꾸는 이유가 될 필요는 없습니다. 나머지 다섯 SDK도 각자의 언어에서 같은 약속을 합니다.

클라이언트 만들기

모든 옵션은 자기 환경 변수로 물러나고, 코드가 이깁니다. 인자가 없는 Mcpd.start_link()는 로컬 데몬과 대화합니다.

연결 

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

감독 트리에 넣으면 다른 워커와 똑같이 동작합니다. 

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

Supervisor.start_link(children, strategy: :one_for_one)

실행 기록하기

span은 매크로입니다. 그래서 실행은 블록이고, 그 값은 블록의 값입니다.

실행 기록하기 

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

매크로는 호출하기 전에 require 해야 합니다. 여기서 고른 것이 아니라 언어의 규칙입니다. use Mcpd가 그 일을 대신해 줍니다. 

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

본문이 블록이 아니라 이미 함수라면 대신 Mcpd.with_span/4를 호출하십시오. 매크로만 뺀 같은 것입니다.

트리는 스스로 조립됩니다

활성 스팬은 프로세스 딕셔너리에 삽니다. 이것은 지름길이 아니라 Elixir에서만큼은 옳은 저장소입니다. 여기서는 프로세스가 동시성의 단위이므로, 같은 시각에 처리되는 두 요청이 서로의 스팬을 볼 수 없고, 트리를 하나로 묶어 두기 위해 무언가를 모든 함수에 꿰어 넘길 필요도 없습니다.

세 번 호출을 타고 내려간 곳에서 시작한 자식도 스스로 부모를 찾습니다. 

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

알아 둘 만한 결과가 하나 있습니다. Task.async 안에서 연 스팬은 새 트레이스를 시작합니다. 그것이 다른 프로세스이기 때문입니다. 트레이스를 건네주려면 Mcpd.traceparent/0과 Mcpd.continue/2를 쓰십시오. HTTP 경계를 넘을 때 쓰는 것과 같은 짝입니다.

  • Spans

    What ran, nested, with timings

  • Logs

    Lines the run emitted

  • Metrics

    Numbers, with exemplars back to a run

  • Scores

    Machine evaluation

  • Feedback

    What a person thought

  • Aspect notes

    A sentence a model wrote about the run

trace_id Store Search Export

아무것도 막지 않고, 아무것도 예외를 일으키지 않습니다

스팬을 기록하는 것은 워커에 메시지를 보내고 돌아오는 일입니다. 네트워크는 그 워커에서 일어나므로, 에이전트의 지연 시간이 mcpd가 빠른지, 살아 있는지, 아예 돌고 있기는 한지에 달리는 일이 없습니다.

블록 안에서 일어난 예외는 스팬에 오류로 기록된 다음 바뀌지 않은 채로 다시 일어납니다. mcpd가 할 일은 무슨 일이 있었는지 보는 것이지, 그것에 대해 코드가 무엇을 할지 정하는 것이 아닙니다.

큐에는 상한이 있고, 가득 차면 가장 오래된 시그널부터 버립니다. 그 방향은 골라서 정한 것입니다. 데몬에 닿지 않을 때 지금 무슨 일이 벌어지고 있는지 말해 주는 것은 가장 새로운 시그널이고, 데몬이 돌아왔을 때 누군가 보고 싶어 할 것도 그것이기 때문입니다. Mcpd.stats/1은 클라이언트가 무엇을 했는지 답합니다. queued, dropped, capacity, exported, failed, rejected입니다.

log_level:, 또는 MCPD_LOG_LEVEL은 Mcpd.log/4 아래에 바닥을 놓습니다. 기본값은 전부를 남기는 debug입니다. 바닥 아래에 있는 줄은 레코드를 만들기 전에 버리므로, 운영 배포에 남아 있는 디버그 한 줄이 치르는 비용은 비교 한 번입니다. 이 바닥은 로그에만 해당합니다. 스팬, 메트릭, 점수, 피드백은 영향을 받지 않습니다.

페이로드를 프로세스 안에 두기

capture_payloads: false, 또는 MCPD_CAPTURE_PAYLOADS=false는 프롬프트와 컴플리션을 있던 자리에 그대로 둡니다. 스팬은 여전히 시각과 상태와 사용량과 모델과 상관 정보와 오류를 싣고 나가고, input:과 output:과 블록이 담은 값은 큐에 닿기 전에 버려집니다. 그래서 민감한 것은 큐에 들어가지도 않고, 플러시를 타고 나가지도 않습니다. 스팬 하나는 capture_payloads: true로 자기 몫만 달리 정할 수 있습니다.

기록한 것을 검증하기

싱크는 비워 낸 배치가 가는 곳입니다. 기본값은 HTTP 전송 계층입니다. Mcpd.Sink.Recorder는 모든 배치를 메모리에 들고 있어서, 테스트가 데몬 없이 계측이 기록한 스팬과 로그와 메트릭과 점수와 피드백을 다시 읽습니다. Mcpd.Sink.Console은 시그널마다 JSON 한 줄을 stderr로 출력합니다. 직접 만든 싱크는 Mcpd.Sink를 구현합니다. 예외를 일으키는 싱크는 실패한 배치이고, 세어서 보고할 뿐 호출자에게 올라오지 않습니다.

Mcpd.record/3은 이미 끝난 스팬을 준 그대로 큐에 넣습니다. 자기 id와 부모와 시작과 끝을 그대로 두고, 부모를 다시 매기지도, 시각을 다시 매기지도, 샘플링을 다시 결정하지도 않습니다. 다른 트레이싱 시스템의 어댑터나 일괄 가져오기가 들고 있는 것이 바로 그것입니다. id가 OpenTelemetry 형태가 아닌 레코드는 거절하고 세며, 예외를 일으키지 않습니다. 어느 스팬에서든 external_parent_span_id:는 mcpd가 아닌 트레이싱 시스템의 부모를 상관 정보로만 실어 나릅니다.

지속성 실행

같은 클라이언트가 데몬의 지속성 실행도 다룹니다. Mcpd.Workflow는 정의를 만들고 단계 본문을 인자 셋짜리 함수로 들고 있습니다. Mcpd.StepWorker는 그 정의를 등록하고, 리스를 롱 폴링으로 기다렸다가 각각을 자기 프로세스 하나에서 처리합니다. Mcpd.start_run/3, Mcpd.watch_run/3, Mcpd.publish_event/4와 나머지는 실행 표면을 하나씩 그대로 옮겨 놓은 것입니다. 본문은 단계의 예산 안에서 다시 시도되도록 class: :retryable을 붙인 Mcpd.StepException을 일으키고, Mcpd.StepContext.span/4를 통해 자기 단계 아래에 스팬을 기록합니다. 본문이 닫는 모든 스팬은 자기 사용량을 그 단계의 사용량에 더하고, 마지막으로 본 모델이 이깁니다. 그래서 데몬은 트레이스가 보여 주는 그대로 값을 매깁니다. set_usage/2가 명시적인 덮어쓰기입니다. 워커에는 on_error 옆에 on_lease, on_step_start, on_step_end, on_conflict가 있고, StepWorker.stats/1은 확보한 것, 완료한 것, 분류별 실패, 사유별 포기를 셉니다.

워크플로 하나 

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

각 구성 요소가 무엇을 하는지, 리스 프로토콜, 실행 API는 워크플로, 워커, 실행에 있습니다.

점수와 피드백

점수와 피드백 

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

다시 읽기

실행 읽기 

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

짧게 사는 프로세스는 플러시해야 합니다

빨리 끝나는 스크립트나 태스크는 아직 보내지 않은 것을 잃습니다. Mcpd.shutdown/1은 플러시하고 워커를 멈추며, 두 번 호출해도 안전합니다.

로컬 데몬을 상대로 예제 돌리기

mcpd serve --data ./mcpd.db elixir examples/elixir/quickstart.exs

다음