Go

Kurulum

Go 1.21 veya üstü. Üçüncü taraf bağımlılığı yok: taşıyıcı, standart kütüphanedeki net/http ve compress/gzip. Yani mcpd eklemek modül grafiğinize başka hiçbir şey eklemez.

İstemciyi oluşturma

Süreç başına bir istemci. Her ayarın bir seçeneği ve bir ortam değişkeni vardır, böylece aynı kod tek bir düzenleme olmadan hem geliştirmede hem üretimde çalışır. Hiç seçenek verilmeyen mcpd.New(), yerel bir daemon ile çalışır.

m := mcpd.New(
	mcpd.WithEndpoint("http://localhost:4317"),
	mcpd.WithProject("support"),
	mcpd.WithEnvironment("development"),
	mcpd.WithService("support-agent", "1.4.2"),
)
// Shutdown flushes what is still queued and stops the background worker.
// A short-lived process that skips it loses whatever had not been sent.
defer m.Shutdown(ctx)

Kod ortam değişkenine, ortam değişkeni de varsayılana üstün gelir.

WithEnabled(false) kaydı, yine de geçerli span'ler ve kimlikler döndüren boş bir işleme çevirir; böylece bir test, çağrı yerlerini değiştirmeden mcpd'yi kapatabilir. Okuma metotları çalışmayı sürdürür, çünkü ölçümlemeyi kapatmak bir sorguyu sessizce boş bir yanıta çevirmemelidir.

WithCapturePayloads(false) prompt'ları ve tamamlamaları sürecin içinde tutar: span'ler yine zamanlamasını, durumunu, token kullanımını, modelini, ilişkilendirmesini ve hatalarını taşır, input ile output ise kuyruğa varmadan atılır. Tek bir span, WithSpanCapturePayloads ile kendisi için tersini söyleyebilir; iki yönde de.

WithSink, yığınların gittiği yeri değiştirir. RecordingSink onları bellekte tutar, böylece bir test ölçümlemenin ne kaydettiğini sunucu olmadan doğrular; ConsoleSink ise sinyal başına tek bir JSON satırı yazar. Başarısız olan bir sink Stats() içinde sayılır ve sinyalleri kaydeden koda hiç ulaşmaz. Record, çoktan bitmiş bir span'i kimlikleri ve zamanlarıyla birlikte verildiği gibi kuyruğa alır; bozuk olan reddedilir ve sayılır, asla hata olarak döndürülmez. WithExternalParent ise başka bir izleme sisteminden gelen üstü yalnızca ilişkilendirme için taşır.

Bir çalıştırmayı kaydetme

Start yeni bir context ve bir span döndürür, defer span.End() de onu kapatır. Go'da kapsamlı biçim bu ikilidir: defer hem erken bir dönüşte hem de bir panik hâlinde çalışır, yani aklınıza gelmeyen bir akış yolu bir span'i açık bırakamaz.

ctx, span := m.Start(ctx, "retrieve_docs", mcpd.WithSpanKind(mcpd.SpanKindRetrieval))
defer span.End()

// Buradan aşağıda bu ctx'i alan her şey span'in altına düşer ve End
// idempotenttir, yani bir defer ile açık bir çağrı süre konusunda
// çelişemez.

// handleTicket is the agent. It drafts an answer with a model, tries to send
// it with a tool, and fails on the send so the error path is recorded too.
func handleTicket(ctx context.Context, m *mcpd.Client, question string) string {
	ctx, run := m.Start(ctx, "support_agent",
		mcpd.WithSpanKind(mcpd.SpanKindAgentRun),
		mcpd.WithEntity(mcpd.EntityKindAgent, "support-agent", "Support agent"),
		mcpd.WithSession("sess-4821"),
		mcpd.WithUser("u-317"),
		mcpd.WithTags("support", "billing"),
		mcpd.WithInput(question),
	)
	defer run.End()

	m.Log(ctx, mcpd.LogLevelInfo, "ticket received", map[string]any{"channel": "email"})

	answer := draftAnswer(ctx, m, question)
	run.SetOutput(answer)

	if err := sendReply(ctx, m, answer); err != nil {
		// The span is now an error with a message, which is what makes it
		// worth reading. End below records the failed status.
		run.RecordError(err, mcpd.ErrorDomain("email"), mcpd.ErrorRetryable(true))
	}
	return run.TraceID()
}

Bağlam

Start, context'te hangi span varsa onun altına bir alt span açar; hiçbiri yoksa bir kök açar. Trace her zaman bir kök başlatır, span.Child(ctx, name) ise span'i bir context içinde taşımadığınızda belirli bir span'in altına alt span başlatır. Burada ikinci bir taşıma düzeneği icat edilmez: üst span, zaten geçirmekte olduğunuz ctx içindedir.

func handle(ctx context.Context, m *mcpd.Client, q string) string {
    ctx, run := m.Start(ctx, "support_agent", mcpd.WithSpanKind(mcpd.SpanKindAgentRun))
    defer run.End()

    // draft'a span değil ctx geçilir. Düzeneğin tamamı budur.
    return draft(ctx, m, q)
}

func draft(ctx context.Context, m *mcpd.Client, q string) string {
    _, span := m.Start(ctx, "draft_answer", mcpd.WithSpanKind(mcpd.SpanKindModelGeneration))
    defer span.End()
    return "..."
}

Servis sınırının ötesinde Inject, giden başlıklara bir W3C traceparent yazar; Extract ise gelen başlıklardaki değeri okuyup bir context'e koyar. İçinde böyle bir değer olmayan başlıklar olduğu gibi döndürülür, yani okunamayan bir başlık, onu taşıyan isteği başarısız kılmak yerine yeni bir trace başlatır. Yukarı akıştaki örnekleme kararı kalıtılır, böylece bir trace'in iki yarısı birbiriyle uyuşur.

// Giden: etkin trace bağlamını isteğin üzerine yazın.
req, _ := http.NewRequestWithContext(ctx, http.MethodPost, url, body)
mcpd.Inject(ctx, req.Header)

// Gelen: onu geri okuyup aynı trace'i sürdürün.
func handler(w http.ResponseWriter, r *http.Request) {
    ctx, span := m.Start(mcpd.Extract(r.Context(), r.Header), "handle_request")
    defer span.End()
}

Model, kullanım ve hatalar

mcpd alım sırasında token sayılarını fiyatlandırır, dolayısıyla trace'te gördüğünüz maliyet size faturalanan maliyettir. RecordError bir dizge değil bir error alır; kaydedilen bir başarısızlığın üzerinde bir mesaj olmasını garanti eden şey budur. End de durumu ok yerine error olarak kaydeder.

span.SetModel("gpt-4o-mini", "openai")
span.SetTokens(812, 96)

// Ya da tam döküm, böylece önbellekten gelen bir prompt öyle fiyatlandırılır.
span.SetUsage(mcpd.TokenUsage{
    InputTokens:  ptr(uint64(812)),
    OutputTokens: ptr(uint64(96)),
    InputDetails: &mcpd.InputTokenDetails{CacheRead: ptr(uint64(640))},
})

// RecordError bir error alır, yani kaydedilen hata her zaman bir mesaj taşır.
span.RecordError(err,
    mcpd.ErrorKind("SMTPTemporaryFailure"),
    mcpd.ErrorRetryable(true),
    mcpd.ErrorStack(mcpd.Stack()),
)

Span'in geri kalan metotları şunlardır: SetInput, SetOutput, SetMetadata, SetAttributes, SetTags, SetStatus, SetSession, SetThread, SetUser, Event ve EndWithStatus. Aynı alanlar, span başlarken With... seçenekleriyle de verilebilir.

Kalıcı çalıştırmalar

Daemon'ın kalıcı çalıştırmalarını da aynı istemci karşılar. NewWorkflow tanımı kurar ve adım gövdelerini saklar; bir Worker bu tanımı kaydeder, lease'ler için uzun yoklama yapar ve her birini bir goroutine üzerinde karşılar; StartRun, WatchRun, PublishEvent 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 mcpd.Retryable(err, wait) döndürür; handler'ın ctx değerinden başlatılan bir span de çalıştırmanın trace'inde kendi adımının altına düşer. Böyle bir span'in harcadığı, adımın kullanımıdır: span'ler bittikçe sayılar toplanır, en son görülen model geçerli olur ve SetUsage kendi verdiği alanları geçersiz kılar. Worker içeriden WorkerOnLease, WorkerOnStepStart, WorkerOnStepEnd, WorkerOnConflict ve worker.Stats() ile izlenebilir.

support := mcpd.NewWorkflow("support_ticket").
	Step("classify", func(ctx context.Context, step *mcpd.StepContext) (any, error) {
		var in struct {
			Ticket string `json:"ticket"`
		}
		if err := step.Input(&in); err != nil {
			return nil, err
		}
		kind := "bug"
		if strings.HasPrefix(in.Ticket, "Q") {
			kind = "question"
		}
		return map[string]any{"kind": kind}, nil
	}).
	Agent("draft", func(ctx context.Context, step *mcpd.StepContext) (any, error) {
		// Your own model call. A span started from ctx lands under this step,
		// and what it spent is the step's usage: the daemon prices the sum.
		_, span := m.Start(ctx, "model_call", mcpd.WithModel("claude-sonnet-4", "anthropic"))
		defer span.End()
		span.SetTokens(812, 37)
		return "an answer", nil
	}, mcpd.Input(map[string]any{"kind": "${steps.classify.output.kind}"})).
	WaitFor("approval", "approved", "${input.ticket}", 7*24*time.Hour,
		mcpd.Guard(mcpd.When("steps.classify.output.kind", "eq", "bug"))).
	Step("send", func(ctx context.Context, step *mcpd.StepContext) (any, error) {
		if err := deliver("the draft", step.IdempotencyKey); err != nil {
			return nil, mcpd.Retryable(err, 2*time.Second)
		}
		return map[string]any{"sent": true}, nil
	}, mcpd.Input(map[string]any{"text": "${steps.draft.output}"}), mcpd.AtMostOnce(),
		mcpd.Retry(mcpd.RetryPolicy{
			MaxAttempts: 3, Backoff: mcpd.BackoffExponential, Initial: 500 * time.Millisecond,
		})).
	Output(map[string]any{"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

Score, min ve max değerlerini sıradan konumsal argümanlar olarak alır, yani ölçek atlanamaz. Çıplak bir 0.86, hangi aralıkta durduğunu bilmeden hiçbir şey ifade etmez. Geri bildirim bir insanın yargısıdır ve puandan ayrı durur; çünkü ikisi farklı sorulara yanıt verir ve asla birlikte ortalanmamalıdır.

// A machine evaluation of the run. min and max are required: 0.86 means
// nothing until you know the scale it sits on.
m.Score(ctx, mcpd.TraceTarget(traceID), "answer_relevance", 0.86, 0, 1,
	mcpd.ScoreReason("Answered the billing question and cited the invoice."),
	mcpd.ScoreVersion("2"),
)

// A person's judgement of the same run. Wire this to your product's
// thumbs-up control and it lands on the exact trace that produced the
// answer.
m.Feedback(ctx, mcpd.TraceTarget(traceID), mcpd.FeedbackKindThumbs, "up",
	mcpd.FeedbackComment("Fixed it first time."),
	mcpd.FeedbackAuthor("u-317", "end_user"),
)

Neyin değerlendirildiğini adlandırmak için mcpd.TraceTarget(id) ya da mcpd.SpanTarget(id, spanID) verin; context'te hangi span varsa onu kullanmak için ise boş bir mcpd.Target{} verin.

Beş sinyalin hepsi, tek bir kimlik kümesi

Log, Metric, Score ve Feedback metotlarının hepsi context alır, yani beş sinyalin hepsi aynı trace ve span id'lerini taşır. Bir grafikteki sıçramadan onu doğuran çalıştırmayı açabilmenizi sağlayan şey budur.

Flush ve kapatma

Start ve End bellekteki bir kuyruğa ekleyip döner; yığınlamayı ve göndermeyi arka plandaki bir goroutine yapar. Flush(ctx) kuyrukta bekleyeni hiçbir şeyi durdurmadan gönderir. Shutdown(ctx) flush eder, worker'ı durdurur ve idempotenttir; New çağrısının yanına bir defer m.Shutdown(ctx) koymak bu yüzden doğru alışkanlıktır.

Serverless bir fonksiyonda runtime, handler döner dönmez süreci dondurur ve worker goroutine'i bir daha çalışmaz, yani hâlâ kuyrukta bekleyen bir yığın kaybolur. Dönmeden önce flush edin.

func Handler(ctx context.Context, event Event) (Response, error) {
    ctx, run := m.Start(ctx, "answer")
    answer := agent(ctx, event.Body)
    run.SetOutput(answer)
    run.End()

    // Yığın burada hâlâ bellekte. Runtime, handler döner dönmez bu süreci
    // dondurur ve worker goroutine'i bir daha çalışmaz.
    if err := m.Flush(ctx); err != nil {
        log.Printf("mcpd flush: %v", err)
    }
    return Response{Body: answer}, nil
}

Güvenebilecekleriniz

• Bir şey kaydeden hiçbir metot hata döndürmez ve panik etmez; nil bir istemcide bile. Başarısız bir dışa aktarım, flush aralığı başına en çok bir kez loglanır ve atılır. Gözlemlenebilirlik, gözlediği şeyi asla devirmemelidir.
• Kayıt bloklamaz, yani ajanınızın gecikmesi mcpd'nin hızlı ya da ayakta olmasına hiç bağlı değildir.
• Bellek sınırlıdır. Kuyruk WithMaxQueue kadar sinyal tutar; bunun ötesinde en eskiler atılır ve sayılır. Bu sayı, dışa aktarılanlar, vazgeçilenler ve Record tarafından reddedilenlerle birlikte Stats() içinde görünür. Sınırsız bir tampon, bir ağ kesintisini bellek yetersizliğinden ölüme çevirir.
• Örnekleme kararı bir kez, kökte, trace id'sinin özeti alınarak verilir ve her alt span'e kalıtılır. Böylece bir filodaki her süreç aynı trace hakkında aynı kararı verir.
• Bitmemiş bir span de dışa aktarılır. Uzun süren span'ler başladıklarında çıkar ve bittiklerinde güncellenir, yani yolun ortasında çöken bir çalıştırma görünür olur. Bir sonraki flush'tan önce biten bir span ise bir kez, bitmiş hâliyle gönderilir.
• Kodlanamayan bir değer, çağrınızı başarısız kılmak yerine atılır ve loglanır; çünkü bir payload'daki serileştirme hatası, ajanınızda bir başarısızlık olarak yüzeye çıkmamalıdır.

Geri okuma

Okuma metotları HTTP API'yi yansıtır ve üretilmiş tipleri döndürür. Hata döndüren tek metotlar bunlardır, çünkü çağıran taraf bir yanıt bekliyordur.

// Recording is asynchronous, so wait for the batch to land before reading
// it back. Outside an example you would never need this.
if err := m.Flush(ctx); err != nil {
	log.Printf("flush: %v", err)
}

detail, err := m.GetTrace(ctx, traceID, "scores", "feedback")
if err != nil {
	log.Fatalf("read the run back: %v", err)
}
fmt.Printf("%s: %s in %s, %d spans\n",
	detail.Trace.TraceID, detail.Trace.Status, duration(detail.Trace.DurationNs), len(detail.Spans))
for _, span := range detail.Spans {
	fmt.Printf("  %-18s %-18s %s\n", span.Kind, span.Name, span.Status)
}

summary, err := m.AnalyticsSummary(ctx, time.Now().Add(-time.Hour), time.Now(), "model")
if err != nil {
	log.Fatalf("read the summary: %v", err)
}
fmt.Printf("last hour: %d runs, %.1f%% failed, %.4f %s\n",
	summary.TraceCount, summary.ErrorRate*100, summary.TotalCost, summary.Currency)

Çalıştırma bulma filtreleri eksiksiz anlatır. Veri modeli bu tiplerin taşıdığı her alanı listeler.