Go
由 context.Context 携带活动的 span,和其他每一个 Go 库的做法一样。没有第三方依赖,也没有哪个记录调用会失败。
安装
Go 1.21 及以上。没有第三方依赖:传输层用的是标准库里的 net/http
和 compress/gzip,所以加上 mcpd 不会往你的模块图里再添别的东西。
go get github.com/mcpd/mcpd/sdks/go 构造客户端
一个进程一个客户端。每一项设置都有一个选项和一个环境变量,所以同一份代码不用
改一个字,就能在开发环境和生产环境里跑。完全不带选项的
mcpd.New() 对着一个本地的守护进程就能用。
连接
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) 代码优先于环境变量,环境变量优先于默认值。
| 选项 | 环境变量 | 默认值 |
|---|---|---|
WithEndpoint | MCPD_ENDPOINT | http://localhost:4317 |
WithAPIKey | MCPD_API_KEY | 无 |
WithProject | MCPD_PROJECT | default |
WithEnvironment | MCPD_ENVIRONMENT | 无 |
WithService | MCPD_SERVICE_NAME、MCPD_SERVICE_VERSION | 无 |
WithSampleRate | MCPD_SAMPLE | 1.0 |
WithBatchSize | MCPD_BATCH_SIZE | 128 |
WithFlushInterval | MCPD_FLUSH_INTERVAL_MS | 2s |
WithMaxQueue | MCPD_MAX_QUEUE | 8192 |
WithTimeout | MCPD_TIMEOUT_MS | 10s |
WithEnabled | MCPD_ENABLED | true |
WithCapturePayloads | MCPD_CAPTURE_PAYLOADS | true |
WithLogLevel | MCPD_LOG_LEVEL | debug |
WithEnabled(false) 把记录变成空操作,但仍然返回合法的 span 和
id,所以一个测试不用改任何调用处就能把 mcpd 关掉。读取方法照常工作,因为
关掉接入,不应该悄悄把一次查询变成一个空答案。
WithCapturePayloads(false) 把提示词和补全留在进程里:span 照样
带着它的时间、状态、用量、模型、关联信息和错误,而 input 和
output 在进入队列之前就被丢掉。单个 span 可以用
WithSpanCapturePayloads 说自己不一样,两个方向都可以。
WithSink 换掉批次去的地方。RecordingSink 把它们留在
内存里,于是一个测试不需要服务器就能断言接入代码记下了什么;
ConsoleSink 则每个信号写一行 JSON。失败的 sink 会计入
Stats(),绝不会传到记录这些信号的那段代码里。
Record 把一个已经结束的 span 排进队列,连同 id 和时间,按给的
样子来;格式不对的会被拒绝并计数,绝不作为一个错误返回。
WithExternalParent 携带的是来自另一套 tracing 系统的父级,
只用于关联。
记录一次运行
Start 返回一个新的 context 和一个 span,
defer span.End() 把它关掉。这一对就是 Go 里带作用域的那种写法:
defer 在提前返回时会跑,在 panic 时也会跑,所以一个 span 不会因为某条你没
想到的控制流而一直开着。
带作用域的写法
ctx, span := m.Start(ctx, "retrieve_docs", mcpd.WithSpanKind(mcpd.SpanKindRetrieval))
defer span.End()
// Everything that takes this ctx from here down becomes a child of the span,
// and End is idempotent, so a defer and an explicit call cannot disagree about
// the duration. 记录一次运行
// 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()
} 上下文
Start 在 context 里那个 span 下面开一个子 span,里面没有时就开
一个根。Trace 永远开一个根;当你没有用 context 携带某个 span
时,span.Child(ctx, name) 在那个指定的 span 下面开一个子 span。
这里没有发明第二套传播机制:父级就在你本来就在传的那个 ctx 里。
嵌套
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 is handed the ctx, not the span. That is the whole mechanism.
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 "..."
}
跨过服务边界时,Inject 把一个 W3C traceparent 写到
出去的请求头上,Extract 则从进来的请求头里读一个出来放进
context。请求头里没有它时原样返回,所以一个读不出来的头只会开一条新的
trace,而不会让携带它的那个请求失败。上游的采样决定会被继承,于是一条 trace
的两半是一致的。
跨过服务边界
// Outbound: write the active trace context onto the request.
req, _ := http.NewRequestWithContext(ctx, http.MethodPost, url, body)
mcpd.Inject(ctx, req.Header)
// Inbound: read it back and continue the same trace.
func handler(w http.ResponseWriter, r *http.Request) {
ctx, span := m.Start(mcpd.Extract(r.Context(), r.Header), "handle_request")
defer span.End()
} 模型、用量与错误
mcpd 在摄取时就是按 token 数定价的,所以 trace 上的成本就是你被计费的那个
成本。RecordError 收的是一个 error,而不是一个
字符串,正是这一点保证了记下来的失败一定带着一条消息。之后 End
记的状态就是 error,而不是 ok。
一个 span 带着什么
span.SetModel("gpt-4o-mini", "openai")
span.SetTokens(812, 96)
// Or the full breakdown, so a cached prompt is priced as a cached prompt.
span.SetUsage(mcpd.TokenUsage{
InputTokens: ptr(uint64(812)),
OutputTokens: ptr(uint64(96)),
InputDetails: &mcpd.InputTokenDetails{CacheRead: ptr(uint64(640))},
})
// RecordError takes an error, so a recorded failure always has a message.
span.RecordError(err,
mcpd.ErrorKind("SMTPTemporaryFailure"),
mcpd.ErrorRetryable(true),
mcpd.ErrorStack(mcpd.Stack()),
)
span 其余的方法是 SetInput、SetOutput、
SetMetadata、SetAttributes、SetTags、
SetStatus、SetSession、SetThread、
SetUser、Event 和 EndWithStatus。
同样这些字段,也可以在 span 开始时通过那些 With... 选项设置。
持久化运行
同一个客户端也服务守护进程的持久化运行。NewWorkflow 构建定义并
持有那些步骤主体;一个 Worker 注册它,用长轮询要租约,并在一个
goroutine 上逐份服务,而 StartRun、WatchRun、
PublishEvent 以及其余的方法与运行接口一一对应。步骤主体返回
mcpd.Retryable(err, wait),表示要在这个步骤的预算内重试;而从
处理函数的 ctx 开出来的 span,会落在这次运行的 trace 里它自己
那个步骤下面。这样一个 span 花掉的,就是这个步骤的用量:这些计数在 span
结束时累加,看到的最后一个模型胜出,而 SetUsage 会覆盖它设置
的那些字段。这个 worker 可以从内部观察,用的是 WorkerOnLease、
WorkerOnStepStart、WorkerOnStepEnd、
WorkerOnConflict 和 worker.Stats()。
一个工作流
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}"}) 每种构造做什么、租约协议和运行 API,在 工作流、 Worker 和 运行这几页上。
评分与反馈
Score 把 min 和 max 当作普通的位置
参数收下,所以量程漏不掉。一个光秃秃的 0.86,在你知道它所在的范围之前没有
任何意义。反馈是人的判断,和评分分开存放,因为这两者回答的是不同的问题,
绝不能被平均到一起。
评分与反馈
// 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"),
)
传 mcpd.TraceTarget(id) 或
mcpd.SpanTarget(id, spanID) 来指明被评判的是什么,或者传一个
空的 mcpd.Target{},用 context 里的那个 span。
五种信号,一套 id
Log、Metric、Score 和
Feedback 都收下这个 context,所以五种信号带的是同样的 trace id
和 span id。正是这一点,让图上的一个尖峰能打开引起它的那次运行。
-
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
刷出与关闭
Start 和 End 往内存里的一个队列追加一条然后返回;
一个后台 goroutine 负责攒批和发送。Flush(ctx) 把排着队的东西
发出去,不停下任何东西。Shutdown(ctx) 先刷出,再停掉 worker,
而且是幂等的,这也是为什么在 New 旁边写一句
defer m.Shutdown(ctx) 是个好习惯。
在一个 serverless 函数里,处理函数一返回,运行时就把进程冻住,worker goroutine 再也不会跑,于是还排在队列里的那个批次就丢了。请在返回之前刷出。
一个 serverless 处理函数
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()
// The batch is still in memory here. The runtime freezes this process the
// moment the handler returns, and the worker goroutine never runs again.
if err := m.Flush(ctx); err != nil {
log.Printf("mcpd flush: %v", err)
}
return Response{Body: answer}, nil
} 你可以依赖的东西
- 任何记录东西的方法都不返回错误,也不 panic,在一个 nil 客户端上调用也是 如此。一次失败的导出,每个刷出间隔最多记一条日志,然后丢掉。可观测性绝不 能把它观测的那个东西搞垮。
- 记录不阻塞,所以你的智能体的延迟,从不取决于 mcpd 快不快,或者活没活着。
-
内存有上限。队列最多放
WithMaxQueue个信号;超过之后最旧的 会被丢弃并计数,在Stats()里看得到,旁边还有导出了多少、 放弃了多少,以及Record拒了多少。一个没有上限的缓冲区,会把 一次网络故障变成一次内存耗尽被杀。 - 采样决定只在根上做一次,做法是对 trace id 取哈希,然后由每个子级继承。 因此一个集群里的每个进程,对同一条 trace 都会做出同样的决定。
- 没结束的 span 一样会导出。长时间运行的 span 在开始时就发出去,在结束时 更新,所以一次跑到半路崩掉的运行是看得见的。在下一次刷出之前就结束的 span 只发一次,发的是它结束后的样子。
- 编码不了的值会被丢弃并记进日志,而不是让你的调用失败,因为一个载荷里的 序列化错误,绝不能冒出来变成你的智能体里的一次失败。
把它读回来
读取方法对应 HTTP API,返回由规范生成的类型。它们是唯一一组会返回错误的 方法,因为调用方正等着一个答案。
检索与分析
// 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)