weichenandCursor
b53a97b36d
A 类硬失败返回；manager 降级读路径用 slog.Warn 保持既有容忍语义。 Co-authored-by: Cursor <cursoragent@cursor.com>
432 lines
13 KiB
Go
432 lines
13 KiB
Go
package ctxmgr
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
"unicode/utf8"
|
||
|
||
"github.com/jackc/pgx/v5"
|
||
)
|
||
|
||
// GenerateDailyDigest 生成单日智能摘要 (优先规则/LLM结构化提取，支持自动降级)
|
||
func (m *ContextManager) GenerateDailyDigest(ctx context.Context, sessionID, dayKey string) (*DailyDigest, error) {
|
||
if sessionID == "" || dayKey == "" {
|
||
return nil, fmt.Errorf("sessionID and dayKey are required")
|
||
}
|
||
|
||
var msgs []ChatMessageRecord
|
||
if m.pool != nil {
|
||
rows, err := m.pool.Query(ctx, `
|
||
SELECT id, session_id, user_id, role, content, intent, branch, metadata, day_key::text, category, created_at
|
||
FROM chat_messages
|
||
WHERE session_id = $1 AND day_key = $2::date AND deleted_at IS NULL
|
||
ORDER BY id ASC
|
||
`, sessionID, dayKey)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("query day messages failed: %w", err)
|
||
}
|
||
defer rows.Close()
|
||
|
||
for rows.Next() {
|
||
var rec ChatMessageRecord
|
||
var metaBytes []byte
|
||
var createdAt time.Time
|
||
var uid *int64
|
||
if err := rows.Scan(
|
||
&rec.ID, &rec.SessionID, &uid, &rec.Role, &rec.Content,
|
||
&rec.Intent, &rec.Branch, &metaBytes, &rec.DayKey, &rec.Category, &createdAt,
|
||
); err != nil {
|
||
return nil, err
|
||
}
|
||
if uid != nil {
|
||
rec.UserID = *uid
|
||
}
|
||
if len(metaBytes) > 0 {
|
||
_ = json.Unmarshal(metaBytes, &rec.Metadata)
|
||
}
|
||
rec.CreatedAt = createdAt
|
||
msgs = append(msgs, rec)
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return nil, fmt.Errorf("scan digest messages failed: %w", err)
|
||
}
|
||
} else {
|
||
m.mu.RLock()
|
||
for _, r := range m.messages[sessionID] {
|
||
if r.DayKey == dayKey {
|
||
msgs = append(msgs, r)
|
||
}
|
||
}
|
||
m.mu.RUnlock()
|
||
}
|
||
|
||
return m.GenerateDailyDigestFromMessages(ctx, sessionID, dayKey, msgs)
|
||
}
|
||
|
||
// GenerateDailyDigestFromMessages 仅使用调用方已授权的消息集合生成摘要，避免把不可见正文写入 Prompt/缓存。
|
||
func (m *ContextManager) GenerateDailyDigestFromMessages(ctx context.Context, sessionID, dayKey string, msgs []ChatMessageRecord) (*DailyDigest, error) {
|
||
if sessionID == "" || dayKey == "" {
|
||
return nil, fmt.Errorf("sessionID and dayKey are required")
|
||
}
|
||
var projectID int64
|
||
if m.pool != nil {
|
||
var pid *int64
|
||
_ = m.pool.QueryRow(ctx, `SELECT project_id FROM chat_sessions WHERE session_id = $1`, sessionID).Scan(&pid)
|
||
if pid != nil {
|
||
projectID = *pid
|
||
}
|
||
} else {
|
||
m.mu.RLock()
|
||
if s, ok := m.sessions[sessionID]; ok {
|
||
projectID = s.ProjectID
|
||
}
|
||
m.mu.RUnlock()
|
||
}
|
||
|
||
if len(msgs) == 0 {
|
||
return &DailyDigest{
|
||
SessionID: sessionID,
|
||
ProjectID: projectID,
|
||
DayKey: dayKey,
|
||
Headline: fmt.Sprintf("%s 暂无对话记录", dayKey),
|
||
KeyPoints: []string{},
|
||
Deliverables: []DigestDeliverable{},
|
||
Decisions: []string{},
|
||
Status: "ready",
|
||
Source: "rule",
|
||
UpdatedAt: time.Now(),
|
||
}, nil
|
||
}
|
||
|
||
digest := m.extractRuleDigest(sessionID, projectID, dayKey, msgs)
|
||
applyDigestSemanticShadow(ctx, digest, msgs, m.semanticSummarizerOrDefault())
|
||
|
||
if m.pool != nil {
|
||
keyPointsJSON, _ := json.Marshal(digest.KeyPoints)
|
||
deliverablesJSON, _ := json.Marshal(digest.Deliverables)
|
||
decisionsJSON, _ := json.Marshal(digest.Decisions)
|
||
openQuestionsJSON, _ := json.Marshal(digest.OpenQuestions)
|
||
|
||
var id int64
|
||
var digestVersion int
|
||
err := m.pool.QueryRow(ctx, `
|
||
INSERT INTO chat_daily_digests (
|
||
session_id, project_id, day_key, headline, key_points, deliverables, decisions,
|
||
open_questions, stage_progress, first_message_id, last_message_id, message_count,
|
||
status, source, model, token_usage, digest_version, generated_at, updated_at
|
||
) VALUES (
|
||
$1, $2, $3::date, $4, $5, $6, $7,
|
||
$8, $9, $10, $11, $12,
|
||
$13, $14, $15, $16, 1, now(), now()
|
||
) ON CONFLICT (session_id, day_key) DO UPDATE SET
|
||
headline = EXCLUDED.headline,
|
||
key_points = EXCLUDED.key_points,
|
||
deliverables = EXCLUDED.deliverables,
|
||
decisions = EXCLUDED.decisions,
|
||
open_questions = EXCLUDED.open_questions,
|
||
stage_progress = EXCLUDED.stage_progress,
|
||
first_message_id = EXCLUDED.first_message_id,
|
||
last_message_id = EXCLUDED.last_message_id,
|
||
message_count = EXCLUDED.message_count,
|
||
status = EXCLUDED.status,
|
||
source = EXCLUDED.source,
|
||
model = EXCLUDED.model,
|
||
token_usage = EXCLUDED.token_usage,
|
||
digest_version = chat_daily_digests.digest_version + 1,
|
||
generated_at = now(),
|
||
updated_at = now()
|
||
RETURNING id, digest_version
|
||
`,
|
||
digest.SessionID, digest.ProjectID, digest.DayKey, digest.Headline,
|
||
keyPointsJSON, deliverablesJSON, decisionsJSON, openQuestionsJSON,
|
||
digest.StageProgress, digest.FirstMessageID, digest.LastMessageID, digest.MessageCount,
|
||
digest.Status, digest.Source, digest.Model, digest.TokenUsage,
|
||
).Scan(&id, &digestVersion)
|
||
|
||
if err != nil {
|
||
return nil, fmt.Errorf("upsert chat_daily_digests failed: %w", err)
|
||
}
|
||
digest.ID = id
|
||
digest.DigestVersion = digestVersion
|
||
|
||
// 异步向量化沉淀进 context_memory_chunks (domain = 'daily_digest')
|
||
go m.asyncVectorizeDigest(sessionID, digest)
|
||
} else {
|
||
m.mu.Lock()
|
||
if m.digests[sessionID] == nil {
|
||
m.digests[sessionID] = make(map[string]*DailyDigest)
|
||
}
|
||
digest.ID = time.Now().UnixNano()
|
||
digest.DigestVersion = 1
|
||
m.digests[sessionID][dayKey] = digest
|
||
m.mu.Unlock()
|
||
}
|
||
|
||
return digest, nil
|
||
}
|
||
|
||
// extractRuleDigest 基于业务规则与元数据推导结构化每日摘要
|
||
func (m *ContextManager) extractRuleDigest(
|
||
sessionID string,
|
||
projectID int64,
|
||
dayKey string,
|
||
msgs []ChatMessageRecord,
|
||
) *DailyDigest {
|
||
firstMsg := msgs[0]
|
||
lastMsg := msgs[len(msgs)-1]
|
||
|
||
var headline string
|
||
var keyPoints []string
|
||
var decisions []string
|
||
var deliverables []DigestDeliverable
|
||
var openQuestions []string
|
||
var stageProgress string
|
||
|
||
// 1. 推导一句话总结 Headline
|
||
for _, msg := range msgs {
|
||
if msg.Role == "user" {
|
||
cleaned := strings.TrimSpace(msg.Content)
|
||
runes := []rune(cleaned)
|
||
if len(runes) > 36 {
|
||
headline = fmt.Sprintf("围绕「%s...」展开深入协同与推进", string(runes[:32]))
|
||
} else if len(runes) > 0 {
|
||
headline = fmt.Sprintf("推进「%s」相关工作与技术论证", cleaned)
|
||
}
|
||
break
|
||
}
|
||
}
|
||
if headline == "" {
|
||
headline = fmt.Sprintf("%s 协同推进了 %d 轮业务讨论与任务执行", dayKey, len(msgs))
|
||
}
|
||
|
||
// 2. 遍历消息提取要点、决策、产出与未决
|
||
seenPoints := make(map[string]bool)
|
||
for _, msg := range msgs {
|
||
trimmed := strings.TrimSpace(msg.Content)
|
||
if trimmed == "" {
|
||
continue
|
||
}
|
||
|
||
// 决策提取: 命中审批、里程碑、同意或驳回
|
||
isMilestone := false
|
||
if msg.Metadata != nil {
|
||
if v, ok := msg.Metadata["milestone"].(bool); ok && v {
|
||
isMilestone = true
|
||
}
|
||
if v, ok := msg.Metadata["is_milestone"].(bool); ok && v {
|
||
isMilestone = true
|
||
}
|
||
}
|
||
|
||
if isMilestone || strings.HasPrefix(msg.Intent, "DOCGOV_") || msg.Category == "milestone" || msg.Category == "review" {
|
||
decText := truncateRunes(trimmed, 60)
|
||
if !seenPoints[decText] && len(decisions) < 5 {
|
||
decisions = append(decisions, decText)
|
||
seenPoints[decText] = true
|
||
}
|
||
}
|
||
|
||
// 交付物提取: 含有 task_key 或 document 元数据
|
||
if msg.Metadata != nil {
|
||
if docTitle, ok := msg.Metadata["doc_title"].(string); ok && docTitle != "" {
|
||
docID, _ := msg.Metadata["doc_id"].(string)
|
||
state, _ := msg.Metadata["lifecycle_state"].(string)
|
||
deliverables = append(deliverables, DigestDeliverable{
|
||
DocID: docID,
|
||
Title: docTitle,
|
||
State: state,
|
||
})
|
||
}
|
||
}
|
||
|
||
// 未决事项 / 疑问提取
|
||
if strings.Contains(trimmed, "？") || strings.Contains(trimmed, "?") || strings.Contains(trimmed, "待确认") || strings.Contains(trimmed, "待排期") {
|
||
qText := truncateRunes(trimmed, 50)
|
||
if !seenPoints[qText] && len(openQuestions) < 3 {
|
||
openQuestions = append(openQuestions, qText)
|
||
seenPoints[qText] = true
|
||
}
|
||
}
|
||
|
||
// 阶段推进标记
|
||
if msg.Intent == "ADVANCE_STAGE" {
|
||
stageProgress = truncateRunes(trimmed, 40)
|
||
}
|
||
|
||
// 工作要点 (每日前 4 个典型用户/助手问答)
|
||
if len(keyPoints) < 4 && msg.Role == "user" && !seenPoints[trimmed] {
|
||
point := truncateRunes(trimmed, 50)
|
||
keyPoints = append(keyPoints, point)
|
||
seenPoints[trimmed] = true
|
||
}
|
||
}
|
||
|
||
if len(keyPoints) == 0 {
|
||
keyPoints = append(keyPoints, fmt.Sprintf("当日完成 %d 轮实时交互对话与技术对齐", len(msgs)))
|
||
}
|
||
|
||
now := time.Now()
|
||
return &DailyDigest{
|
||
SessionID: sessionID,
|
||
ProjectID: projectID,
|
||
DayKey: dayKey,
|
||
Headline: headline,
|
||
KeyPoints: keyPoints,
|
||
Deliverables: deliverables,
|
||
Decisions: decisions,
|
||
OpenQuestions: openQuestions,
|
||
StageProgress: stageProgress,
|
||
FirstMessageID: firstMsg.ID,
|
||
LastMessageID: lastMsg.ID,
|
||
MessageCount: len(msgs),
|
||
Status: "ready",
|
||
Source: "rule",
|
||
Model: "rule-heuristic-v1",
|
||
DigestVersion: 1,
|
||
GeneratedAt: &now,
|
||
UpdatedAt: now,
|
||
}
|
||
}
|
||
|
||
// ListDailyDigests 获取会话的历史每日摘要列表 (时间轴侧栏数据源)
|
||
func (m *ContextManager) ListDailyDigests(ctx context.Context, sessionID, from, to string) ([]DailyDigest, error) {
|
||
if sessionID == "" {
|
||
return nil, fmt.Errorf("sessionID is required")
|
||
}
|
||
|
||
if m.pool == nil {
|
||
m.mu.RLock()
|
||
defer m.mu.RUnlock()
|
||
var res []DailyDigest
|
||
if dayMap, ok := m.digests[sessionID]; ok {
|
||
for _, d := range dayMap {
|
||
if from != "" && d.DayKey < from {
|
||
continue
|
||
}
|
||
if to != "" && d.DayKey > to {
|
||
continue
|
||
}
|
||
res = append(res, *d)
|
||
}
|
||
}
|
||
return res, nil
|
||
}
|
||
|
||
var builder strings.Builder
|
||
args := []any{sessionID}
|
||
builder.WriteString(`
|
||
SELECT id, session_id, COALESCE(project_id, 0), day_key::text, headline,
|
||
key_points, deliverables, decisions, open_questions, stage_progress,
|
||
first_message_id, last_message_id, message_count, status, source,
|
||
model, token_usage, digest_version, generated_at, updated_at
|
||
FROM chat_daily_digests
|
||
WHERE session_id = $1
|
||
`)
|
||
|
||
argIdx := 2
|
||
if from != "" {
|
||
builder.WriteString(fmt.Sprintf(" AND day_key >= $%d::date", argIdx))
|
||
args = append(args, from)
|
||
argIdx++
|
||
}
|
||
if to != "" {
|
||
builder.WriteString(fmt.Sprintf(" AND day_key <= $%d::date", argIdx))
|
||
args = append(args, to)
|
||
argIdx++
|
||
}
|
||
builder.WriteString(" ORDER BY day_key DESC")
|
||
|
||
rows, err := m.pool.Query(ctx, builder.String(), args...)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("query chat_daily_digests failed: %w", err)
|
||
}
|
||
defer rows.Close()
|
||
|
||
var digests []DailyDigest
|
||
for rows.Next() {
|
||
var d DailyDigest
|
||
var kpBytes, delivBytes, decBytes, oqBytes []byte
|
||
var firstID, lastID *int64
|
||
if err := rows.Scan(
|
||
&d.ID, &d.SessionID, &d.ProjectID, &d.DayKey, &d.Headline,
|
||
&kpBytes, &delivBytes, &decBytes, &oqBytes, &d.StageProgress,
|
||
&firstID, &lastID, &d.MessageCount, &d.Status, &d.Source,
|
||
&d.Model, &d.TokenUsage, &d.DigestVersion, &d.GeneratedAt, &d.UpdatedAt,
|
||
); err != nil {
|
||
return nil, err
|
||
}
|
||
if firstID != nil {
|
||
d.FirstMessageID = *firstID
|
||
}
|
||
if lastID != nil {
|
||
d.LastMessageID = *lastID
|
||
}
|
||
_ = json.Unmarshal(kpBytes, &d.KeyPoints)
|
||
_ = json.Unmarshal(delivBytes, &d.Deliverables)
|
||
_ = json.Unmarshal(decBytes, &d.Decisions)
|
||
_ = json.Unmarshal(oqBytes, &d.OpenQuestions)
|
||
digests = append(digests, d)
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return nil, fmt.Errorf("scan digests failed: %w", err)
|
||
}
|
||
|
||
return digests, nil
|
||
}
|
||
|
||
// asyncVectorizeDigest 异步将每日摘要向量化沉淀至 context_memory_chunks (domain = 'daily_digest')
|
||
func (m *ContextManager) asyncVectorizeDigest(sessionID string, digest *DailyDigest) {
|
||
if m.embedder == nil || m.pool == nil {
|
||
return
|
||
}
|
||
|
||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||
defer cancel()
|
||
|
||
content := fmt.Sprintf("日期: %s\n总结: %s\n要点: %s\n决策: %s",
|
||
digest.DayKey, digest.Headline,
|
||
strings.Join(digest.KeyPoints, "; "),
|
||
strings.Join(digest.Decisions, "; "),
|
||
)
|
||
|
||
vecs, err := m.embedder.Embed(ctx, []string{content})
|
||
if err != nil || len(vecs) == 0 || len(vecs[0]) == 0 {
|
||
return
|
||
}
|
||
|
||
refID := fmt.Sprintf("digest:%s:%s", sessionID, digest.DayKey)
|
||
vecStr := formatFloatArray(vecs[0])
|
||
|
||
_, _ = m.pool.Exec(ctx, `
|
||
INSERT INTO context_memory_chunks (
|
||
domain, project_id, user_id, reference_id, title, content, chunk_hash,
|
||
embedding, importance_weight, metadata, created_at
|
||
) VALUES (
|
||
'daily_digest', $1, 0, $2, $3, $4, md5($4),
|
||
$5::vector, 0.85, '{}'::jsonb, now()
|
||
) ON CONFLICT (domain, reference_id) DO UPDATE SET
|
||
title = EXCLUDED.title,
|
||
content = EXCLUDED.content,
|
||
chunk_hash = EXCLUDED.chunk_hash,
|
||
embedding = EXCLUDED.embedding,
|
||
importance_weight = EXCLUDED.importance_weight,
|
||
created_at = now()
|
||
`, digest.ProjectID, refID, digest.Headline, content, vecStr)
|
||
}
|
||
|
||
// truncateRunes 字符串按字符数安全截断
|
||
func truncateRunes(s string, maxRunes int) string {
|
||
runes := []rune(s)
|
||
if len(runes) <= maxRunes {
|
||
return s
|
||
}
|
||
return string(runes[:maxRunes]) + "..."
|
||
}
|
||
|
||
// 确保 utf8 引用
|
||
var _ = utf8.RuneCountInString
|
||
var _ = pgx.ErrNoRows
|