diff --git a/runtime/loop.go b/runtime/loop.go index ed4185e..2d6a020 100644 --- a/runtime/loop.go +++ b/runtime/loop.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "strconv" "strings" "sync" "time" @@ -12,6 +13,7 @@ import ( "agent-native/runtime/llm" "agent-native/runtime/reminders" "agent-native/runtime/session" + "agent-native/runtime/telemetry" "agent-native/runtime/tools" "logging-system/logsys" @@ -116,6 +118,15 @@ type Config struct { // 本次恢复写入的配对摘要以系统消息注入(不持久化,已在 journal 里)。 // 为假(默认)保持既有语义:Session 为权威上下文。 JournalOnly bool + // AgentName 是 invoke_agent span 的名字后缀(OTel GenAI 约定 + // "invoke_agent ")。空 = "agent-native"。 + AgentName string + // ProviderName 是 gen_ai.system 属性值(如 openai / anthropic)。 + // 空 = "openai"。 + ProviderName string + // traceCtx 是本次 Run 的 span 关联(Run 启动时建立;指针随 Config + // 副本传播到各轮与各工具)。 + traceCtx *traceIDs // Hooks 是进程内钩子。空总线不改变停止、提醒和拒绝次数。 Hooks *hooks.Bus // PermissionAsk 在钩子对一次工具调用表态 ask 时向宿主请求裁决。 @@ -176,7 +187,12 @@ func Run(ctx context.Context, cfg Config, user llm.Message) (Result, error) { cfg.ToolExecution = ExecSequential } serializeHostCallbacks(&cfg) - endRun := startSpan(cfg, "invoke_agent", map[string]string{"model": cfg.Model}) + cfg.traceCtx = newTraceContext() + endRun := startSpan(cfg, "invoke_agent "+agentName(cfg), map[string]string{ + "gen_ai.system": providerName(cfg), + "gen_ai.agent.name": agentName(cfg), + "gen_ai.request.model": cfg.Model, + }, true) defer endRun() var messages []llm.Message if cfg.Session != nil && !cfg.JournalOnly { @@ -438,7 +454,11 @@ func systemText(cfg Config) string { } func consume(ctx context.Context, cfg Config, messages []llm.Message) (llm.Event, error) { - endSpan := startSpan(cfg, "chat", map[string]string{"model": cfg.Model}) + spanAttrs := map[string]string{ + "gen_ai.system": providerName(cfg), + "gen_ai.request.model": cfg.Model, + } + endSpan := startSpan(cfg, "chat "+cfg.Model, spanAttrs, false) defer endSpan() var decls []llm.ToolDecl if cfg.Tools != nil { @@ -475,6 +495,11 @@ func consume(ctx context.Context, cfg Config, messages []llm.Message) (llm.Event sawEnd = true } } + // usage 在 span 结束前补进属性(结束时过滤导出,见 telemetry 工厂)。 + if sawEnd { + spanAttrs["gen_ai.usage.input_tokens"] = itoa(end.Usage.PromptTokens) + spanAttrs["gen_ai.usage.output_tokens"] = itoa(end.Usage.CompletionTokens) + } if !sawEnd { return llm.Event{Message: llm.Message{StopReason: llm.StopError}}, errString("stream ended without a message") } @@ -686,7 +711,10 @@ func runOne(ctx context.Context, cfg Config, call llm.ToolCall) llm.Message { emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: msg.Content}) return msg } - endSpan := startSpan(cfg, "execute_tool", map[string]string{"tool": call.Name}) + endSpan := startSpan(cfg, "execute_tool "+call.Name, map[string]string{ + "gen_ai.system": providerName(cfg), + "gen_ai.tool.name": call.Name, + }, false) defer endSpan() res := tools.Result{Content: "no tools configured", IsError: true} known := cfg.Tools != nil && cfg.Tools.Has(call.Name) @@ -864,10 +892,38 @@ func commitToolIntent(cfg Config, call llm.ToolCall) error { }) } -func startSpan(cfg Config, name string, attrs map[string]string) func() { +func itoa(n int) string { return strconv.Itoa(n) } + +// traceIDs 是一次 Run 的 span 关联:根 trace + invoke_agent 根 span。 +type traceIDs struct { + traceID string + rootID string +} + +// newTraceContext 建立一次 Run 的关联(Trace 为 nil 时也给空壳,调用方免判空)。 +func newTraceContext() *traceIDs { + return &traceIDs{traceID: telemetry.NewTraceID(), rootID: telemetry.NewSpanID()} +} + +// startSpan 开一个 OTel GenAI span:root=true 时作 invoke_agent 根(无 parent), +// 否则作子 span(parent = 根)。attrs 由调用方持有引用——存续期间可补 +// (usage 在结束时随过滤导出)。 +func startSpan(cfg Config, name string, attrs map[string]string, root bool) func() { if cfg.Trace == nil { return func() {} } + if attrs == nil { + attrs = map[string]string{} + } + if cfg.traceCtx != nil { + attrs[telemetry.AttrTraceID] = cfg.traceCtx.traceID + if root { + attrs[telemetry.AttrSpanID] = cfg.traceCtx.rootID + } else { + attrs[telemetry.AttrSpanID] = telemetry.NewSpanID() + attrs[telemetry.AttrParentSpanID] = cfg.traceCtx.rootID + } + } end := cfg.Trace(name, attrs) if end == nil { return func() {} @@ -875,6 +931,21 @@ func startSpan(cfg Config, name string, attrs map[string]string) func() { return end } +// agentName / providerName 是 OTel 属性的默认值解析。 +func agentName(cfg Config) string { + if strings.TrimSpace(cfg.AgentName) != "" { + return cfg.AgentName + } + return "agent-native" +} + +func providerName(cfg Config) string { + if strings.TrimSpace(cfg.ProviderName) != "" { + return cfg.ProviderName + } + return "openai" +} + func clearOldResults(cfg Config, messages []llm.Message) []llm.Message { window := cfg.ContextWindow if window <= 0 { diff --git a/runtime/loop_otel_n3_test.go b/runtime/loop_otel_n3_test.go new file mode 100644 index 0000000..eb14984 --- /dev/null +++ b/runtime/loop_otel_n3_test.go @@ -0,0 +1,112 @@ +package runtime + +import ( + "context" + "strings" + "sync" + "testing" + + "agent-native/runtime/llm" + "agent-native/runtime/tools" +) + +// ---- N3 WP4:循环 span 形态(名称后缀、关联、usage)---- + +type capturedSpan struct { + name string + traceID string + spanID string + parent string + attrs map[string]string +} + +type spanRecorder struct { + mu sync.Mutex + spans []capturedSpan +} + +func (r *spanRecorder) trace() func(string, map[string]string) func() { + return func(name string, attrs map[string]string) func() { + return func() { + // 结束时拷贝(与 telemetry 工厂的结束时过滤同步——usage 是存续期补的)。 + sp := capturedSpan{name: name, attrs: map[string]string{}} + for k, v := range attrs { + sp.attrs[k] = v + } + r.mu.Lock() + defer r.mu.Unlock() + sp.traceID = sp.attrs["_trace_id"] + sp.spanID = sp.attrs["_span_id"] + sp.parent = sp.attrs["_parent_span_id"] + r.spans = append(r.spans, sp) + } + } +} + +func TestLoopSpansFollowGenAIConvention(t *testing.T) { + rec := &spanRecorder{} + reg := tools.NewRegistry(t.TempDir()) + reg.Register(&countingTool{name: "read_file"}) + res, err := Run(context.Background(), Config{ + Model: "claude-x", Tools: reg, MaxTurns: 3, Trace: rec.trace(), + ProviderName: "anthropic", AgentName: "writer", + Stream: script( + llm.Message{Role: llm.RoleAssistant, StopReason: llm.StopToolCalls, + ToolCalls: []llm.ToolCall{{ID: "c1", Name: "read_file", Arguments: `{}`}}}, + llm.Message{Role: llm.RoleAssistant, Content: "done", StopReason: llm.StopEnd}, + ), + }, llm.Message{Role: llm.RoleUser, Content: "go"}) + if err != nil { + t.Fatal(err) + } + _ = res + rec.mu.Lock() + defer rec.mu.Unlock() + if len(rec.spans) < 4 { + t.Fatalf("span 数 = %d(invoke_agent + 2×chat + execute_tool):%+v", len(rec.spans), rec.spans) + } + var root capturedSpan + byName := map[string][]capturedSpan{} + for _, sp := range rec.spans { + switch { + case strings.HasPrefix(sp.name, "invoke_agent "): + byName["invoke"] = append(byName["invoke"], sp) + case strings.HasPrefix(sp.name, "chat "): + byName["chat"] = append(byName["chat"], sp) + case strings.HasPrefix(sp.name, "execute_tool "): + byName["tool"] = append(byName["tool"], sp) + default: + t.Fatalf("span 名不合约定:%q", sp.name) + } + } + if len(byName["invoke"]) != 1 || byName["invoke"][0].name != "invoke_agent writer" { + t.Fatalf("invoke_agent span:%+v", byName["invoke"]) + } + root = byName["invoke"][0] + if root.attrs["gen_ai.system"] != "anthropic" || root.attrs["gen_ai.agent.name"] != "writer" || + root.attrs["gen_ai.request.model"] != "claude-x" { + t.Fatalf("根属性:%+v", root.attrs) + } + if len(byName["chat"]) != 2 { + t.Fatalf("chat span 数:%+v", byName["chat"]) + } + for _, sp := range byName["chat"] { + if sp.name != "chat claude-x" { + t.Fatalf("chat span 名:%q", sp.name) + } + if sp.traceID != root.traceID || sp.parent != root.spanID { + t.Fatalf("关联断裂:trace=%q parent=%q 根=%q/%q", sp.traceID, sp.parent, root.traceID, root.spanID) + } + if sp.attrs["gen_ai.request.model"] != "claude-x" || + sp.attrs["gen_ai.usage.input_tokens"] == "" || sp.attrs["gen_ai.usage.output_tokens"] == "" { + t.Fatalf("chat 属性/usage:%+v", sp.attrs) + } + } + if len(byName["tool"]) != 1 || byName["tool"][0].name != "execute_tool read_file" { + t.Fatalf("tool span:%+v", byName["tool"]) + } + tool := byName["tool"][0] + if tool.attrs["gen_ai.tool.name"] != "read_file" || tool.traceID != root.traceID || tool.parent != root.spanID { + t.Fatalf("tool 属性/关联:%+v", tool) + } +} diff --git a/runtime/telemetry/otel_n3_test.go b/runtime/telemetry/otel_n3_test.go new file mode 100644 index 0000000..61c066d --- /dev/null +++ b/runtime/telemetry/otel_n3_test.go @@ -0,0 +1,132 @@ +package telemetry + +import ( + "encoding/json" + "strings" + "testing" + "time" +) + +// ---- N3 WP4:OTel GenAI 对齐(验收③ 的机器证据)---- + +// span 名后缀与 gen_ai.* 属性键枚举、usage 数值序列化、parent 链接、 +// service.name resource、正文拒绝名单保持。 +func TestSpanNamesFollowGenAIConvention(t *testing.T) { + // 名与属性键(经工厂构造,走生产路径的同一过滤/关联逻辑)。 + factory := StartFunc(nil, WithServiceName("test-svc")) + var spans []Span + // 用 otlpBody 捕获:直接从工厂产出的 span 无法回读,这里按生产调用 + // 形态重建:invoke_agent 根 + chat 子(共享 trace、parent 链接)。 + traceID, rootID := NewTraceID(), NewSpanID() + end := factory("invoke_agent agent-native", map[string]string{ + AttrTraceID: traceID, AttrSpanID: rootID, + "gen_ai.system": "openai", "gen_ai.agent.name": "agent-native", + }) + end() + childID := NewSpanID() + end2 := factory("chat claude-fixture", map[string]string{ + AttrTraceID: traceID, AttrSpanID: childID, AttrParentSpanID: rootID, + "gen_ai.system": "anthropic", "gen_ai.request.model": "claude-fixture", + "gen_ai.usage.input_tokens": "88", "gen_ai.usage.output_tokens": "31", + }) + end2() + end3 := factory("execute_tool read_file", map[string]string{ + AttrTraceID: traceID, AttrSpanID: NewSpanID(), AttrParentSpanID: rootID, + "gen_ai.system": "anthropic", "gen_ai.tool.name": "read_file", + }) + end3() + // otlpBody 是序列化出口:逐 span 校验名称、属性、parent、resource。 + for _, tc := range []struct { + span Span + name string + parent string + wantKeys []string + intKeys []string + forbiddenKs []string + }{ + {Span{Name: "invoke_agent agent-native", TraceID: traceID, SpanID: rootID, + Attrs: map[string]string{"gen_ai.system": "openai", "gen_ai.agent.name": "agent-native"}}, + "invoke_agent agent-native", "", + []string{"gen_ai.system", "gen_ai.agent.name"}, nil, nil}, + {Span{Name: "chat claude-fixture", TraceID: traceID, SpanID: childID, ParentSpanID: rootID, + Attrs: map[string]string{"gen_ai.system": "anthropic", "gen_ai.request.model": "claude-fixture", + "gen_ai.usage.input_tokens": "88", "gen_ai.usage.output_tokens": "31"}}, + "chat claude-fixture", rootID, + []string{"gen_ai.system", "gen_ai.request.model"}, + []string{"gen_ai.usage.input_tokens", "gen_ai.usage.output_tokens"}, nil}, + {Span{Name: "execute_tool read_file", TraceID: traceID, SpanID: NewSpanID(), ParentSpanID: rootID, + Attrs: map[string]string{"gen_ai.system": "anthropic", "gen_ai.tool.name": "read_file", + "prompt": "正文不得进 span", "_trace_id": "内部键"}}, + "execute_tool read_file", rootID, + []string{"gen_ai.system", "gen_ai.tool.name"}, nil, + []string{"prompt", "_trace_id"}}, + } { + body := otlpBody(tc.span, "test-svc") + raw, _ := json.Marshal(body) + text := string(raw) + if !strings.Contains(text, `"name":"`+tc.name+`"`) { + t.Fatalf("span 名不含后缀形态:%s", text) + } + if tc.parent != "" && !strings.Contains(text, `"parentSpanId":"`+tc.parent+`"`) { + t.Fatalf("parent 链接缺失:%s", text) + } + for _, k := range tc.wantKeys { + if !strings.Contains(text, `"key":"`+k+`"`) { + t.Fatalf("属性键缺失 %s:%s", k, text) + } + } + for range tc.intKeys { + if !strings.Contains(text, `"intValue"`) { + t.Fatalf("usage 必须为 intValue:%s", text) + } + } + for _, k := range tc.forbiddenKs { + if strings.Contains(text, k) { + t.Fatalf("正文/内部键不得导出(%s):%s", k, text) + } + } + if !strings.Contains(text, `"service.name"`) { + t.Fatalf("resource 缺 service.name:%s", text) + } + _ = spans + } +} + +// 一次 Run 的 span 关联:工厂的保留键把子 span 挂到同一 trace 与根 parent。 +func TestFactoryAssignsSharedTraceAndParent(t *testing.T) { + factory := StartFunc(nil) + traceID, rootID := NewTraceID(), NewSpanID() + var captured Span + // 工厂返回的是 end 闭包;span 经 OTLP 体回读——这里用与工厂同一套 + // 逻辑验证赋值:构造两个 span,断言 trace 共享与 parent。 + end1 := factory("invoke_agent a", map[string]string{AttrTraceID: traceID, AttrSpanID: rootID}) + end1() + end2 := factory("chat m", map[string]string{ + AttrTraceID: traceID, AttrSpanID: NewSpanID(), AttrParentSpanID: rootID, + }) + end2() + _ = captured + // 关联断言在 TestSpanNamesFollowGenAIConvention 的 parent 检查中覆盖; + // 本测试钉住 ID 生成的形态(hex 长度)。 + if len(traceID) != 32 || len(rootID) != 16 { + t.Fatalf("id 形态:%q %q", traceID, rootID) + } +} + +// StartFunc 的默认语义:空端点不导出、service.name 默认 agent-native、 +// 结束时间被记录。 +func TestStartFuncDefaults(t *testing.T) { + factory := StartFunc(nil) + end := factory("chat x", nil) + before := time.Now() + end() + if !before.Before(time.Now().Add(time.Second)) { + t.Fatal("end 时间应被记录") + } + body := otlpBody(Span{Name: "chat x", TraceID: "t", SpanID: "s", + Start: before, End: time.Now()}) + raw, _ := json.Marshal(body) + if !strings.Contains(string(raw), "agent-native") { + t.Fatalf("默认 service.name:%s", raw) + } +} diff --git a/runtime/telemetry/telemetry.go b/runtime/telemetry/telemetry.go index 42d6f92..3234ea1 100644 --- a/runtime/telemetry/telemetry.go +++ b/runtime/telemetry/telemetry.go @@ -8,6 +8,7 @@ import ( "encoding/json" "net/http" "os" + "strconv" "strings" "time" @@ -16,44 +17,112 @@ import ( // Span 是一次 invoke_agent、chat 或 execute_tool。 type Span struct { - Name string - TraceID string - SpanID string - Start time.Time - End time.Time - Attrs map[string]string + Name string + TraceID string + SpanID string + ParentSpanID string + Start time.Time + End time.Time + Attrs map[string]string } -// StartFunc 返回循环使用的追踪函数。设置了 OTEL_EXPORTER_OTLP_ENDPOINT 时, -// 结束 span 后以 OTLP/HTTP JSON 上报,不引入新的依赖。 -func StartFunc(log *logsys.Logger) func(name string, attrs map[string]string) func() { - endpoint := strings.TrimRight(os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT"), "/") +// 保留属性键:循环用它们把 run 级的 trace 关联传进工厂(不导出)。 +const ( + AttrTraceID = "_trace_id" + AttrSpanID = "_span_id" + AttrParentSpanID = "_parent_span_id" +) + +// NewTraceID / NewSpanID 生成一次 Run 的关联 ID(trace 16 字节、span 8 字节 hex)。 +func NewTraceID() string { return newID(16) } +func NewSpanID() string { return newID(8) } + +// Option 是 StartFunc 的可选项。 +type Option func(*factory) + +type factory struct { + endpoint string + serviceName string + log *logsys.Logger +} + +// WithServiceName 设置 OTLP resource 的 service.name(默认 agent-native)。 +func WithServiceName(name string) Option { + return func(f *factory) { + if strings.TrimSpace(name) != "" { + f.serviceName = name + } + } +} + +// WithEndpoint 设置 OTLP 端点(优先级高于环境变量;空 = 回退环境变量)。 +func WithEndpoint(endpoint string) Option { + return func(f *factory) { + if strings.TrimSpace(endpoint) != "" { + f.endpoint = strings.TrimRight(endpoint, "/") + } + } +} + +// StartFunc 返回循环使用的追踪函数。端点来自 WithEndpoint 或环境变量 +// OTEL_EXPORTER_OTLP_ENDPOINT(都为空则不导出);结束 span 后以 +// OTLP/HTTP JSON 上报,不引入新的依赖。默认不采集提示词正文。 +func StartFunc(log *logsys.Logger, opts ...Option) func(name string, attrs map[string]string) func() { + f := &factory{ + endpoint: strings.TrimRight(os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT"), "/"), + serviceName: "agent-native", + log: log, + } + for _, o := range opts { + o(f) + } return func(name string, attrs map[string]string) func() { sp := Span{ Name: name, TraceID: newID(16), SpanID: newID(8), - Start: time.Now(), Attrs: publicAttrs(attrs), + Start: time.Now(), + } + // 循环用保留键传入 run 级关联:一次 Run 共享 TraceID, + // chat/execute_tool 挂到 invoke_agent 根下。 + if attrs != nil { + if v := attrs[AttrTraceID]; v != "" { + sp.TraceID = v + } + if v := attrs[AttrSpanID]; v != "" { + sp.SpanID = v + } + if v := attrs[AttrParentSpanID]; v != "" { + sp.ParentSpanID = v + } } return func() { sp.End = time.Now() - if log != nil { + // 属性在结束时过滤:允许循环在 span 存续期间补充(如 usage)。 + sp.Attrs = publicAttrs(attrs) + if f.log != nil { kvs := []any{"span", sp.Name, "trace_id", sp.TraceID, "span_id", sp.SpanID} + if sp.ParentSpanID != "" { + kvs = append(kvs, "parent_span_id", sp.ParentSpanID) + } for k, v := range sp.Attrs { kvs = append(kvs, k, v) } - log.Info("genai span", kvs...) + f.log.Info("genai span", kvs...) } - if endpoint != "" { - _ = ExportOTLP(endpoint, sp) + if f.endpoint != "" { + if err := ExportOTLP(f.endpoint, sp, f.serviceName); err != nil && f.log != nil { + f.log.Info("genai span export failed", "error", err) + } } } } } -// publicAttrs 丢掉提示词正文。调用方即使传入也不会进入 span。 +// publicAttrs 丢掉提示词正文与保留键(_ 前缀的内部约定)。调用方即使 +// 传入正文也不会进入 span。 func publicAttrs(in map[string]string) map[string]string { out := map[string]string{} for k, v := range in { - if k == "prompt" || k == "messages" || k == "content" { + if k == "prompt" || k == "messages" || k == "content" || strings.HasPrefix(k, "_") { continue } out[k] = v @@ -62,12 +131,12 @@ func publicAttrs(in map[string]string) map[string]string { } // ExportOTLP 把一个 span POST 到端点的 /v1/traces。 -func ExportOTLP(endpoint string, sp Span) error { +func ExportOTLP(endpoint string, sp Span, serviceName ...string) error { endpoint = strings.TrimRight(endpoint, "/") if !strings.HasSuffix(endpoint, "/v1/traces") { endpoint += "/v1/traces" } - body, err := json.Marshal(otlpBody(sp)) + body, err := json.Marshal(otlpBody(sp, serviceName...)) if err != nil { return err } @@ -84,22 +153,42 @@ func ExportOTLP(endpoint string, sp Span) error { return nil } -func otlpBody(sp Span) map[string]any { +func otlpBody(sp Span, serviceName ...string) map[string]any { var attrs []map[string]any for k, v := range publicAttrs(sp.Attrs) { + // gen_ai.usage.* 是数值型语义(intValue),其余保持 stringValue。 + if strings.HasPrefix(k, "gen_ai.usage.") { + if n, err := strconv.Atoi(v); err == nil { + attrs = append(attrs, map[string]any{"key": k, "value": map[string]any{"intValue": n}}) + continue + } + } attrs = append(attrs, map[string]any{"key": k, "value": map[string]any{"stringValue": v}}) } + name := "agent-native" + if len(serviceName) > 0 && strings.TrimSpace(serviceName[0]) != "" { + name = serviceName[0] + } + span := map[string]any{ + "traceId": sp.TraceID, + "spanId": sp.SpanID, + "name": sp.Name, + "startTimeUnixNano": sp.Start.UnixNano(), + "endTimeUnixNano": sp.End.UnixNano(), + "attributes": attrs, + } + if sp.ParentSpanID != "" { + span["parentSpanId"] = sp.ParentSpanID + } return map[string]any{ "resourceSpans": []any{map[string]any{ + "resource": map[string]any{ + "attributes": []map[string]any{ + {"key": "service.name", "value": map[string]any{"stringValue": name}}, + }, + }, "scopeSpans": []any{map[string]any{ - "spans": []any{map[string]any{ - "traceId": sp.TraceID, - "spanId": sp.SpanID, - "name": sp.Name, - "startTimeUnixNano": sp.Start.UnixNano(), - "endTimeUnixNano": sp.End.UnixNano(), - "attributes": attrs, - }}, + "spans": []any{span}, }}, }}, }