diff --git a/backend/ai-core/cmd/main.go b/backend/ai-core/cmd/main.go index d61f167..812c2df 100644 --- a/backend/ai-core/cmd/main.go +++ b/backend/ai-core/cmd/main.go @@ -426,6 +426,9 @@ func main() { // 健康检查与对话API的HTTP mux mux := http.NewServeMux() + // 注册全链路追踪事件端点 + registerTraceEndpoint(mux) + // 初始化子会话管理器 subManager := subsession.NewManager(chatAdapter) diff --git a/backend/ai-core/cmd/trace.go b/backend/ai-core/cmd/trace.go new file mode 100644 index 0000000..2f545a3 --- /dev/null +++ b/backend/ai-core/cmd/trace.go @@ -0,0 +1,87 @@ +package main + +import ( + "encoding/json" + "fmt" + "net/http" + "strconv" + "sync" + "time" +) + +// TraceEvent represents a single step in the message processing pipeline. +type TraceEvent struct { + ID string `json:"id"` + Timestamp time.Time `json:"timestamp"` + SessionID string `json:"session_id,omitempty"` + UserID string `json:"user_id,omitempty"` + Hop string `json:"hop"` // message_received, intent, subsession, llm_call, tool_call, vision, synthesis, review, response, think + Label string `json:"label"` + Status string `json:"status"` // success, error, running + Detail string `json:"detail,omitempty"` + DurationMs int64 `json:"duration_ms,omitempty"` + Data map[string]interface{} `json:"data,omitempty"` +} + +var ( + traceMu sync.Mutex + traceEvents []TraceEvent + traceMax = 200 +) + +// AddTraceEvent appends a trace event to the in-memory ring buffer. +func AddTraceEvent(hop, sessionID, userID, label, status, detail string, durationMs int64, data map[string]interface{}) { + traceMu.Lock() + defer traceMu.Unlock() + ev := TraceEvent{ + ID: fmt.Sprintf("%s-%d", hop, time.Now().UnixNano()), + Timestamp: time.Now(), + SessionID: sessionID, + UserID: userID, + Hop: hop, + Label: label, + Status: status, + Detail: detail, + DurationMs: durationMs, + Data: data, + } + traceEvents = append(traceEvents, ev) + if len(traceEvents) > traceMax { + traceEvents = traceEvents[len(traceEvents)-traceMax:] + } +} + +// GetTraceEvents returns recent trace events, optionally filtered by session. +func GetTraceEvents(sessionID string, limit int) []TraceEvent { + traceMu.Lock() + defer traceMu.Unlock() + if limit <= 0 || limit > len(traceEvents) { + limit = len(traceEvents) + } + var result []TraceEvent + // Return newest first. + for i := len(traceEvents) - 1; i >= 0 && len(result) < limit; i-- { + ev := traceEvents[i] + if sessionID == "" || ev.SessionID == sessionID { + result = append(result, ev) + } + } + return result +} + +// registerTraceEndpoint adds the trace events API. +func registerTraceEndpoint(mux *http.ServeMux) { + mux.HandleFunc("/api/v1/trace/events", func(w http.ResponseWriter, r *http.Request) { + sessionID := r.URL.Query().Get("session_id") + limit := 100 + if l, err := strconv.Atoi(r.URL.Query().Get("limit")); err == nil && l > 0 && l <= 500 { + limit = l + } + events := GetTraceEvents(sessionID, limit) + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]interface{}{ + "events": events, + "total": len(events), + }) + }) +} diff --git a/ethend/public/index.html b/ethend/public/index.html index bf6694e..42cb285 100644 --- a/ethend/public/index.html +++ b/ethend/public/index.html @@ -6284,15 +6284,51 @@ async function refreshTrace() { } } - // 如果是首次加载（content 只有 loading 提示），全量渲染 + // Try pipeline view first (new format) + var pipelines = data.pipelines || []; + if (pipelines.length > 0) { + var html = '
'; + for (var pi = 0; pi < pipelines.length; pi++) { + var pl = pipelines[pi]; + var steps = pl.steps || []; + var hasMsg = steps.some(function(s){ return s.hop === 'msg_received'; }); + html += '
'; + // Pipeline header + html += '
'; + html += '📨 ' + escHtml(pl.sender || '?') + (pl.platform ? ' · ' + escHtml(pl.platform) : '') + ''; + html += '' + (pl.timestamp || '').substring(11,19) + ''; + html += '
'; + // Pipeline steps + html += '
'; + for (var si = 0; si < steps.length; si++) { + var s = steps[si]; + var hopIcon = {msg_received:'📥',msg_sent:'📤',llm_call:'🧠',tool_call:'🔧',session:'👤'}[s.hop] || '●'; + var hopColor = s.status === 'error' ? 'var(--red)' : (s.hop === 'msg_received' ? 'var(--blue)' : s.hop === 'msg_sent' ? 'var(--green)' : 'var(--text2)'); + html += '
'; + html += '' + hopIcon + ''; + html += '' + escHtml(s.label) + ''; + if (s.detail) html += '
' + escHtml(s.detail) + ''; + if (s.durationMs > 0) html += ' (' + (s.durationMs >= 1000 ? (s.durationMs/1000).toFixed(1)+'s' : s.durationMs+'ms') + ')'; + html += '
'; + } + html += '
'; + } + html += '
'; + contentEl.innerHTML = html; + // Also update stats + updateTraceStatsEl(statsEl, traces, stats, data.session); + return; + } + + // Fallback: flat timeline var timeline = contentEl.querySelector('.trace-timeline'); if (!timeline) { - if (traces.length === 0) { + if (traces.length === 0 && pipelines.length === 0) { contentEl.innerHTML = '
📭
暂无链路追踪数据
请确保服务正在运行且有消息活动
'; statsEl.innerHTML = ''; return; } - // 全量渲染 + // 全量渲染 (old format) var html = '
'; for (var i = 0; i < traces.length; i++) { html += traceHopHtml(traces[i], i > 0 ? traces[i-1] : null); diff --git a/ethend/src/index.js b/ethend/src/index.js index da46223..ce16d5c 100644 --- a/ethend/src/index.js +++ b/ethend/src/index.js @@ -1277,16 +1277,16 @@ function parseLogTimestamp(line) { app.get('/api/trace/recent', async (req, res) => { const limit = Math.min(parseInt(req.query.limit) || 50, 200); try { - const [llmResult, toolResult, sessionsResult, bridgeLogsResult, aiCoreLogsResult] = await Promise.all([ + const [llmResult, toolResult, sessionsResult, bridgeLogsResult, traceEventsResult] = await Promise.all([ proxyToAICore(`/api/v1/llm-calls?limit=${limit}`).catch(() => ({ status: 502, body: [] })), proxyToAICore(`/api/v1/tools/calls?limit=${limit}`).catch(() => ({ status: 502, body: { calls: [] } })), proxyToGateway('/api/v1/admin/sessions/active').catch(() => ({ status: 502, body: { users: {} } })), // 拉 platform-bridge 最近的消息日志 - Promise.all((['obv11', 'qq']).map(platform => - proxyToPlatformBridge(`/api/v1/logs/${platform}?limit=50`).catch(() => ({ status: 502, body: { entries: [] } })) + Promise.all((['obv11']).map(platform => + proxyToPlatformBridge(`/api/v1/logs/${platform}?limit=50`).catch(() => ({ status: 502, body: { logs: [] } })) )), - // 拉 ai-core 最近的日志（解析处理事件） - proxyToAICore('/api/v1/llm-calls/stream?limit=0').catch(() => ({ status: 502, body: [] })), + // 拉 ai-core 追踪事件 + proxyToAICore(`/api/v1/trace/events?limit=${limit}`).catch(() => ({ status: 502, body: { events: [] } })), ]); const traces = []; @@ -1349,6 +1349,23 @@ app.get('/api/trace/recent', async (req, res) => { }); } + // === ai-core 追踪事件 (intent, subsession, synthesis, etc.) === + const traceEvents = traceEventsResult.body?.events || []; + for (const ev of traceEvents) { + const ts = ev.timestamp ? new Date(ev.timestamp).getTime() : Date.now(); + traces.push({ + id: ev.id || `trace-${ts}`, + timestamp: new Date(ts).toISOString(), ts, + service: 'ai-core', + hop: ev.hop || 'trace', + label: ev.label || '', + status: ev.status || 'success', + durationMs: ev.duration_ms || 0, + detail: ev.detail || '', + data: ev.data || {}, + }); + } + // === 活跃会话 === const users = sessionsResult.body?.users || {}; for (const [userID, sessions] of Object.entries(users)) {