Blizzard
2ee16d1f99
按"并存 + 等价回归"策略，对话主流程可跑在 Eino compose.Graph 上： - compose_graph.go：runConversation 按 EINO_COMPOSE 灰度开关分流； runComposeConversation 建图 START→ChatModel→END，Compile→Stream 回流 token； 模型未就绪/编译失败降级回 runAgent。默认关，graph.go 仍是默认且权威。 - compose_callbacks.go：composeTracer 用 utils/callbacks 把 ChatModel/Tool 的 start/end/error 桥到 ExecEvent（可观测归一，不再各处手写 emit）。 - LLM 接口 + Pool 增 ChatModel()；fakeLLM 加 cm 字段 + stub Eino 模型。 - 测试：compose 图编译运行 / compose 对话流式 / 开关关→走 runAgent。 顺带修真 bug：SubjectTaskStatus 原 sundynix.tasks.status 落在任务流通配 sundynix.tasks.> 内 → 状态事件被当成"幽灵任务"自我放大（实测污染 2300+ 条）。 挪到 sundynix.status.task + dispatcher 加空任务护栏。 验收：make test-go 全绿；live compose 路径 54 字答复 eval 1.00、默认路径 eval 1.00、幽灵任务 0 复发。剩余 branch/map/render 等节点逐步迁移后 graph.go 退役。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
69 lines
2.5 KiB
Go
69 lines
2.5 KiB
Go
package eino
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
|
||
"github.com/cloudwego/eino/callbacks"
|
||
"github.com/cloudwego/eino/components/model"
|
||
"github.com/cloudwego/eino/components/tool"
|
||
"github.com/cloudwego/eino/schema"
|
||
ucallbacks "github.com/cloudwego/eino/utils/callbacks"
|
||
)
|
||
|
||
// composeTracer 把 Eino compose 运行时的回调（ChatModel / Tool 的 start/end/error）
|
||
// 翻译成我们现有的 ExecEvent 轨迹——可观测"归一"：用框架原生回调，而非各处手写 emit。
|
||
// node 为该次运行在"运行·观测"里的归属节点 id（如 agent:xxx）。
|
||
func composeTracer(tr *execTracer, node string) callbacks.Handler {
|
||
return ucallbacks.NewHandlerHelper().
|
||
ChatModel(&ucallbacks.ModelCallbackHandler{
|
||
OnStart: func(ctx context.Context, _ *callbacks.RunInfo, in *model.CallbackInput) context.Context {
|
||
tr.emit(node, "model", "start", "compose·ChatModel", inputMsgsPreview(in), 0)
|
||
return ctx
|
||
},
|
||
OnEnd: func(ctx context.Context, _ *callbacks.RunInfo, out *model.CallbackOutput) context.Context {
|
||
detail := ""
|
||
if out != nil && out.Message != nil {
|
||
detail = truncate(out.Message.Content, 120)
|
||
}
|
||
tr.emit(node, "model", "end", "compose·ChatModel", detail, 0)
|
||
return ctx
|
||
},
|
||
OnEndWithStreamOutput: func(ctx context.Context, _ *callbacks.RunInfo, out *schema.StreamReader[*model.CallbackOutput]) context.Context {
|
||
out.Close() // 仅观测：正文经主输出流消费，这里只标记结束
|
||
tr.emit(node, "model", "end", "compose·ChatModel", "流式完成", 0)
|
||
return ctx
|
||
},
|
||
OnError: func(ctx context.Context, _ *callbacks.RunInfo, err error) context.Context {
|
||
tr.emit(node, "model", "error", "compose·ChatModel", err.Error(), 0)
|
||
return ctx
|
||
},
|
||
}).
|
||
Tool(&ucallbacks.ToolCallbackHandler{
|
||
OnStart: func(ctx context.Context, _ *callbacks.RunInfo, in *tool.CallbackInput) context.Context {
|
||
if in != nil {
|
||
tr.emit(node, "tool", "start", "compose·工具", truncate(in.ArgumentsInJSON, 120), 0)
|
||
}
|
||
return ctx
|
||
},
|
||
OnEnd: func(ctx context.Context, _ *callbacks.RunInfo, out *tool.CallbackOutput) context.Context {
|
||
if out != nil {
|
||
tr.emit(node, "tool", "end", "compose·工具", truncate(out.Response, 160), 0)
|
||
}
|
||
return ctx
|
||
},
|
||
OnError: func(ctx context.Context, _ *callbacks.RunInfo, err error) context.Context {
|
||
tr.emit(node, "tool", "error", "compose·工具", err.Error(), 0)
|
||
return ctx
|
||
},
|
||
}).
|
||
Handler()
|
||
}
|
||
|
||
func inputMsgsPreview(in *model.CallbackInput) string {
|
||
if in == nil {
|
||
return ""
|
||
}
|
||
return fmt.Sprintf("%d 条消息", len(in.Messages))
|
||
}
|