feat: 全链路追踪系统 — ai-core追踪事件 + ethend管道视图
ai-core: - 新增 TraceEvent 内存环形缓冲区(200条) - GET /api/v1/trace/events 端点(支持session_id过滤) - 预置 trace.go 追踪事件定义 ethend: - trace/recent 聚合 platform-bridge消息日志 + ai-core追踪事件 - 消息按管线分组: msg_received→llm_call→tool_call→msg_sent - 前端管线卡片视图: 每条消息显示完整处理步骤 Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -426,6 +426,9 @@ func main() {
|
|||||||
// 健康检查与对话API的HTTP mux
|
// 健康检查与对话API的HTTP mux
|
||||||
mux := http.NewServeMux()
|
mux := http.NewServeMux()
|
||||||
|
|
||||||
|
// 注册全链路追踪事件端点
|
||||||
|
registerTraceEndpoint(mux)
|
||||||
|
|
||||||
// 初始化子会话管理器
|
// 初始化子会话管理器
|
||||||
subManager := subsession.NewManager(chatAdapter)
|
subManager := subsession.NewManager(chatAdapter)
|
||||||
|
|
||||||
|
|||||||
@@ -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),
|
||||||
|
})
|
||||||
|
})
|
||||||
|
}
|
||||||
@@ -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 = '<div class="trace-timeline">';
|
||||||
|
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 += '<div class="trace-pipeline" style="margin:12px 0;border:1px solid var(--border);border-radius:8px;overflow:hidden">';
|
||||||
|
// Pipeline header
|
||||||
|
html += '<div style="padding:8px 12px;background:var(--bg2);display:flex;justify-content:space-between;align-items:center">';
|
||||||
|
html += '<span style="font-weight:600;font-size:13px">📨 ' + escHtml(pl.sender || '?') + (pl.platform ? ' · ' + escHtml(pl.platform) : '') + '</span>';
|
||||||
|
html += '<span style="font-size:11px;color:var(--text2)">' + (pl.timestamp || '').substring(11,19) + '</span>';
|
||||||
|
html += '</div>';
|
||||||
|
// Pipeline steps
|
||||||
|
html += '<div style="padding:4px 0">';
|
||||||
|
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 += '<div style="display:flex;align-items:flex-start;padding:4px 12px;gap:8px;font-size:12px">';
|
||||||
|
html += '<span style="color:' + hopColor + ';flex-shrink:0;width:20px;text-align:center">' + hopIcon + '</span>';
|
||||||
|
html += '<span style="flex:1"><b>' + escHtml(s.label) + '</b>';
|
||||||
|
if (s.detail) html += '<br><span style="color:var(--text2);font-size:11px">' + escHtml(s.detail) + '</span>';
|
||||||
|
if (s.durationMs > 0) html += ' <span style="color:var(--text3);font-size:10px">(' + (s.durationMs >= 1000 ? (s.durationMs/1000).toFixed(1)+'s' : s.durationMs+'ms') + ')</span>';
|
||||||
|
html += '</span></div>';
|
||||||
|
}
|
||||||
|
html += '</div></div>';
|
||||||
|
}
|
||||||
|
html += '</div>';
|
||||||
|
contentEl.innerHTML = html;
|
||||||
|
// Also update stats
|
||||||
|
updateTraceStatsEl(statsEl, traces, stats, data.session);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Fallback: flat timeline
|
||||||
var timeline = contentEl.querySelector('.trace-timeline');
|
var timeline = contentEl.querySelector('.trace-timeline');
|
||||||
if (!timeline) {
|
if (!timeline) {
|
||||||
if (traces.length === 0) {
|
if (traces.length === 0 && pipelines.length === 0) {
|
||||||
contentEl.innerHTML = '<div class="trace-empty"><div class="icon">📭</div>暂无链路追踪数据<br><span style="font-size:11px;color:var(--text3)">请确保服务正在运行且有消息活动</span></div>';
|
contentEl.innerHTML = '<div class="trace-empty"><div class="icon">📭</div>暂无链路追踪数据<br><span style="font-size:11px;color:var(--text3)">请确保服务正在运行且有消息活动</span></div>';
|
||||||
statsEl.innerHTML = '';
|
statsEl.innerHTML = '';
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
// 全量渲染
|
// 全量渲染 (old format)
|
||||||
var html = '<div class="trace-timeline">';
|
var html = '<div class="trace-timeline">';
|
||||||
for (var i = 0; i < traces.length; i++) {
|
for (var i = 0; i < traces.length; i++) {
|
||||||
html += traceHopHtml(traces[i], i > 0 ? traces[i-1] : null);
|
html += traceHopHtml(traces[i], i > 0 ? traces[i-1] : null);
|
||||||
|
|||||||
+22
-5
@@ -1277,16 +1277,16 @@ function parseLogTimestamp(line) {
|
|||||||
app.get('/api/trace/recent', async (req, res) => {
|
app.get('/api/trace/recent', async (req, res) => {
|
||||||
const limit = Math.min(parseInt(req.query.limit) || 50, 200);
|
const limit = Math.min(parseInt(req.query.limit) || 50, 200);
|
||||||
try {
|
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/llm-calls?limit=${limit}`).catch(() => ({ status: 502, body: [] })),
|
||||||
proxyToAICore(`/api/v1/tools/calls?limit=${limit}`).catch(() => ({ status: 502, body: { calls: [] } })),
|
proxyToAICore(`/api/v1/tools/calls?limit=${limit}`).catch(() => ({ status: 502, body: { calls: [] } })),
|
||||||
proxyToGateway('/api/v1/admin/sessions/active').catch(() => ({ status: 502, body: { users: {} } })),
|
proxyToGateway('/api/v1/admin/sessions/active').catch(() => ({ status: 502, body: { users: {} } })),
|
||||||
// 拉 platform-bridge 最近的消息日志
|
// 拉 platform-bridge 最近的消息日志
|
||||||
Promise.all((['obv11', 'qq']).map(platform =>
|
Promise.all((['obv11']).map(platform =>
|
||||||
proxyToPlatformBridge(`/api/v1/logs/${platform}?limit=50`).catch(() => ({ status: 502, body: { entries: [] } }))
|
proxyToPlatformBridge(`/api/v1/logs/${platform}?limit=50`).catch(() => ({ status: 502, body: { logs: [] } }))
|
||||||
)),
|
)),
|
||||||
// 拉 ai-core 最近的日志(解析处理事件)
|
// 拉 ai-core 追踪事件
|
||||||
proxyToAICore('/api/v1/llm-calls/stream?limit=0').catch(() => ({ status: 502, body: [] })),
|
proxyToAICore(`/api/v1/trace/events?limit=${limit}`).catch(() => ({ status: 502, body: { events: [] } })),
|
||||||
]);
|
]);
|
||||||
|
|
||||||
const traces = [];
|
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 || {};
|
const users = sessionsResult.body?.users || {};
|
||||||
for (const [userID, sessions] of Object.entries(users)) {
|
for (const [userID, sessions] of Object.entries(users)) {
|
||||||
|
|||||||
Reference in New Issue
Block a user