432 lines
14 KiB
Go
432 lines
14 KiB
Go
package ctxmgr
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"sync"
|
|
"time"
|
|
|
|
"chat-agent-system/internal/storage"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
// ContextManager 多级与多类型上下文总管理器
|
|
type ContextManager struct {
|
|
pool *pgxpool.Pool
|
|
cas *storage.CAS
|
|
reflector *PreferenceReflector
|
|
decayCfg TimeDecayConfig
|
|
|
|
// 离线降级内存缓存
|
|
mu sync.RWMutex
|
|
users map[int64]*UserPreference
|
|
projects map[int64]*ProjectCollectiveContext
|
|
messages map[string][]ChatMessageRecord // session_id -> messages
|
|
activities map[int64][]ActivityItem // project_id -> activities
|
|
nextActivityID int64
|
|
nextMessageID int64
|
|
}
|
|
|
|
// NewContextManager 创建上下文管理器
|
|
func NewContextManager(pool *pgxpool.Pool, cas *storage.CAS, decayCfg TimeDecayConfig) *ContextManager {
|
|
if decayCfg.HalfLifeDays <= 0 {
|
|
decayCfg = DefaultTimeDecayConfig()
|
|
}
|
|
return &ContextManager{
|
|
pool: pool,
|
|
cas: cas,
|
|
reflector: NewPreferenceReflector(),
|
|
decayCfg: decayCfg,
|
|
users: make(map[int64]*UserPreference),
|
|
projects: make(map[int64]*ProjectCollectiveContext),
|
|
messages: make(map[string][]ChatMessageRecord),
|
|
activities: make(map[int64][]ActivityItem),
|
|
nextActivityID: 1,
|
|
nextMessageID: 1,
|
|
}
|
|
}
|
|
|
|
// RecordUserMessage 记录单轮对话消息并自动驱动工作习惯偏好更新
|
|
func (m *ContextManager) RecordUserMessage(ctx context.Context, sessionID string, userID int64, role, content, intent, branch string, meta map[string]any) error {
|
|
if sessionID == "" || content == "" {
|
|
return errors.New("sessionID and content required")
|
|
}
|
|
|
|
rec := ChatMessageRecord{
|
|
SessionID: sessionID,
|
|
UserID: userID,
|
|
Role: role,
|
|
Content: content,
|
|
Intent: intent,
|
|
Branch: branch,
|
|
Metadata: meta,
|
|
CreatedAt: time.Now(),
|
|
}
|
|
|
|
if m.pool != nil {
|
|
// 确保会话存在
|
|
_, _ = m.pool.Exec(ctx, `
|
|
INSERT INTO chat_sessions (session_id, user_id, title, created_at, updated_at)
|
|
VALUES ($1, $2, '对话协同会话', now(), now())
|
|
ON CONFLICT (session_id) DO UPDATE SET updated_at = now()
|
|
`, sessionID, userID)
|
|
|
|
metaJSON, _ := json.Marshal(meta)
|
|
_, err := m.pool.Exec(ctx, `
|
|
INSERT INTO chat_messages (session_id, user_id, role, content, intent, branch, metadata, created_at)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, now())
|
|
`, sessionID, userID, role, content, intent, branch, metaJSON)
|
|
if err != nil {
|
|
return fmt.Errorf("insert chat message failed: %w", err)
|
|
}
|
|
} else {
|
|
m.mu.Lock()
|
|
rec.ID = m.nextMessageID
|
|
m.nextMessageID++
|
|
m.messages[sessionID] = append(m.messages[sessionID], rec)
|
|
m.mu.Unlock()
|
|
}
|
|
|
|
// 若为用户输入，触发工作习惯偏好异步更新
|
|
if role == "user" && userID > 0 {
|
|
_ = m.ReflectUserPreference(ctx, userID, sessionID)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// RecordActivity 记录项目协同操作/决策/上传/下载活动
|
|
func (m *ContextManager) RecordActivity(ctx context.Context, projectID, userID int64, userName, actType, title string, detail map[string]any, weight float64, isMilestone bool) (*ActivityItem, error) {
|
|
if projectID <= 0 || title == "" {
|
|
return nil, errors.New("projectID and title required")
|
|
}
|
|
if weight <= 0 {
|
|
weight = 1.00
|
|
}
|
|
|
|
item := &ActivityItem{
|
|
ProjectID: projectID,
|
|
UserID: userID,
|
|
UserName: userName,
|
|
ActivityType: actType,
|
|
Title: title,
|
|
Detail: detail,
|
|
RawWeight: weight,
|
|
IsMilestone: isMilestone,
|
|
CreatedAt: time.Now(),
|
|
}
|
|
|
|
if m.pool != nil {
|
|
detailJSON, _ := json.Marshal(detail)
|
|
var actID int64
|
|
var createdAt time.Time
|
|
err := m.pool.QueryRow(ctx, `
|
|
INSERT INTO project_activities (project_id, user_id, user_name, activity_type, title, detail, weight, is_milestone, created_at)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, now())
|
|
RETURNING id, created_at
|
|
`, projectID, userID, userName, actType, title, detailJSON, weight, isMilestone).Scan(&actID, &createdAt)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("insert activity failed: %w", err)
|
|
}
|
|
item.ID = actID
|
|
item.CreatedAt = createdAt
|
|
} else {
|
|
m.mu.Lock()
|
|
item.ID = m.nextActivityID
|
|
m.nextActivityID++
|
|
m.activities[projectID] = append(m.activities[projectID], *item)
|
|
m.mu.Unlock()
|
|
}
|
|
|
|
item.DecayedWeight = CalculateDecayedWeight(item.RawWeight, item.CreatedAt, item.IsMilestone, m.decayCfg)
|
|
return item, nil
|
|
}
|
|
|
|
// GetUserContext 装配单用户私有上下文 (包含个人偏好画像、近期会话、个人操作与经时间衰减的文件资料)
|
|
func (m *ContextManager) GetUserContext(ctx context.Context, projectID, userID int64, sessionID string) (*SingleUserContext, error) {
|
|
pref, _ := m.GetUserPreference(ctx, userID)
|
|
|
|
userCtx := &SingleUserContext{
|
|
UserID: userID,
|
|
Preference: pref,
|
|
SessionID: sessionID,
|
|
RecentChats: make([]ChatMessageRecord, 0),
|
|
MyActivities: make([]ActivityItem, 0),
|
|
MyFiles: make([]*storage.FileReference, 0),
|
|
AssembledAt: time.Now(),
|
|
}
|
|
|
|
// 1. 获取近期会话
|
|
if sessionID != "" {
|
|
if m.pool != nil {
|
|
rows, err := m.pool.Query(ctx, `
|
|
SELECT id, session_id, user_id, role, content, intent, branch, metadata, created_at
|
|
FROM chat_messages
|
|
WHERE session_id = $1
|
|
ORDER BY created_at ASC LIMIT 30
|
|
`, sessionID)
|
|
if err == nil {
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var r ChatMessageRecord
|
|
var metaJSON []byte
|
|
if err := rows.Scan(&r.ID, &r.SessionID, &r.UserID, &r.Role, &r.Content, &r.Intent, &r.Branch, &metaJSON, &r.CreatedAt); err == nil {
|
|
_ = json.Unmarshal(metaJSON, &r.Metadata)
|
|
userCtx.RecentChats = append(userCtx.RecentChats, r)
|
|
}
|
|
}
|
|
}
|
|
} else {
|
|
m.mu.RLock()
|
|
userCtx.RecentChats = append(userCtx.RecentChats, m.messages[sessionID]...)
|
|
m.mu.RUnlock()
|
|
}
|
|
}
|
|
|
|
// 2. 获取用户个人在项目内的活动并应用时间衰减
|
|
var rawActs []ActivityItem
|
|
if m.pool != nil {
|
|
rows, err := m.pool.Query(ctx, `
|
|
SELECT id, project_id, user_id, user_name, activity_type, title, detail, weight, is_milestone, created_at
|
|
FROM project_activities
|
|
WHERE project_id = $1 AND user_id = $2
|
|
ORDER BY created_at DESC LIMIT 50
|
|
`, projectID, userID)
|
|
if err == nil {
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var it ActivityItem
|
|
var detJSON []byte
|
|
if err := rows.Scan(&it.ID, &it.ProjectID, &it.UserID, &it.UserName, &it.ActivityType, &it.Title, &detJSON, &it.RawWeight, &it.IsMilestone, &it.CreatedAt); err == nil {
|
|
_ = json.Unmarshal(detJSON, &it.Detail)
|
|
rawActs = append(rawActs, it)
|
|
}
|
|
}
|
|
}
|
|
} else {
|
|
m.mu.RLock()
|
|
for _, it := range m.activities[projectID] {
|
|
if it.UserID == userID {
|
|
rawActs = append(rawActs, it)
|
|
}
|
|
}
|
|
m.mu.RUnlock()
|
|
}
|
|
|
|
userCtx.MyActivities = ApplyTimeDecayToActivities(rawActs, m.decayCfg)
|
|
return userCtx, nil
|
|
}
|
|
|
|
// GetProjectContext 装配项目级多方协同上下文 (黑板事实、里程碑决策、团队活动流与共享文件)
|
|
func (m *ContextManager) GetProjectContext(ctx context.Context, projectID int64) (*ProjectCollectiveContext, error) {
|
|
pCtx := &ProjectCollectiveContext{
|
|
ProjectID: projectID,
|
|
BlackboardFacts: make(map[string]string),
|
|
AllActivities: make([]ActivityItem, 0),
|
|
MilestoneDecisions: make([]ActivityItem, 0),
|
|
SharedFiles: make([]*storage.FileReference, 0),
|
|
ParticipatingUsers: make([]UserSummary, 0),
|
|
AssembledAt: time.Now(),
|
|
}
|
|
|
|
if m.pool != nil {
|
|
// 1. 获取项目基准与黑板事实
|
|
var pCode, pName, cStage string
|
|
_ = m.pool.QueryRow(ctx, "SELECT code, name, current_stage FROM projects WHERE id = $1", projectID).Scan(&pCode, &pName, &cStage)
|
|
pCtx.ProjectCode = pCode
|
|
pCtx.ProjectName = pName
|
|
pCtx.CurrentStage = cStage
|
|
|
|
var factsJSON []byte
|
|
err := m.pool.QueryRow(ctx, "SELECT facts FROM project_blackboards WHERE project_id = $1", projectID).Scan(&factsJSON)
|
|
if err == nil {
|
|
_ = json.Unmarshal(factsJSON, &pCtx.BlackboardFacts)
|
|
}
|
|
|
|
// 2. 获取全项目活动
|
|
rows, err := m.pool.Query(ctx, `
|
|
SELECT id, project_id, user_id, user_name, activity_type, title, detail, weight, is_milestone, created_at
|
|
FROM project_activities
|
|
WHERE project_id = $1
|
|
ORDER BY created_at DESC LIMIT 100
|
|
`, projectID)
|
|
if err == nil {
|
|
defer rows.Close()
|
|
var rawActs []ActivityItem
|
|
for rows.Next() {
|
|
var it ActivityItem
|
|
var detJSON []byte
|
|
if err := rows.Scan(&it.ID, &it.ProjectID, &it.UserID, &it.UserName, &it.ActivityType, &it.Title, &detJSON, &it.RawWeight, &it.IsMilestone, &it.CreatedAt); err == nil {
|
|
_ = json.Unmarshal(detJSON, &it.Detail)
|
|
rawActs = append(rawActs, it)
|
|
if it.IsMilestone || it.ActivityType == "decision" {
|
|
pCtx.MilestoneDecisions = append(pCtx.MilestoneDecisions, it)
|
|
}
|
|
}
|
|
}
|
|
pCtx.AllActivities = ApplyTimeDecayToActivities(rawActs, m.decayCfg)
|
|
}
|
|
} else {
|
|
m.mu.RLock()
|
|
if existing, ok := m.projects[projectID]; ok {
|
|
pCtx.ProjectName = existing.ProjectName
|
|
pCtx.CurrentStage = existing.CurrentStage
|
|
for k, v := range existing.BlackboardFacts {
|
|
pCtx.BlackboardFacts[k] = v
|
|
}
|
|
}
|
|
rawActs := m.activities[projectID]
|
|
for _, it := range rawActs {
|
|
if it.IsMilestone || it.ActivityType == "decision" {
|
|
pCtx.MilestoneDecisions = append(pCtx.MilestoneDecisions, it)
|
|
}
|
|
}
|
|
pCtx.AllActivities = ApplyTimeDecayToActivities(rawActs, m.decayCfg)
|
|
m.mu.RUnlock()
|
|
}
|
|
|
|
return pCtx, nil
|
|
}
|
|
|
|
// BuildBlendedSquadContext 混合上下文装配引擎 (Context Blender)
|
|
// 将项目事实、里程碑决策、触发用户的工作习惯偏好合成最适合目标 Agent 角色执行的混合上下文
|
|
func (m *ContextManager) BuildBlendedSquadContext(ctx context.Context, projectID, triggerUserID int64, taskKey, role string) (*BlendedSquadContext, error) {
|
|
projCtx, err := m.GetProjectContext(ctx, projectID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("get project context failed: %w", err)
|
|
}
|
|
|
|
userPref, _ := m.GetUserPreference(ctx, triggerUserID)
|
|
guidance := m.reflector.GeneratePromptGuidance(userPref)
|
|
|
|
var decisions []string
|
|
for _, d := range projCtx.MilestoneDecisions {
|
|
decisions = append(decisions, fmt.Sprintf("[%s] %s (%s)", d.ActivityType, d.Title, d.UserName))
|
|
}
|
|
|
|
return &BlendedSquadContext{
|
|
ProjectName: projCtx.ProjectName,
|
|
CurrentStage: projCtx.CurrentStage,
|
|
TaskKey: taskKey,
|
|
AgentRole: role,
|
|
BlackboardFacts: projCtx.BlackboardFacts,
|
|
KeyDecisions: decisions,
|
|
TriggerUserPreference: userPref,
|
|
PersonalizedGuidance: guidance,
|
|
AssembledAt: time.Now(),
|
|
}, nil
|
|
}
|
|
|
|
// GetUserPreference 获取用户偏好画像
|
|
func (m *ContextManager) GetUserPreference(ctx context.Context, userID int64) (*UserPreference, error) {
|
|
if userID <= 0 {
|
|
return nil, errors.New("invalid userID")
|
|
}
|
|
|
|
if m.pool != nil {
|
|
var pref UserPreference
|
|
var focusArr []string
|
|
var factsJSON []byte
|
|
err := m.pool.QueryRow(ctx, `
|
|
SELECT u.id, u.account_id, u.name, u.role, p.communication_style, p.focus_areas, p.output_format_pref, p.custom_instructions, p.memory_facts, p.updated_at
|
|
FROM users u
|
|
LEFT JOIN user_preferences p ON u.id = p.user_id
|
|
WHERE u.id = $1
|
|
`, userID).Scan(&pref.UserID, &pref.AccountID, &pref.Name, &pref.Role, &pref.CommunicationStyle, &focusArr, &pref.OutputFormatPref, &pref.CustomInstructions, &factsJSON, &pref.UpdatedAt)
|
|
|
|
if err == nil {
|
|
pref.FocusAreas = focusArr
|
|
_ = json.Unmarshal(factsJSON, &pref.MemoryFacts)
|
|
if pref.CommunicationStyle == "" {
|
|
pref.CommunicationStyle = StyleConciseStructured
|
|
}
|
|
return &pref, nil
|
|
}
|
|
}
|
|
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
if pref, ok := m.users[userID]; ok {
|
|
return pref, nil
|
|
}
|
|
|
|
return &UserPreference{
|
|
UserID: userID,
|
|
CommunicationStyle: StyleConciseStructured,
|
|
FocusAreas: []string{"reg_compliance", "timeline"},
|
|
OutputFormatPref: "standard_report",
|
|
MemoryFacts: make(map[string]string),
|
|
UpdatedAt: time.Now(),
|
|
}, nil
|
|
}
|
|
|
|
// UpdateUserPreference 保存用户偏好画像
|
|
func (m *ContextManager) UpdateUserPreference(ctx context.Context, pref *UserPreference) error {
|
|
if pref == nil || pref.UserID <= 0 {
|
|
return errors.New("invalid user preference")
|
|
}
|
|
|
|
pref.UpdatedAt = time.Now()
|
|
|
|
if m.pool != nil {
|
|
factsJSON, _ := json.Marshal(pref.MemoryFacts)
|
|
_, err := m.pool.Exec(ctx, `
|
|
INSERT INTO user_preferences (user_id, communication_style, focus_areas, output_format_pref, custom_instructions, memory_facts, updated_at)
|
|
VALUES ($1, $2, $3, $4, $5, $6, now())
|
|
ON CONFLICT (user_id) DO UPDATE SET
|
|
communication_style = EXCLUDED.communication_style,
|
|
focus_areas = EXCLUDED.focus_areas,
|
|
output_format_pref = EXCLUDED.output_format_pref,
|
|
custom_instructions = EXCLUDED.custom_instructions,
|
|
memory_facts = EXCLUDED.memory_facts,
|
|
updated_at = now()
|
|
`, pref.UserID, string(pref.CommunicationStyle), pref.FocusAreas, pref.OutputFormatPref, pref.CustomInstructions, factsJSON)
|
|
if err != nil {
|
|
return fmt.Errorf("upsert user preference failed: %w", err)
|
|
}
|
|
}
|
|
|
|
m.mu.Lock()
|
|
m.users[pref.UserID] = pref
|
|
m.mu.Unlock()
|
|
return nil
|
|
}
|
|
|
|
// ReflectUserPreference 根据用户在指定会话的历史记录智能总结工作习惯与偏好
|
|
func (m *ContextManager) ReflectUserPreference(ctx context.Context, userID int64, sessionID string) error {
|
|
var msgs []ChatMessageRecord
|
|
if m.pool != nil {
|
|
rows, err := m.pool.Query(ctx, `
|
|
SELECT id, session_id, user_id, role, content, intent, branch, metadata, created_at
|
|
FROM chat_messages
|
|
WHERE user_id = $1 OR session_id = $2
|
|
ORDER BY created_at DESC LIMIT 20
|
|
`, userID, sessionID)
|
|
if err == nil {
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var r ChatMessageRecord
|
|
var metaJSON []byte
|
|
if err := rows.Scan(&r.ID, &r.SessionID, &r.UserID, &r.Role, &r.Content, &r.Intent, &r.Branch, &metaJSON, &r.CreatedAt); err == nil {
|
|
_ = json.Unmarshal(metaJSON, &r.Metadata)
|
|
msgs = append(msgs, r)
|
|
}
|
|
}
|
|
}
|
|
} else {
|
|
m.mu.RLock()
|
|
msgs = append(msgs, m.messages[sessionID]...)
|
|
m.mu.RUnlock()
|
|
}
|
|
|
|
if len(msgs) == 0 {
|
|
return nil
|
|
}
|
|
|
|
currentPref, _ := m.GetUserPreference(ctx, userID)
|
|
updatedPref := m.reflector.ReflectFromMessages(msgs, currentPref)
|
|
return m.UpdateUserPreference(ctx, updatedPref)
|
|
}
|