ambiguous-pointer commented on code in PR #1536: URL: https://github.com/apache/dubbo-admin/pull/1536#discussion_r3821028253
########## ai/go.mod: ########## Review Comment: Go 1.26 下 sonic v1.14.1 无法编译, 请先行合入远程更改 ########## ai/component/hooks/event.go: ########## @@ -0,0 +1,199 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package hooks + +import ( + "encoding/json" + "errors" + "fmt" + "reflect" + "strconv" + "sync" + "time" +) + +// Event identifies a stable boundary in an agent interaction. +type Event string + +const ( + EventInteractionStart Event = "interaction.start" + EventInteractionEnd Event = "interaction.end" + EventIterationStart Event = "iteration.start" + EventIterationEnd Event = "iteration.end" + EventStageStart Event = "stage.start" + EventStageEnd Event = "stage.end" + EventModelCallStart Event = "model_call.start" + EventModelCallEnd Event = "model_call.end" + EventToolCallStart Event = "tool_call.start" + EventToolCallEnd Event = "tool_call.end" +) Review Comment: - **潜在问题**： 1. **没有错误/降级/取消类事件**。工具失败、observe 超时/解析失败、显式取消，在 PR 里都只是 `State` 上的字段（`Degraded`/`FallbackUsed`/`Error`）。对 tracing/logging 两个消费方够用（span 上有 `error.type`、`agent.degraded` 属性）， 但对未来的 **metrics / 审计类 hook**（它们需要"按事件种类订阅"）就缺了第一类公民： - 例：生产上想统计"每月 observe 超时次数"或"工具失败率"，现在只能订阅 `stage.end` 然后过滤 `State.FallbackUsed`， 而不是订阅一个语义明确的 `agent.degraded` 事件；若未来事件字段演进，这类消费方会静默错算。 - **个人建议**: 事件种类扩展为 `agent/stage/llm/tool × start/end/error` + `agent.degraded` + `agent.cancel` + `llm.chunk`（预留），并给 `State` 增加 `Seq uint64`。事件字段只读约定保持。因为模型部署侧可能不一定都是稳定的模型，例如 VLLM 私有化部署的时候，工具调用参数模板没有绑定正确的时候，调用工具会出现偶发性的直接中断。所以会需要预设详细一些 - **生产场景**：Dubbo 服务诊断场景（agent 通过 MCP 调 `get_service_detail`/诊断工具 #1499）——SRE 想要"按工具维度"的失败率报表，若没有独立 `tool.error` 事件种类，报表逻辑要散落在每个消费方里重复过滤，接入点越多越容易漏。 ########## ai/component/hooks/event.go: ########## @@ -0,0 +1,199 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package hooks + +import ( + "encoding/json" + "errors" + "fmt" + "reflect" + "strconv" + "sync" + "time" +) + +// Event identifies a stable boundary in an agent interaction. +type Event string + +const ( + EventInteractionStart Event = "interaction.start" + EventInteractionEnd Event = "interaction.end" + EventIterationStart Event = "iteration.start" + EventIterationEnd Event = "iteration.end" + EventStageStart Event = "stage.start" + EventStageEnd Event = "stage.end" + EventModelCallStart Event = "model_call.start" + EventModelCallEnd Event = "model_call.end" + EventToolCallStart Event = "tool_call.start" + EventToolCallEnd Event = "tool_call.end" +) + +const ( + FallbackReasonTimeout = "timeout" + FallbackReasonParseError = "parse_error" +) + +var allEvents = []Event{ + EventInteractionStart, + EventInteractionEnd, + EventIterationStart, + EventIterationEnd, + EventStageStart, + EventStageEnd, + EventModelCallStart, + EventModelCallEnd, + EventToolCallStart, + EventToolCallEnd, +} + +// AllEvents returns a copy of every supported event in lifecycle order. +func AllEvents() []Event { + events := make([]Event, len(allEvents)) + copy(events, allEvents) + return events +} + +func (e Event) valid() bool { + for _, candidate := range allEvents { + if e == candidate { + return true + } + } + return false +} + +func (e Event) toolCall() bool { + return e == EventToolCallStart || e == EventToolCallEnd +} + +// State is a read-only snapshot supplied to hooks. Go context propagation is +// carried separately through context.Context and must not be stored here. +type State struct { + Event Event + InteractionID string + SessionID string + Iteration int + Stage string + Model string + ToolName string + ToolCallID string + Input string + Output string + Error string + ErrorType string + Degraded bool + FallbackUsed bool + FallbackReason string + InputTokens int + OutputTokens int + TotalTokens int + StartedAt time.Time + EndedAt time.Time + inputContent *lazyContentSnapshot + outputContent *lazyContentSnapshot +} + +type lazyContentSnapshot struct { + once sync.Once + provider func() any + content string +} + +func newLazyContentSnapshot(provider func() any) *lazyContentSnapshot { + if provider == nil { + return nil + } + return &lazyContentSnapshot{provider: provider} +} + +func (s *lazyContentSnapshot) snapshot() string { + if s == nil { + return "" + } + s.once.Do(func() { + s.content = SnapshotContent(s.provider()) + s.provider = nil + }) + return s.content +} + +// WithInputContent attaches an immutable input snapshot that is materialized +// only if a matching content-capturing hook requests it. +func (s State) WithInputContent(provider func() any) State { + s.inputContent = newLazyContentSnapshot(provider) + return s +} + +// WithOutputContent attaches an immutable output snapshot that is materialized +// only if a matching content-capturing hook requests it. +func (s State) WithOutputContent(provider func() any) State { + s.outputContent = newLazyContentSnapshot(provider) + return s +} + +func (s State) snapshotInputContent() string { + if s.Input != "" { + return s.Input + } + return s.inputContent.snapshot() +} + +func (s State) snapshotOutputContent() string { + if s.Output != "" { + return s.Output + } + return s.outputContent.snapshot() +} Review Comment: - **潜在问题**：`lazyContent` 字段（`manager.go:44,96`）**只有包内 `NewTracingRegistration` 能用**（未导出）， 外部捕获内容的 hook 一律走 `manager.go:193-195` 的**急切快照**分支——也就是说"懒"只对内置 tracing hook 成立，对将来第三方内容型 hook（如审计）不成立。PR文档里"Content is serialized lazily"的表述容易误导。 - **个人建议**：把 `lazyContent` 语义并入公开的 `Registration`（如 `CaptureContent: CaptureLazy`）或至少在 `Registration` 上注释清楚两档行为。 - **生产场景**：审计 hook 需要"模型输入/输出原文"留档——如果它被急切序列化，每次模型调用都会多一次完整 JSON marshal（大对话可能几百 KB），生产热点路径上不可忽略；同时内容进内存=更大的 PII 暴露面。应能声明"延迟到真正落盘前才序列化"。 ########## ai/component/hooks/hooks.yaml: ########## Review Comment: 这里可能对于开发者或者需要基于钩子实现私有化能力的时候,会存在 **我不知道有哪些 hook、不知道接入点** ```yaml type: hooks spec: hooks: - name: "logging" enabled: true events: ["agent.start", "agent.end", "agent.error", "agent.degraded", "agent.cancel", "stage.start", "stage.end", "stage.error", "llm.start", "llm.end", "llm.error", "tool.start", "tool.end", "tool.error"] config: { level: "info" } ``` ########## ai/component/server/engine/handlers.go: ########## @@ -74,7 +77,18 @@ func (h *AgentHandler) StreamChat(c *gin.Context) { } }() Review Comment: // ← 这里没有 go discardAgentOutput(channels)！ 对比另外两条退出路径（handlers.go:104-107 和 134-136） 1. handler 在流中途 panic（比如 `MessageDelta` 遇到未知 final 类型、或任何将来加的代码）； 2. defer recover 触发 → 发一条 `internal_error` SSE → handler 返回 → gin 关连接； 3. **交互 goroutine 不会死**——`interactionCtx := context.WithoutCancel(extractedCtx)`（`handlers.go:87`）已经把请求取消剥掉了，客户端断开对它是透明的； 4. 交互 goroutine 继续生成，调用 `chans.Send`（`agent.go:62-70`）： ```go func (chans *Channels) Send(sf *schema.StreamFeedback) { sf.SetIndex(chans.nextIndex) chans.nextIndex++ chans.UserRespChan <- sf // ← 有界阻塞发送，缓冲满就永久卡住 } ``` 5. 缓冲（bufferSize）塞满 ~16 条后，**没有任何人排空** → 交互 goroutine 永久阻塞在 `Send`； 6. 连锁反应：阻塞在 `Send` 意味着 goroutine 的 defer 永远不执行——`interaction.end` 事件**发不出去**（trace 缺尾）、`finishInteraction` 不执行（`ra.active` 表里的条目永不删除）； 7. 进程关闭时 `Stop()`（`react.go:207-222`）对这条交互执行 `activeWG.Wait()` → **Stop 也跟着挂起**，直到外层 20s 超时兜底，关闭质量劣化。 **一句话**：handler 侧一个 panic，产生一个永久阻塞的 goroutine + 一条不完整的 trace + Stop 挂起——而这一切本来用一行 `go discardAgentOutput(channels)` 就能避免。 根子是 `Channels.Send` 的**阻塞语义**——`discardAgentOutput` 只是防呆补丁，而且只在 handler 这一侧有 ########## ai/schema/json/hooks.schema.json: ########## Review Comment: `events` 字段以 `enum` 形式列出全部事件种类 https://github.com/apache/dubbo-admin/blob/5d6df6e9b5d4b8c6120a0919a7bbed83d08b7947/ai/schema/json/memory.schema.json#L15-L20 ########## ai/component/server/engine/handlers.go: ########## @@ -74,7 +77,18 @@ func (h *AgentHandler) StreamChat(c *gin.Context) { } }() Review Comment: 个人理解 这个和整个 agent 调用的 : 交互持久化 + 事件日志 + 整段重放 有关 可能得 #1534 完成后全面的思考一下这个地方如何实现 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
