aiops/docs/agent/10-会话管理.md

10 - 会话管理

本文由 opsxcli 同名设计文档适配改写，结合 aiops Agent CLI 的 Web 实现（backend/model/agent_cli.go + controller/agent_cli.go）。 核心改造：opsxcli 用本地 JSONL 文件存储会话（单用户 CLI 场景）；aiops 用 GORM 数据库表存储会话与消息，支持多用户并发、按 user_id 隔离、REST 查询与历史回放。

设计理念

• 会话管理：Agent 引擎本身无状态（每次任务独立），所有对话过程持久化到会话层，支持历史回溯、审计、断点排查
• 多用户隔离：Web 场景下多个用户同时使用 Agent CLI，会话必须按 user_id 隔离
• 结构化查询：会话需要列表、筛选、删除等业务操作，必须走 DB 而非文件

为什么改用 DB？opsxcli 是 CLI 工具，单用户、单机使用，JSONL 文件够用且零依赖；aiops 是 Web 平台，多用户并发、需要列表查询和按主机/用户筛选、需要与其他业务表（Server、User）关联 —— DB 是必然选择。

架构总览

┌────────────────────────────────────────────────────┐
│                  REST + WebSocket                   │
│   /ws/agent-cli       (WebSocket 推流)              │
│   /aiops/agent-cli/*  (REST 会话管理)               │
├────────────────────────────────────────────────────┤
│            controller/agent_cli.go                  │
│   AgentCLIHandler                                    │
│   ├─ Run (WebSocket)                                │
│   ├─ ListSessions / GetSession / DeleteSession      │
│   └─ agentConn (单连接运行态)                        │
├────────────────────────────────────────────────────┤
│            model/agent_cli.go (GORM)                │
│   AgentCLISession  (agent_cli_sessions 表)          │
│   AgentCLIMessage  (agent_cli_messages 表)          │
├────────────────────────────────────────────────────┤
│            MySQL / SQLite (aiops 既有 DB)            │
└────────────────────────────────────────────────────┘

核心数据模型

AgentCLISession（会话）

// backend/model/agent_cli.go

type AgentCLISession struct {
    ID          uint      `gorm:"primarykey" json:"id"`
    UserID      uint      `gorm:"index" json:"user_id"`                  // 发起用户（隔离）
    ServerID    uint      `gorm:"index" json:"server_id"`                // 目标主机
    ServerName  string    `gorm:"size:128" json:"server_name"`           // 主机名快照
    ServerHost  string    `gorm:"size:64" json:"server_host"`            // 主机 IP 快照
    Title       string    `gorm:"size:255" json:"title"`                 // 任务标题（query 截断）
    Status      string    `gorm:"size:32;default:running" json:"status"` // running/done/aborted/error
    Provider    string    `gorm:"size:50" json:"provider"`               // LLM 提供商
    Model       string    `gorm:"size:100" json:"model"`                 // 模型名
    Rounds      int       `json:"rounds"`                                // ReAct 轮次
    TotalTokens int       `json:"total_tokens"`                          // token 消耗
    Duration    int64     `json:"duration"`                              // 耗时（毫秒）
    Error       string    `gorm:"type:text" json:"error,omitempty"`      // 错误信息
    CreatedAt   time.Time `json:"created_at"`
    UpdatedAt   time.Time `json:"updated_at"`
}

func (AgentCLISession) TableName() string { return "agent_cli_sessions" }

设计要点：

• UserID 索引：多用户隔离的核心，所有查询都按 user_id 过滤
• ServerID + ServerName + ServerHost：关联主机的同时保存快照（主机 later 被删/改名时历史仍可读）
• Status 四态：running（进行中）/ done（完成）/ aborted（用户中断）/ error（出错或达最大轮次）
• Title：query 截断到 60 字符，作为会话列表的显示标题

AgentCLIMessage（消息）

type AgentCLIMessage struct {
    ID         uint      `gorm:"primarykey" json:"id"`
    SessionID  uint      `gorm:"index;not null" json:"session_id"`
    Round      int       `json:"round"`                        // ReAct 轮次
    Type       string    `gorm:"size:32" json:"type"`          // thought/tool_call/tool_result/done/error/session
    Role       string    `gorm:"size:32" json:"role"`          // user/assistant/tool
    Content    string    `gorm:"type:text" json:"content"`     // 思考文本/最终答案
    ToolName   string    `gorm:"size:64" json:"tool_name"`     // 工具名
    ToolArgs   string    `gorm:"type:text" json:"tool_args"`   // 工具参数 JSON
    ToolOutput string    `gorm:"type:text" json:"tool_output"` // 工具输出
    Risk       string    `gorm:"size:32" json:"risk"`          // 风险等级
    Success    bool      `json:"success"`
    Duration   string    `gorm:"size:32" json:"duration"`      // 工具耗时
    Tokens     int       `json:"tokens"`
    CreatedAt  time.Time `gorm:"index" json:"created_at"`
}

func (AgentCLIMessage) TableName() string { return "agent_cli_messages" }

消息类型映射引擎事件：

Event.Type	DB Type	DB Role	Content	ToolName	ToolOutput
用户输入	(无 Type)	user	query	—	—
thought	thought	assistant	AI 思考文本	—	—
tool_call	tool_call	tool	(空)	工具名	(空，Args 在 ToolArgs)
tool_result	tool_result	tool	输出	工具名	输出（冗余便于查询）
done	done	assistant	最终答案	—	—
error	error	assistant	错误信息	—	—
progress	—	—	—	—	—

会话生命周期

WebSocket start {query}
    │
    ▼
startTask
    ├─ 创建 AgentCLISession (status=running)
    ├─ 存用户消息 (role=user, content=query)
    ├─ 构建 Engine、注入 Callback
    └─ go engine.Run(ctx, query, serverInfo)
         │
         │  每个事件触发 onEvent → 落库 agent_cli_messages
         │  ├─ thought   → persistMsg(type=thought, role=assistant)
         │  ├─ tool_call → persistMsg(type=tool_call, role=tool, args, risk)
         │  ├─ tool_result → persistMsg(type=tool_result, role=tool, output, success)
         │  ├─ done      → persistMsg(type=done, role=assistant, answer, tokens)
         │  └─ error     → persistMsg(type=error, role=assistant, message)
         │
         ▼
    finish (更新 session.status / rounds / duration / error)
         ├─ ctx.Err() != nil → status=aborted
         ├─ err != nil       → status=error
         └─ 否则              → status=done

controller 实现

startTask（创建会话）

// backend/controller/agent_cli.go

func (ac *agentConn) startTask(sshCfg remoteexec.SSHConfig, llmCfg model.LLMConfig, query string) {
    ac.abort()  // 取消上一次任务

    // 1. 创建 DB 会话
    title := query
    if r := []rune(title); len(r) > 60 {
        title = string(r[:60]) + "..."
    }
    sess := &model.AgentCLISession{
        UserID: ac.userID, ServerID: ac.server.ID,
        ServerName: ac.server.Hostname, ServerHost: ac.server.IP,
        Title: title, Status: "running",
        Provider: llmCfg.Provider, Model: llmCfg.Model,
    }
    ac.db.Create(sess)
    ac.session = sess

    // 2. 存用户消息
    ac.persistMsg(sess.ID, 0, "user", "user", query, "", "", "", "", false, "", 0)

    // 3. 构建 Engine
    engine, _ := agentcli.NewEngine(...)
    engine.SetCallback(ac.onEvent)
    engine.SetConfirmFn(ac.confirmFn)

    // 4. 启动 ReAct 循环（goroutine）
    go func() {
        defer engine.Close()
        _, err := engine.Run(ctx, query, serverInfo)
        // ... finish
    }()
}

onEvent（事件落库 + 推 WS）

func (ac *agentConn) onEvent(e core.Event) {
    ev := wsEvent{...}  // 序列化
    if ac.session != nil {
        ev.SessionID = ac.session.ID
    }
    ac.write(ev)  // 推 WS

    // 持久化（progress 不入库）
    if ac.session == nil { return }
    sid := ac.session.ID
    switch e.Type {
    case core.EventThought:
        ac.persistMsg(sid, e.Round, "thought", "assistant", e.Thought, ...)
    case core.EventToolCall:
        argsJSON, _ := json.Marshal(e.ToolArgs)
        ac.persistMsg(sid, e.Round, "tool_call", "tool", "", e.ToolName, string(argsJSON), "", e.Risk, ...)
    case core.EventToolResult:
        ac.persistMsg(sid, e.Round, "tool_result", "tool", e.Output, e.ToolName, "", "", "", e.Success, e.Duration, 0)
    case core.EventDone:
        ac.persistMsg(sid, e.Round, "done", "assistant", e.Answer, ..., true, "", e.Tokens)
    case core.EventError:
        ac.persistMsg(sid, e.Round, "error", "assistant", e.Message, ...)
    }
}

finish（结束会话）

func (ac *agentConn) finish(sess *model.AgentCLISession, status, errMsg string, rounds int, dur time.Duration) {
    // 推 WS 结束事件
    ac.write(wsEvent{Type: status, SessionID: sess.ID, Message: errMsg, Round: rounds})
    // 更新 DB
    updates := map[string]interface{}{
        "status": status, "rounds": rounds, "duration": dur.Milliseconds(),
    }
    if errMsg != "" {
        updates["error"] = errMsg
    }
    ac.db.Model(sess).Updates(updates)
}

REST 接口

// backend/router/routes.go

agentCLIGroup := authorized.Group("/aiops/agent-cli")
{
    agentCLIGroup.GET("/sessions",        agentCLIHandler.ListSessions)   // 列表（按 user_id 过滤）
    agentCLIGroup.GET("/sessions/:id",    agentCLIHandler.GetSession)     // 详情（含全部消息）
    agentCLIGroup.DELETE("/sessions/:id", agentCLIHandler.DeleteSession)  // 删除（校验归属）
}

前端 frontend/src/api/agentCli.ts 调用这些接口实现「会话历史」页面：

• 列表展示用户的所有会话（标题、主机、状态、耗时、token）
• 点击某会话查看完整 ReAct 推理过程（thought / tool_call / tool_result 时间线）
• 删除会话（连带消息）

WebSocket 协议

后端 → 前端（wsEvent）

type wsEvent struct {
    Type      string                 `json:"type"`       // session/thought/tool_call/tool_result/progress/done/error/confirm
    Round     int                    `json:"round,omitempty"`
    SessionID uint                   `json:"session_id,omitempty"`
    Thought   string                 `json:"thought,omitempty"`
    ToolName  string                 `json:"tool_name,omitempty"`
    ToolArgs  map[string]interface{} `json:"tool_args,omitempty"`
    Risk      string                 `json:"risk,omitempty"`
    Output    string                 `json:"output,omitempty"`
    Success   bool                   `json:"success,omitempty"`
    Duration  string                 `json:"duration,omitempty"`
    Tokens    int                    `json:"tokens,omitempty"`
    Answer    string                 `json:"answer,omitempty"`
    Message   string                 `json:"message,omitempty"`
    ConfirmID string                 `json:"confirm_id,omitempty"`  // confirm 事件专属
    Command   string                 `json:"command,omitempty"`      // confirm 事件专属
}

前端 → 后端（wsControl）

type wsControl struct {
    Type      string `json:"type"`                 // start/confirm/deny/abort
    Query     string `json:"query,omitempty"`      // start 时任务描述
    ConfirmID string `json:"confirm_id,omitempty"` // confirm/deny 时
}

并发与隔离

单连接串行

每个 WebSocket 连接（agentConn）同一时刻只运行一个任务：

func (ac *agentConn) startTask(...) {
    ac.abort()  // 取消上一次任务，再启动新的
}

多用户隔离

• 每个会话绑定 UserID（JWT 中间件注入）
• REST 查询自动按 user_id 过滤
• 删除会话时校验归属

线程安全

• agentConn.writeMu sync.Mutex —— 串行化 WebSocket 写入
• agentConn.cfMu sync.Mutex —— 保护 confirms map
• 引擎 core.Agent.mu sync.Mutex —— 保护 token 计数等内部状态

持久化策略

落库失败不阻断

func (ac *agentConn) persistMsg(...) {
    if err := ac.db.Create(msg).Error; err != nil {
        _ = err  // 持久化失败不阻断主流程
    }
}

DB 写入失败时，WS 推送仍正常，用户体验不受影响（只是历史回放缺这条）。

progress 不入库

progress 事件每个轮次都发，且只含 round 和 token，价值低、频率高，不入库（避免 DB 压力）。

历史回放

前端通过 GET /sessions/:id 拿到完整消息列表，按时间顺序渲染：

[会话 #42] 检查 nginx 启动失败的原因
  ├─ 用户：检查 nginx 启动失败的原因
  ├─ Round 1 [thought]：我先查看 nginx 服务状态...
  ├─ Round 1 [tool_call] ssh_execute "systemctl status nginx" (高风险)
  ├─ Round 1 [tool_result] failed (1.2s) [exit_code=3]
  ├─ Round 2 [thought]：服务 failed，查看日志...
  ├─ Round 2 [tool_call] ssh_execute "journalctl -u nginx -n 500" (安全)
  ├─ Round 2 [tool_result] 500 行 (0.8s)
  ├─ Round 3 [thought]：日志太长，先分析...
  ├─ Round 3 [tool_call] analyze_output(output, focus="error")
  ├─ Round 3 [tool_result] 8 行疑似错误 (0.05s)
  ├─ Round 4 [thought]：端口被占，确认一下...
  ├─ Round 4 [tool_call] ssh_execute "ss -tlnp | grep :80" (安全)
  ├─ Round 4 [tool_result] (0.3s)
  └─ [done] 根因：80 端口被 apache 占用。建议：systemctl stop httpd && systemctl start nginx

这是 opsxcli「会话导出为 Markdown」的 Web 升级版 —— 不需要导出，直接在线回放。

扩展指南

添加会话标签 / 收藏

1. 在 AgentCLISession 增加 Tags string / Starred bool 字段
2. 增加 REST 接口 PATCH /sessions/:id
3. 前端列表增加标签筛选

会话统计

// 例：用户本月使用的 token 总数
db.Model(&AgentCLIMessage{}).
    Where("session_id IN (?) AND type = ?",
        db.Model(&AgentCLISession{}).Select("id").Where("user_id = ?", uid),
        "done").
    Select("COALESCE(SUM(tokens), 0)").Scan(&totalTokens)

导出会话为 Markdown

参考 opsxcli 的 ExportMarkdown，在 controller 增加一个 REST 接口，遍历消息组装 Markdown 文本返回。

常见陷阱

陷阱	解决方案
多用户看到别人的会话	所有查询按 user_id 过滤，删除时校验归属
WebSocket 断开导致会话悬挂	ctx.Err() 触发 finish(status=aborted)；前端重连后可查看历史
消息表无限增长	定期清理老会话（运维任务）；或按 created_at 分区
落库失败丢失历史	不阻断主流程，但应在日志记录；progress 不入库减少压力
长任务超过 WS 超时	ReAct 循环每步检查 ctx.Err()，前端断开 → ctx cancel → finish(aborted)
并发写 WS 错乱	writeMu sync.Mutex 串行化写入
ConfirmFn 无人响应	5 分钟超时自动拒绝
lastRound 不准	从 DB 查 MAX(round) 而非内存计数（断线重连场景）

与 opsxcli 的差异总结

维度	opsxcli	aiops
存储后端	JSONL 文件（~/.opsxcli/agent/sessions/）	GORM DB（agent_cli_sessions / agent_cli_messages 表）
多用户	不支持（单用户 CLI）	原生支持（user_id 索引隔离）
会话结构	{session-id}/meta.json + messages.jsonl	单表 sessions + 单表 messages（一对多）
查询能力	只能遍历文件	SQL 任意查询、JOIN、聚合
导出	ExportMarkdown / ExportJSON / ExportToFile	在线回放（前端时间线）
消息字段	role/content/tool_calls	type/role/content/tool_name/tool_args/tool_output/risk/success/duration/tokens
接口	Manager 接口（10 方法）	REST（3 方法）+ WebSocket 推流

存储载体变了，但「会话 = 元数据 + 消息流」的核心模型保留。DB 方案在多用户 Web 场景下是必然选择，也使会话与其他业务表（Server、User、LLMConfig）的关联变得自然。

作者：Logan.Li