feat(observability): 执行可视化 — 节点级实时轨迹(运行·观测)
把任务执行做成可观测:Dispatcher 在每个节点/阶段发结构化 ExecEvent, 经独立 NATS 通道回流,前端逐节点点亮(状态/耗时/工具入参产出)。 - shared: contract.ExecEvent + ExecSubject(sundynix.exec.<id>,与 Token 流分流); bus.PublishExec/CompleteExec/SubscribeExec(core NATS,复用结束头) - dispatcher: execTracer(自增 Seq 保序 + span 自动计耗时); Orchestrator 加 ExecSink;通用图(init 召回 / 各 tool 入参→产出 / prompt / model 首token+token数)与报告编排(规划大纲 / 各章并行 start-end / 渲染)全程埋点 - gateway: SubscribeExec + GET /tasks/:id/exec SSE(与 token 流并行) - desktop: streamExec + deriveNodes(按 node 归并 start/end/error/info); 复用组件 ExecTrace(竖向轨道,按 kind 着色,运行中脉冲灯); 新 RunsView(运行·观测:轨迹+输出双栏);BottomDrawer 轨迹/工具调用 tab 接真实数据; ReportView 加执行轨迹栏;左导航「运行」置就绪 实测: - 报告任务 /exec:规划(2680ms,4章) → 4 章并行(seq 交错,各~7-8s 重叠=真并行, 每章带 docs 知识库检索预览+成稿字数) → 渲染(docx 落盘) - 通用图 /exec:tool:kb_search(678ms,入参→Milvus 产出) → prompt(2消息) → model(首token 860ms / 4 tokens) - 浏览器(Preview):报告页执行轨迹逐节点点亮、章节带耗时/字数/检索片段,完成后下载 Word Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -2,6 +2,7 @@ package eino
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
@@ -30,7 +31,7 @@ type RunCtx struct {
|
||||
//
|
||||
// 工具/检索节点按拓扑序真实调用 MCP(sundynix.tools.go.*),结果注入模型上下文。
|
||||
// 分支/并行节点暂未编译(TODO:compose.Branch / fan-out)。
|
||||
func (o *Orchestrator) compileFlow(ctx context.Context, t *contract.Task) (compose.Runnable[*contract.Task, *schema.Message], error) {
|
||||
func (o *Orchestrator) compileFlow(ctx context.Context, t *contract.Task, tr *execTracer) (compose.Runnable[*contract.Task, *schema.Message], error) {
|
||||
plan := dsl.Compile(t.Graph) // 系统提示词 / 用户输入 / 默认兜底
|
||||
flow, _ := dsl.Parse(t.Graph)
|
||||
|
||||
@@ -41,13 +42,17 @@ func (o *Orchestrator) compileFlow(ctx context.Context, t *contract.Task) (compo
|
||||
func(ctx context.Context, task *contract.Task) (*RunCtx, error) {
|
||||
uid, _ := task.Meta[contract.MetaUserID].(string)
|
||||
sid, _ := task.Meta[contract.MetaSessionID].(string)
|
||||
end := tr.span("init", "memory", "召回画像与历史")
|
||||
profile := o.fetchMemory(ctx, uid, plan.Query)
|
||||
history := o.fetchHistory(ctx, sid)
|
||||
end(fmt.Sprintf("画像 %d 字 · 历史 %d 条", len([]rune(profile)), len(history)), nil)
|
||||
return &RunCtx{
|
||||
UserID: uid,
|
||||
SessionID: sid,
|
||||
System: plan.System,
|
||||
Query: plan.Query,
|
||||
Profile: o.fetchMemory(ctx, uid, plan.Query),
|
||||
History: o.fetchHistory(ctx, sid),
|
||||
Profile: profile,
|
||||
History: history,
|
||||
}, nil
|
||||
})); err != nil {
|
||||
return nil, err
|
||||
@@ -64,7 +69,7 @@ func (o *Orchestrator) compileFlow(ctx context.Context, t *contract.Task) (compo
|
||||
}
|
||||
key := fmt.Sprintf("tool_%d", idx)
|
||||
idx++
|
||||
if err := g.AddLambdaNode(key, compose.InvokableLambda(o.makeToolNode(t.ID, tool, args))); err != nil {
|
||||
if err := g.AddLambdaNode(key, compose.InvokableLambda(o.makeToolNode(t.ID, tool, args, tr))); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := g.AddEdge(prev, key); err != nil {
|
||||
@@ -75,7 +80,12 @@ func (o *Orchestrator) compileFlow(ctx context.Context, t *contract.Task) (compo
|
||||
}
|
||||
|
||||
// prompt:黑板 → []*schema.Message(系统提示词 + 画像 + 工具产出 + 历史 + 用户输入)。
|
||||
if err := g.AddLambdaNode("prompt", compose.InvokableLambda(buildMessages)); err != nil {
|
||||
if err := g.AddLambdaNode("prompt", compose.InvokableLambda(
|
||||
func(ctx context.Context, rc *RunCtx) ([]*schema.Message, error) {
|
||||
msgs, err := buildMessages(ctx, rc)
|
||||
tr.info("prompt", "prompt", "组装提示词", fmt.Sprintf("%d 条消息 · 工具产出 %d 段", len(msgs), len(rc.ToolOut)))
|
||||
return msgs, err
|
||||
})); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := g.AddEdge(prev, "prompt"); err != nil {
|
||||
@@ -101,9 +111,11 @@ func (o *Orchestrator) compileFlow(ctx context.Context, t *contract.Task) (compo
|
||||
}
|
||||
|
||||
// makeToolNode 返回一个真实调用 MCP 工具的图节点:把结果增补进黑板,失败降级不阻断。
|
||||
func (o *Orchestrator) makeToolNode(taskID, tool string, args map[string]any) func(context.Context, *RunCtx) (*RunCtx, error) {
|
||||
func (o *Orchestrator) makeToolNode(taskID, tool string, args map[string]any, tr *execTracer) func(context.Context, *RunCtx) (*RunCtx, error) {
|
||||
node := "tool:" + tool
|
||||
return func(ctx context.Context, rc *RunCtx) (*RunCtx, error) {
|
||||
if o.tools == nil {
|
||||
tr.info(node, "tool", "工具 "+tool, "工具总线未接入,跳过")
|
||||
return rc, nil
|
||||
}
|
||||
// 未显式带查询词则注入当前用户输入,便于检索类工具。
|
||||
@@ -114,19 +126,34 @@ func (o *Orchestrator) makeToolNode(taskID, tool string, args map[string]any) fu
|
||||
if call["q"] == nil && call["query"] == nil {
|
||||
call["q"] = rc.Query
|
||||
}
|
||||
end := tr.span(node, "tool", "调用工具 "+tool)
|
||||
cctx, cancel := context.WithTimeout(ctx, toolCallTimeout)
|
||||
defer cancel()
|
||||
res, err := o.tools.CallTool(cctx, contract.ToolSubjectGo(tool), &contract.ToolCall{
|
||||
Tool: tool, TaskID: taskID, Args: call,
|
||||
})
|
||||
if err != nil || res == nil || !res.OK || res.Content == "" {
|
||||
if err != nil {
|
||||
end("调用失败,降级跳过", err)
|
||||
return rc, nil
|
||||
}
|
||||
if res == nil || !res.OK || res.Content == "" {
|
||||
end("无结果,降级跳过", nil)
|
||||
return rc, nil // 工具不可用/无结果 → 降级跳过
|
||||
}
|
||||
end("入参 "+previewArgs(call)+" → 产出 "+truncate(res.Content, 160), nil)
|
||||
rc.ToolOut = append(rc.ToolOut, "["+tool+"] "+res.Content)
|
||||
return rc, nil
|
||||
}
|
||||
}
|
||||
|
||||
// previewArgs 把工具入参压成一行短预览。
|
||||
func previewArgs(args map[string]any) string {
|
||||
if data, err := json.Marshal(args); err == nil {
|
||||
return truncate(string(data), 120)
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// buildMessages 把黑板组装为发给模型的消息序列。
|
||||
func buildMessages(_ context.Context, rc *RunCtx) ([]*schema.Message, error) {
|
||||
var sys strings.Builder
|
||||
|
||||
Reference in New Issue
Block a user