feat(agent): 任务内主 Agent 支持多会话，可切换且上下文可重置
主 Agent 会话原为一个任务恒定单例，对话历史随 transcript 只增不减、 上下文越滚越大。现引入会话分段(seq)：新建会话切一个干净的 transcript 段，任务的图谱/资产/目标共享保留，旧段以只读历史随时切回查看。 - db: 新增 main_sessions 表与 activity.main_seg 列(ALTER ADD COLUMN IF NOT EXISTS 兼容旧库)；ActivitySessionFilter 按段过滤(段 0 兼容 旧 NULL 行)；新增 CurrentMainSeg/ListMainSessions/NewMainSession； per-session token 统计按 main:<seg> 分段 - agent: MainAgent.Chat 增加 mainSeg，transcript SessionID 分段 (段 0 保留旧名 exp<id>-main 兼容既有对话) - server: parseActivitySession 支持 main:<seq>；chat handler 每轮 stamp 当前段；新增 GET main-sessions / POST main-session/new - web: 主 Agent 分组渲染多段会话，支持新建(轻确认)、切换、旧段只读
This commit is contained in:
Autumn-27 committed
1 parent
9ad88d5ce8
commit
ee3eb22433
8 files changed
+360
-51
No files matched your search
+79
-12
@@ -68,6 +68,9 @@ type Activity struct {
|
||||
CacheWriteTokens *int `json:"cache_write_tokens,omitempty"`
|
||||
SourceTaskID int64 `json:"source_task_id,omitempty"`
|
||||
Inherited bool `json:"inherited,omitempty"`
|
||||
// MainSeg is the main-agent conversation segment (nil for non-mainagent rows;
|
||||
// nil/0 == the original session). Lets the UI route a mainagent row to its segment.
|
||||
MainSeg *int `json:"main_seg,omitempty"`
|
||||
}
|
||||
|
||||
// TokenUsage is a per-worker token aggregate (TokenStatsByWorker).
|
||||
@@ -982,10 +985,10 @@ func (s *ExplorationStore) AppendActivity(a Activity) (int64, error) {
|
||||
}
|
||||
var id int64
|
||||
err := s.db.QueryRow(`
|
||||
INSERT INTO activity(exploration_id, node_id, worker, kind, tool, tool_use_id, is_error, summary, detail, metadata, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens)
|
||||
VALUES ($1,$2,NULLIF($3,''),NULLIF($4,''),NULLIF($5,''),NULLIF($6,''),$7,NULLIF($8,''),NULLIF($9,''),$10,$11,$12,$13,$14)
|
||||
INSERT INTO activity(exploration_id, node_id, worker, kind, tool, tool_use_id, is_error, summary, detail, metadata, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, main_seg)
|
||||
VALUES ($1,$2,NULLIF($3,''),NULLIF($4,''),NULLIF($5,''),NULLIF($6,''),$7,NULLIF($8,''),NULLIF($9,''),$10,$11,$12,$13,$14,$15)
|
||||
RETURNING id`, s.expID, a.NodeID, utf8Clean(a.Worker), utf8Clean(a.Kind), utf8Clean(a.Tool), utf8Clean(a.ToolUseID), a.IsError, utf8Clean(a.Summary), utf8Clean(a.Detail),
|
||||
metadata, a.InputTokens, a.OutputTokens, a.CacheReadTokens, a.CacheWriteTokens).Scan(&id)
|
||||
metadata, a.InputTokens, a.OutputTokens, a.CacheReadTokens, a.CacheWriteTokens, a.MainSeg).Scan(&id)
|
||||
return id, err
|
||||
}
|
||||
|
||||
@@ -1197,7 +1200,7 @@ func (s *ExplorationStore) TokenStatsByWorker() ([]TokenUsage, error) {
|
||||
func (s *ExplorationStore) TokenStatsBySession() ([]SessionTokenUsage, error) {
|
||||
rows, err := s.db.Query(`SELECT
|
||||
CASE
|
||||
WHEN COALESCE(a.worker,'')='mainagent' THEN 'main'
|
||||
WHEN COALESCE(a.worker,'')='mainagent' THEN 'main:' || COALESCE(a.main_seg,0)::text
|
||||
WHEN COALESCE(a.worker,'')='planner' THEN 'plan'
|
||||
WHEN n.id IS NOT NULL THEN 'intent:' || n.id::text
|
||||
ELSE ''
|
||||
@@ -1234,7 +1237,7 @@ func (s *ExplorationStore) ActivityList(nodeID *int64, sinceID int64, limit int)
|
||||
}
|
||||
var rows *sql.Rows
|
||||
var err error
|
||||
const cols = `id, node_id, COALESCE(worker,''), COALESCE(kind,''), COALESCE(tool,''), COALESCE(tool_use_id,''), is_error, COALESCE(summary,''), metadata, created_at, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens`
|
||||
const cols = `id, node_id, COALESCE(worker,''), COALESCE(kind,''), COALESCE(tool,''), COALESCE(tool_use_id,''), is_error, COALESCE(summary,''), metadata, created_at, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, main_seg`
|
||||
if nodeID != nil {
|
||||
rows, err = s.db.Query(`SELECT `+cols+`
|
||||
FROM activity WHERE exploration_id=$1 AND node_id=$2 AND id>$3 ORDER BY id LIMIT $4`, s.expID, *nodeID, sinceID, limit)
|
||||
@@ -1251,7 +1254,7 @@ FROM activity WHERE exploration_id=$1 AND id>$2 ORDER BY id LIMIT $3`, s.expID,
|
||||
for rows.Next() {
|
||||
var a Activity
|
||||
if err := rows.Scan(&a.ID, &a.NodeID, &a.Worker, &a.Kind, &a.Tool, &a.ToolUseID, &a.IsError, &a.Summary, &a.Metadata, &a.CreatedAt,
|
||||
&a.InputTokens, &a.OutputTokens, &a.CacheReadTokens, &a.CacheWriteTokens); err != nil {
|
||||
&a.InputTokens, &a.OutputTokens, &a.CacheReadTokens, &a.CacheWriteTokens, &a.MainSeg); err != nil {
|
||||
return nil, sinceID, err
|
||||
}
|
||||
if a.ID > cursor {
|
||||
@@ -1301,20 +1304,33 @@ func (s *ExplorationStore) ActivityListForTerminalIntent(nodeID, sinceID int64,
|
||||
|
||||
// ActivitySessionFilter selects one UI "session" within a task's activity stream.
|
||||
// Exactly one field is meaningful:
|
||||
// - Worker != "" → filter by worker name (Main = "mainagent", Plan = "planner").
|
||||
// - Main == true → the main-agent session (worker="mainagent") for one segment
|
||||
// (MainSeg; nil/0 = the original segment, which also matches legacy NULL rows).
|
||||
// - Worker != "" → filter by worker name (Plan = "planner").
|
||||
// Goal Agent 的第 0 轮拆解也以 worker="planner" 落库，故 Plan 会话完整覆盖 Goal+Planner。
|
||||
// - NodeID != nil → a Worker session, filtered by node_id (= intent id).
|
||||
//
|
||||
// A zero value (both empty) matches the whole task (no session filter).
|
||||
// A zero value (all empty) matches the whole task (no session filter).
|
||||
type ActivitySessionFilter struct {
|
||||
Worker string
|
||||
NodeID *int64
|
||||
Worker string
|
||||
NodeID *int64
|
||||
Main bool // main-agent session
|
||||
MainSeg *int // segment for the main session (nil == current, resolved by caller; 0 == original)
|
||||
}
|
||||
|
||||
func (f ActivitySessionFilter) cond(argStart int) (string, []any) {
|
||||
switch {
|
||||
case f.NodeID != nil:
|
||||
return fmt.Sprintf(" AND node_id=$%d", argStart), []any{*f.NodeID}
|
||||
case f.Main:
|
||||
seg := 0
|
||||
if f.MainSeg != nil {
|
||||
seg = *f.MainSeg
|
||||
}
|
||||
if seg == 0 { // original segment also owns legacy rows with NULL main_seg
|
||||
return " AND worker='mainagent' AND COALESCE(main_seg,0)=0", nil
|
||||
}
|
||||
return fmt.Sprintf(" AND worker='mainagent' AND main_seg=$%d", argStart), []any{seg}
|
||||
case f.Worker != "":
|
||||
return fmt.Sprintf(" AND worker=$%d", argStart), []any{f.Worker}
|
||||
default:
|
||||
@@ -1331,7 +1347,7 @@ func (s *ExplorationStore) ActivityPage(f ActivitySessionFilter, before int64, l
|
||||
if limit <= 0 {
|
||||
limit = 200
|
||||
}
|
||||
const cols = `id, node_id, COALESCE(worker,''), COALESCE(kind,''), COALESCE(tool,''), COALESCE(tool_use_id,''), is_error, COALESCE(summary,''), metadata, created_at, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens`
|
||||
const cols = `id, node_id, COALESCE(worker,''), COALESCE(kind,''), COALESCE(tool,''), COALESCE(tool_use_id,''), is_error, COALESCE(summary,''), metadata, created_at, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, main_seg`
|
||||
args := []any{s.expID}
|
||||
cond, cargs := f.cond(len(args) + 1)
|
||||
args = append(args, cargs...)
|
||||
@@ -1351,7 +1367,7 @@ ORDER BY id DESC LIMIT $%d`, cond, beforeArg, beforeArg, limitArg)
|
||||
for rows.Next() {
|
||||
var a Activity
|
||||
if err := rows.Scan(&a.ID, &a.NodeID, &a.Worker, &a.Kind, &a.Tool, &a.ToolUseID, &a.IsError, &a.Summary, &a.Metadata, &a.CreatedAt,
|
||||
&a.InputTokens, &a.OutputTokens, &a.CacheReadTokens, &a.CacheWriteTokens); err != nil {
|
||||
&a.InputTokens, &a.OutputTokens, &a.CacheReadTokens, &a.CacheWriteTokens, &a.MainSeg); err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
desc = append(desc, a)
|
||||
@@ -1428,6 +1444,57 @@ func (s *ExplorationStore) ActivityMaxID() (int64, error) {
|
||||
return max.Int64, nil
|
||||
}
|
||||
|
||||
// MainSession is one resettable main-agent conversation segment of a task. Segment 0
|
||||
// is the original session (implicit, never stored); further segments are created by
|
||||
// "新建会话" to start the main agent on a clean transcript while the task's graph,
|
||||
// assets and goal stay shared.
|
||||
type MainSession struct {
|
||||
Seq int `json:"seq"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
}
|
||||
|
||||
// CurrentMainSeg returns the highest main-session segment (0 when none created yet).
|
||||
func (s *ExplorationStore) CurrentMainSeg() (int, error) {
|
||||
var seg int
|
||||
err := s.db.QueryRow(`SELECT COALESCE(MAX(seq),0) FROM main_sessions WHERE exploration_id=$1`, s.expID).Scan(&seg)
|
||||
return seg, err
|
||||
}
|
||||
|
||||
// ListMainSessions returns every main-session segment newest-first, always including
|
||||
// the implicit original segment 0. Segment 0 carries no stored timestamp.
|
||||
func (s *ExplorationStore) ListMainSessions() ([]MainSession, error) {
|
||||
rows, err := s.db.Query(`SELECT seq, created_at FROM main_sessions WHERE exploration_id=$1 ORDER BY seq DESC`, s.expID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := []MainSession{}
|
||||
for rows.Next() {
|
||||
var m MainSession
|
||||
if err := rows.Scan(&m.Seq, &m.CreatedAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, m)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, MainSession{Seq: 0}) // implicit original session (oldest)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// NewMainSession creates the next main-session segment (seq = current+1) and returns
|
||||
// it. It touches only main_sessions — the task's exploration graph/assets/goal are
|
||||
// untouched, so the new session starts with a clean transcript over the same task.
|
||||
func (s *ExplorationStore) NewMainSession() (MainSession, error) {
|
||||
var m MainSession
|
||||
err := s.db.QueryRow(`
|
||||
INSERT INTO main_sessions(exploration_id, seq)
|
||||
VALUES ($1, COALESCE((SELECT MAX(seq) FROM main_sessions WHERE exploration_id=$1),0)+1)
|
||||
RETURNING seq, created_at`, s.expID).Scan(&m.Seq, &m.CreatedAt)
|
||||
return m, err
|
||||
}
|
||||
|
||||
// ActivityDetail lazily returns the full detail blob for one step.
|
||||
func (s *ExplorationStore) ActivityDetail(id int64) (string, error) {
|
||||
var d sql.NullString
|
||||
|
||||
@@ -309,11 +309,20 @@ CREATE TABLE IF NOT EXISTS activity (
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
ALTER TABLE activity ADD COLUMN IF NOT EXISTS metadata JSONB NOT NULL DEFAULT '{}';
|
||||
-- main_seg segments the main-agent session into resettable conversations: a new
|
||||
-- main session bumps the segment so its transcript + activity start clean while the
|
||||
-- task's graph/assets/goal are untouched. NULL == legacy rows == segment 0 (the
|
||||
-- original session). Only worker='mainagent' rows carry it.
|
||||
ALTER TABLE activity ADD COLUMN IF NOT EXISTS main_seg INTEGER;
|
||||
CREATE INDEX IF NOT EXISTS idx_act_node ON activity(exploration_id, node_id, id);
|
||||
CREATE INDEX IF NOT EXISTS idx_act_since ON activity(exploration_id, id);
|
||||
-- Main/Plan history pages filter by worker (both carry NULL node_id, so idx_act_node
|
||||
-- can't distinguish them); this covers reverse pagination of those sessions.
|
||||
CREATE INDEX IF NOT EXISTS idx_act_worker ON activity(exploration_id, worker, id);
|
||||
-- Main-session pages filter by segment on top of worker='mainagent'; this partial
|
||||
-- index covers reverse pagination within one segment.
|
||||
CREATE INDEX IF NOT EXISTS idx_act_main_seg ON activity(exploration_id, main_seg, id)
|
||||
WHERE worker='mainagent';
|
||||
-- Task-list polls aggregate result usage and find the latest event repeatedly.
|
||||
-- Cover the token columns for index-only aggregation and the timestamp order for
|
||||
-- per-exploration latest-activity lookups.
|
||||
@@ -322,6 +331,18 @@ CREATE INDEX IF NOT EXISTS idx_act_result_usage ON activity(exploration_id)
|
||||
WHERE kind='result';
|
||||
CREATE INDEX IF NOT EXISTS idx_act_latest ON activity(exploration_id, created_at DESC);
|
||||
|
||||
-- main_sessions records the resettable main-agent conversation segments of a task.
|
||||
-- Segment 0 (the original session) is implicit and never stored; this table holds
|
||||
-- only the extra segments created by "新建会话" (seq >= 1). The current segment is
|
||||
-- MAX(seq) or 0. Each segment gets its own transcript file + activity slice; the
|
||||
-- task's exploration graph/assets/goal are shared and never reset.
|
||||
CREATE TABLE IF NOT EXISTS main_sessions (
|
||||
exploration_id BIGINT NOT NULL REFERENCES explorations(id) ON DELETE CASCADE,
|
||||
seq INTEGER NOT NULL,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (exploration_id, seq)
|
||||
);
|
||||
|
||||
-- =====================================================================
|
||||
-- C. LLM profiles
|
||||
-- =====================================================================
|
||||
|
||||
Reference in new issue
Block a user