366 lines
13 KiB
Go
366 lines
13 KiB
Go
package service
|
||
|
||
// ========================================================================
|
||
// RunLog —— Agent 运行轨迹记录器（内存环形缓冲）
|
||
// ========================================================================
|
||
// 记录最近 N 次 Enhance 运行的完整轨迹（每一步干了什么、时间节点、token），
|
||
// 供 /api/v1/agent/runs 系列接口与 /agent/view 可视化面板查询。
|
||
//
|
||
// 为什么用内存环形缓冲而不是落库：
|
||
// 1. 历史明细 PHP 端已经写入 xk_ai_generation_step（权威审计数据）；
|
||
// Go 端的定位是「实时观测最近运行」，重启清空可接受
|
||
// 2. 内存缓冲能记录到"PHP 侧看不到"的运行：守卫拦截、resolveClient 失败
|
||
// 这类在返回 PHP 之前就终止的请求
|
||
// 3. 零依赖：DB 抖动时面板依然可用（观测工具不能依赖被观测对象）
|
||
//
|
||
// 并发安全：所有读写都经过 sync.RWMutex，Add 是 O(1) 覆盖写
|
||
// ========================================================================
|
||
|
||
import (
|
||
"encoding/json"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
)
|
||
|
||
// AgentRunStatus 运行状态枚举（与面板徽标颜色对应）
|
||
const (
|
||
RunStatusOK = 1 // 成功
|
||
RunStatusFail = 2 // 失败（LLM 调用失败 / 配置解析失败等）
|
||
RunStatusBlocked = 3 // 医疗守卫拦截
|
||
)
|
||
|
||
// AgentRunRecord 一次 Agent 运行的完整轨迹
|
||
//
|
||
// Steps 直接复用 EnhanceStep（与返回 PHP 的结构一致），
|
||
// 每个 step 自带 started_at / finished_at / duration_ms，就是面板时间线的数据源
|
||
type AgentRunRecord struct {
|
||
ID int64 `json:"id"` // 递增 ID（进程内唯一，重启重置）
|
||
Scene string `json:"scene"` // 场景：medical_record / prescription
|
||
Provider string `json:"provider"` // 实际使用的供应商
|
||
Model string `json:"model"` // 实际使用的模型名
|
||
CfgSource string `json:"cfg_source"` // 配置来源：active / yaml_default 等
|
||
Status int `json:"status"` // 1成功 2失败 3守卫拦截
|
||
Error string `json:"error,omitempty"` // 失败原因（成功时为空）
|
||
TotalMs int `json:"total_ms"` // 总耗时毫秒
|
||
StartedAt int64 `json:"started_at"` // 开始时间（unix 秒）
|
||
FinishedAt int64 `json:"finished_at"` // 结束时间（unix 秒）
|
||
PromptTokens int `json:"prompt_tokens"` // 输入 token 合计
|
||
CompletionTokens int `json:"completion_tokens"` // 输出 token 合计
|
||
TotalTokens int `json:"total_tokens"` // 总 token 合计
|
||
Steps []EnhanceStep `json:"steps"` // 完整步骤明细（时间线数据源）
|
||
// RequestSnapshot 请求入参快照（截断后的 JSON 字符串）
|
||
// 用途：管理前端的「失败一键重放」——把入参回填进 Enhance 调试台重跑。
|
||
// 只在详情接口返回（列表摘要不含），内容截断控制内存（200 条 × ~2KB ≈ 400KB 上限）
|
||
RequestSnapshot string `json:"request_snapshot,omitempty"`
|
||
}
|
||
|
||
// AgentRunSummary 列表视图用的轻量摘要（不含 step 的 detail 全文，节省传输）
|
||
type AgentRunSummary struct {
|
||
ID int64 `json:"id"`
|
||
Scene string `json:"scene"`
|
||
Provider string `json:"provider"`
|
||
Model string `json:"model"`
|
||
CfgSource string `json:"cfg_source"`
|
||
Status int `json:"status"`
|
||
Error string `json:"error,omitempty"`
|
||
TotalMs int `json:"total_ms"`
|
||
StartedAt int64 `json:"started_at"`
|
||
TotalTokens int `json:"total_tokens"`
|
||
StepBriefs []RunStepBrief `json:"step_briefs"` // 每步的类型/状态/耗时（列表里画小圆点用）
|
||
}
|
||
|
||
// RunStepBrief 步骤摘要（列表视图用）
|
||
type RunStepBrief struct {
|
||
StepType string `json:"step_type"`
|
||
Status int `json:"status"`
|
||
DurationMs int `json:"duration_ms"`
|
||
}
|
||
|
||
// runLogBuffer 环形缓冲本体
|
||
type runLogBuffer struct {
|
||
mu sync.RWMutex
|
||
items []AgentRunRecord // 定长环形数组
|
||
size int // 容量
|
||
count int // 已写入条数（<= size）
|
||
head int // 下一个写入位置
|
||
nextID int64 // 自增 ID
|
||
}
|
||
|
||
// 全局单例：保留最近 200 次运行
|
||
var runLog = &runLogBuffer{
|
||
items: make([]AgentRunRecord, 200),
|
||
size: 200,
|
||
}
|
||
|
||
// Add 追加一条运行记录（环形覆盖写，O(1)）
|
||
func (b *runLogBuffer) Add(rec AgentRunRecord) int64 {
|
||
b.mu.Lock()
|
||
defer b.mu.Unlock()
|
||
b.nextID++
|
||
rec.ID = b.nextID
|
||
b.items[b.head] = rec
|
||
b.head = (b.head + 1) % b.size
|
||
if b.count < b.size {
|
||
b.count++
|
||
}
|
||
return rec.ID
|
||
}
|
||
|
||
// List 按时间倒序返回运行摘要（最新在前）
|
||
//
|
||
// 过滤参数：
|
||
// - limit : 返回条数上限（<=0 时取 50）
|
||
// - scene : 非空时只返回该场景
|
||
// - status : >0 时只返回该状态
|
||
func (b *runLogBuffer) List(limit int, scene string, status int) []AgentRunSummary {
|
||
if limit <= 0 {
|
||
limit = 50
|
||
}
|
||
b.mu.RLock()
|
||
defer b.mu.RUnlock()
|
||
|
||
out := make([]AgentRunSummary, 0, min(limit, b.count))
|
||
// 从最新写入的位置往回遍历（head-1 是最新一条）
|
||
for i := 0; i < b.count && len(out) < limit; i++ {
|
||
idx := (b.head - 1 - i + b.size*2) % b.size
|
||
rec := b.items[idx]
|
||
if scene != "" && rec.Scene != scene {
|
||
continue
|
||
}
|
||
if status > 0 && rec.Status != status {
|
||
continue
|
||
}
|
||
briefs := make([]RunStepBrief, 0, len(rec.Steps))
|
||
for _, st := range rec.Steps {
|
||
briefs = append(briefs, RunStepBrief{
|
||
StepType: st.StepType,
|
||
Status: st.Status,
|
||
DurationMs: st.DurationMs,
|
||
})
|
||
}
|
||
out = append(out, AgentRunSummary{
|
||
ID: rec.ID,
|
||
Scene: rec.Scene,
|
||
Provider: rec.Provider,
|
||
Model: rec.Model,
|
||
CfgSource: rec.CfgSource,
|
||
Status: rec.Status,
|
||
Error: rec.Error,
|
||
TotalMs: rec.TotalMs,
|
||
StartedAt: rec.StartedAt,
|
||
TotalTokens: rec.TotalTokens,
|
||
StepBriefs: briefs,
|
||
})
|
||
}
|
||
return out
|
||
}
|
||
|
||
// Get 按 ID 取完整记录（含全部 step detail）
|
||
func (b *runLogBuffer) Get(id int64) (*AgentRunRecord, bool) {
|
||
b.mu.RLock()
|
||
defer b.mu.RUnlock()
|
||
for i := 0; i < b.count; i++ {
|
||
idx := (b.head - 1 - i + b.size*2) % b.size
|
||
if b.items[idx].ID == id {
|
||
// 返回拷贝，避免调用方拿到内部切片引用后被后续覆盖写污染
|
||
rec := b.items[idx]
|
||
steps := make([]EnhanceStep, len(rec.Steps))
|
||
copy(steps, rec.Steps)
|
||
rec.Steps = steps
|
||
return &rec, true
|
||
}
|
||
}
|
||
return nil, false
|
||
}
|
||
|
||
// RunLogStatsResult 统计聚合结果（面板「统计概览」Tab 数据源）
|
||
type RunLogStatsResult struct {
|
||
Total int `json:"total"` // 缓冲内总运行数
|
||
Success int `json:"success"` // 成功数
|
||
Failed int `json:"failed"` // 失败数
|
||
Blocked int `json:"blocked"` // 守卫拦截数
|
||
AvgMs int `json:"avg_ms"` // 平均耗时（毫秒，只算成功的）
|
||
TotalTokens int `json:"total_tokens"` // token 消耗合计
|
||
SceneCounts map[string]int `json:"scene_counts"` // 按场景分布
|
||
LastError string `json:"last_error"` // 最近一次失败的原因
|
||
LastErrorAt int64 `json:"last_error_at"` // 最近一次失败的时间
|
||
BufferSize int `json:"buffer_size"` // 缓冲容量（面板显示"最近 N 条"）
|
||
}
|
||
|
||
// Stats 聚合统计（O(n)，n<=200 可忽略）
|
||
func (b *runLogBuffer) Stats() RunLogStatsResult {
|
||
b.mu.RLock()
|
||
defer b.mu.RUnlock()
|
||
|
||
res := RunLogStatsResult{
|
||
SceneCounts: map[string]int{},
|
||
BufferSize: b.size,
|
||
}
|
||
sumMs := 0
|
||
okCount := 0
|
||
for i := 0; i < b.count; i++ {
|
||
idx := (b.head - 1 - i + b.size*2) % b.size
|
||
rec := b.items[idx]
|
||
res.Total++
|
||
res.TotalTokens += rec.TotalTokens
|
||
res.SceneCounts[rec.Scene]++
|
||
switch rec.Status {
|
||
case RunStatusOK:
|
||
res.Success++
|
||
sumMs += rec.TotalMs
|
||
okCount++
|
||
case RunStatusBlocked:
|
||
res.Blocked++
|
||
default:
|
||
res.Failed++
|
||
// 只记录最新的一条失败（遍历是从新到旧，第一条命中即最新）
|
||
if res.LastError == "" {
|
||
res.LastError = rec.Error
|
||
res.LastErrorAt = rec.StartedAt
|
||
}
|
||
}
|
||
}
|
||
if okCount > 0 {
|
||
res.AvgMs = sumMs / okCount
|
||
}
|
||
return res
|
||
}
|
||
|
||
// ------------------------------------------------------------------
|
||
// 包级导出函数（供 router 层调用，隐藏 buffer 实现细节）
|
||
// ------------------------------------------------------------------
|
||
|
||
// RunLogList 查询运行摘要列表
|
||
func RunLogList(limit int, scene string, status int) []AgentRunSummary {
|
||
return runLog.List(limit, scene, status)
|
||
}
|
||
|
||
// RunLogGet 查询单条完整记录
|
||
func RunLogGet(id int64) (*AgentRunRecord, bool) {
|
||
return runLog.Get(id)
|
||
}
|
||
|
||
// RunLogStats 查询聚合统计
|
||
func RunLogStats() RunLogStatsResult {
|
||
return runLog.Stats()
|
||
}
|
||
|
||
// recordAgentRun 把一次 Enhance 运行写入环形缓冲
|
||
//
|
||
// 调用方：EnhancerService.Enhance（包装层，成功/失败/拦截统一走这里）
|
||
//
|
||
// 状态推断规则：
|
||
// - err == nil → 成功
|
||
// - steps 里有 medical_guard 步骤 → 守卫拦截（在 err != nil 的前提下）
|
||
// - 其他 → 失败
|
||
func recordAgentRun(req *EnhanceRequest, resp *EnhanceResponse, err error, startedAt time.Time, cfgSource string) {
|
||
rec := AgentRunRecord{
|
||
Scene: req.Scene,
|
||
CfgSource: cfgSource,
|
||
StartedAt: startedAt.Unix(),
|
||
FinishedAt: time.Now().Unix(),
|
||
TotalMs: int(time.Since(startedAt).Milliseconds()),
|
||
RequestSnapshot: buildRequestSnapshot(req),
|
||
}
|
||
|
||
if resp != nil {
|
||
rec.Provider = resp.Provider
|
||
rec.Model = resp.Model
|
||
rec.Steps = resp.Steps
|
||
if resp.TotalMs > 0 {
|
||
rec.TotalMs = resp.TotalMs
|
||
}
|
||
// token 汇总 + 从步骤里兜底 provider/model（守卫拦截时 resp.Provider 为空）
|
||
for _, st := range resp.Steps {
|
||
rec.PromptTokens += st.PromptTokens
|
||
rec.CompletionTokens += st.CompletionTokens
|
||
rec.TotalTokens += st.TotalTokens
|
||
if rec.Provider == "" && st.Provider != "" {
|
||
rec.Provider = st.Provider
|
||
}
|
||
if rec.Model == "" && st.Model != "" {
|
||
rec.Model = st.Model
|
||
}
|
||
}
|
||
}
|
||
|
||
if err == nil {
|
||
rec.Status = RunStatusOK
|
||
} else {
|
||
rec.Error = err.Error()
|
||
rec.Status = RunStatusFail
|
||
// 守卫拦截的错误信息以"医疗守卫拦截"开头（enhancer.go 里约定的前缀）
|
||
if strings.HasPrefix(rec.Error, "医疗守卫拦截") {
|
||
rec.Status = RunStatusBlocked
|
||
}
|
||
}
|
||
|
||
runLog.Add(rec)
|
||
}
|
||
|
||
// snapshotMessage 快照里的单条消息（只留角色 + 截断后的内容）
|
||
type snapshotMessage struct {
|
||
Role string `json:"role"`
|
||
Content string `json:"content"`
|
||
}
|
||
|
||
// requestSnapshot 快照结构（重放时前端按这个结构回填调试台表单）
|
||
type requestSnapshot struct {
|
||
Scene string `json:"scene"`
|
||
Context string `json:"context,omitempty"`
|
||
KBEnabled bool `json:"kb_enabled"`
|
||
TopK int `json:"top_k,omitempty"`
|
||
Provider string `json:"provider,omitempty"`
|
||
Messages []snapshotMessage `json:"messages"`
|
||
Truncated bool `json:"truncated,omitempty"` // 有内容被截断时置 true，前端提示"非完整入参"
|
||
}
|
||
|
||
// buildRequestSnapshot 构造请求入参快照
|
||
//
|
||
// 为什么不直接 json.Marshal(req) 再截断字符串：
|
||
// 粗暴截断会产生非法 JSON，重放时前端没法解析回填。
|
||
// 这里逐字段限长（rune 级，中文安全）后再序列化，保证输出永远是合法 JSON：
|
||
// - context 最多 800 字
|
||
// - 每条消息 content 最多 600 字，最多保留前 8 条
|
||
func buildRequestSnapshot(req *EnhanceRequest) string {
|
||
if req == nil {
|
||
return ""
|
||
}
|
||
const (
|
||
maxContextRunes = 800
|
||
maxMsgRunes = 600
|
||
maxMsgCount = 8
|
||
)
|
||
snap := requestSnapshot{
|
||
Scene: req.Scene,
|
||
KBEnabled: req.KBEnabled,
|
||
TopK: req.TopK,
|
||
Provider: req.Provider,
|
||
}
|
||
snap.Context, snap.Truncated = truncateSnapRunes(req.Context, maxContextRunes, snap.Truncated)
|
||
for i, m := range req.Messages {
|
||
if i >= maxMsgCount {
|
||
snap.Truncated = true
|
||
break
|
||
}
|
||
content, truncated := truncateSnapRunes(m.Content, maxMsgRunes, snap.Truncated)
|
||
snap.Truncated = truncated
|
||
snap.Messages = append(snap.Messages, snapshotMessage{Role: m.Role, Content: content})
|
||
}
|
||
b, err := json.Marshal(snap)
|
||
if err != nil {
|
||
return ""
|
||
}
|
||
return string(b)
|
||
}
|
||
|
||
// truncateSnapRunes 按 rune 截断字符串，并传递"是否发生过截断"标记
|
||
func truncateSnapRunes(s string, max int, alreadyTruncated bool) (string, bool) {
|
||
r := []rune(s)
|
||
if len(r) <= max {
|
||
return s, alreadyTruncated
|
||
}
|
||
return string(r[:max]), true
|
||
}
|