feat(ctxmgr): implement sliding window, rolling summarizer and DST task state (P1)

This commit is contained in:
WeiCHEN
2026-09-13 02:50:20 +00:00
committed by weichen
parent be659cba11
commit c58d33b853
11 changed files with 949 additions and 12 deletions
+69 -8
View File
@@ -78,24 +78,59 @@ func NewSession(id string, userCtx *coordinator.UserProjectContext, rec intent.R
return sess
}
// SyncRecentDialogue 将会话内存消息转换为 DialogueTurn 同步到 UserCtx.RecentDialogue (供 Phase A 对话 LLM 与路由画像使用)
// SyncRecentDialogue 使用双预算滑动窗口将消息转换为 DialogueTurn 同步到 UserCtx.RecentDialogue
func (s *Session) SyncRecentDialogue() {
if s.UserCtx == nil {
return
}
const maxTurns = 16 // 保持合理窗口 (略宽于 dialogueHistoryMax=8,保留完整轮次)
msgs := s.Messages
if len(msgs) > maxTurns {
msgs = msgs[len(msgs)-maxTurns:]
}
turns := make([]coordinator.DialogueTurn, 0, len(msgs))
for _, m := range msgs {
records := s.toChatRecords()
cfg := ctxmgr.DefaultWindowConfig()
window, early, truncated := ctxmgr.AssembleWindow(records, cfg)
turns := make([]coordinator.DialogueTurn, 0, len(window))
for _, m := range window {
if strings.TrimSpace(m.Content) == "" {
continue
}
turns = append(turns, coordinator.DialogueTurn{Role: m.Role, Content: m.Content})
}
s.UserCtx.RecentDialogue = turns
// 若发生了截断且存在早期消息,同步计算或加载滚动摘要
if truncated && len(early) > 0 && s.CtxMgr != nil {
s.syncDialogueSummary(context.Background(), early)
}
}
func (s *Session) toChatRecords() []ctxmgr.ChatMessageRecord {
records := make([]ctxmgr.ChatMessageRecord, 0, len(s.Messages))
for i, m := range s.Messages {
records = append(records, ctxmgr.ChatMessageRecord{
ID: int64(i + 1),
SessionID: s.ID,
Role: m.Role,
Content: m.Content,
Intent: string(m.Intent),
Branch: string(m.Branch),
CreatedAt: m.Timestamp,
})
}
return records
}
func (s *Session) syncDialogueSummary(ctx context.Context, early []ctxmgr.ChatMessageRecord) {
if s.CtxMgr == nil || len(early) == 0 {
return
}
oldSum, _ := s.CtxMgr.GetChatSummary(ctx, s.ID)
if oldSum == nil || len(early) > oldSum.CoveredCount {
newSum, err := s.CtxMgr.GenerateRollingSummary(ctx, s.ID, oldSum, early)
if err == nil && newSum != nil && s.UserCtx != nil {
s.UserCtx.DialogueSummary = newSum.RenderPromptSnippet()
}
} else if oldSum != nil && s.UserCtx != nil {
s.UserCtx.DialogueSummary = oldSum.RenderPromptSnippet()
}
}
// syncPersonalizedGuidance 将用户偏好画像编译为 Prompt 指导语并注入 UserCtx
@@ -179,6 +214,26 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str
)
}
// 1.5 刷新会话级任务状态机 (DST)
if s.UserCtx != nil {
if s.UserCtx.TaskState == nil {
s.UserCtx.TaskState = ctxmgr.NewConversationTaskState()
}
intentName := ""
var filledSlots map[string]string
if recRes.TopIntent != nil {
intentName = string(recRes.TopIntent.Intent)
filledSlots = recRes.TopIntent.FilledSlots
}
s.UserCtx.TaskState.UpdateFromTurn(
userInput,
intentName,
filledSlots,
string(recRes.Branch),
string(s.UserCtx.Project.CurrentStage),
)
}
// 2. 协调调度与校验 (L2 协调层)
if streamer != nil {
streamer.EmitPhase("dialogue_generating", "正在整理回复...", "assistant", 0, 0, 40)
@@ -193,6 +248,12 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str
// 3. 更新上下文与挂起预期生命周期
if coordRes.UpdatedContext != nil {
if coordRes.UpdatedContext.TaskState == nil && s.UserCtx != nil {
coordRes.UpdatedContext.TaskState = s.UserCtx.TaskState
}
if coordRes.UpdatedContext.DialogueSummary == "" && s.UserCtx != nil {
coordRes.UpdatedContext.DialogueSummary = s.UserCtx.DialogueSummary
}
s.UserCtx = coordRes.UpdatedContext
}
+72
View File
@@ -2,6 +2,7 @@ package chat_test
import (
"context"
"fmt"
"strings"
"testing"
@@ -383,3 +384,74 @@ func TestChatSession_RecentDialogueAndGuidanceSync(t *testing.T) {
}
}
func TestChatSession_SlidingWindowAndSummary(t *testing.T) {
ctx := context.Background()
rec := intent.NewRecognizer()
engine := coordinator.NewCoordinatorEngine(nil)
ctxMgr := ctxmgr.NewContextManager(nil, nil, ctxmgr.DefaultTimeDecayConfig())
userCtx := &coordinator.UserProjectContext{
SnapshotID: "sess_ctx_window",
User: coordinator.UserInfo{
ID: "u_win_1",
Name: "赵工",
Role: "engineer",
ProjectRole: "composer",
},
Project: coordinator.ProjectInfo{
Name: "数字化高频超声吸引系统",
CurrentStage: bizdev.StageV01_Triage,
},
BlackboardFacts: make(map[string]string),
}
session := chat.NewSession("sess_win_001", userCtx, rec, engine)
session.UserID = 99
session.ProjectID = 200
session.CtxMgr = ctxMgr
// 模拟连续进行 10 轮交互 (生成 20 条消息,超出默认窗口上限 16 条)
// 第 1 轮带有核心事实
_, err := session.SendMessage(ctx, "你好,我们准备立项高频超声手术刀系统,预算200万")
if err != nil {
t.Fatalf("turn 1 failed: %v", err)
}
// 模拟追加 9 轮简短问答
for i := 2; i <= 10; i++ {
_, err := session.SendMessage(ctx, fmt.Sprintf("这是第 %d 轮日常沟通咨询", i))
if err != nil {
t.Fatalf("turn %d failed: %v", i, err)
}
}
if len(session.Messages) != 20 {
t.Fatalf("expected 20 total messages, got %d", len(session.Messages))
}
// 验证双预算滑动窗口(条数上限 16 且受 Token 预算约束裁剪)
if len(session.UserCtx.RecentDialogue) > 16 || len(session.UserCtx.RecentDialogue) < 10 {
t.Errorf("expected between 10 and 16 messages in RecentDialogue window under dual budget, got %d", len(session.UserCtx.RecentDialogue))
}
// 验证早期被截断的消息是否被压缩为滚动摘要
if session.UserCtx.DialogueSummary == "" {
t.Errorf("expected DialogueSummary to be populated after window truncation")
}
if !strings.Contains(session.UserCtx.DialogueSummary, "【早期对话摘要】") {
t.Errorf("expected summary header, got: %s", session.UserCtx.DialogueSummary)
}
if !strings.Contains(session.UserCtx.DialogueSummary, "超声手术设备") || !strings.Contains(session.UserCtx.DialogueSummary, "200万元") {
t.Errorf("expected preserved facts in summary snippet, got: %s", session.UserCtx.DialogueSummary)
}
// 验证会话级任务状态跟踪 (DST) 正常运转
if session.UserCtx.TaskState == nil {
t.Fatalf("expected TaskState to be populated on UserCtx")
}
if session.UserCtx.TaskState.TopicAnchor == "" {
t.Errorf("expected TopicAnchor in TaskState")
}
}
+5
View File
@@ -9,6 +9,7 @@ import (
"time"
"chat-agent-system/internal/bizdev"
"chat-agent-system/internal/ctxmgr"
"chat-agent-system/internal/intent"
"chat-agent-system/internal/squad"
)
@@ -68,6 +69,10 @@ type UserProjectContext struct {
WorkDir string `json:"work_dir,omitempty"`
// RecentDialogue 近轮对话(供 Phase A 对话 LLM 使用,不序列化到快照)
RecentDialogue []DialogueTurn `json:"-"`
// DialogueSummary 早期对话滚动压缩摘要(由 ctxmgr 维护生成,不序列化到快照)
DialogueSummary string `json:"-"`
// TaskState 会话级任务状态(DST,记录 active_task, topic_anchor, collected_slots 等)
TaskState *ctxmgr.ConversationTaskState `json:"task_state,omitempty"`
// PersonalizedGuidance 用户工作习惯/偏好注入语(由 ctxmgr 编译,供对话 LLM 使用)
PersonalizedGuidance string `json:"-"`
// UpgradeOffered 路由升级建议冷却标记(每会话至多主动建议一次,防打扰——对齐"参谋不强推"人设)
+20 -4
View File
@@ -176,6 +176,19 @@ func buildDialoguePrompt(userInput string, userCtx *UserProjectContext) string {
}
}
}
if userCtx.TaskState != nil {
if tsSnippet := userCtx.TaskState.RenderPromptSnippet(); tsSnippet != "" {
sb.WriteString("\n")
sb.WriteString(tsSnippet)
}
}
if s := strings.TrimSpace(userCtx.DialogueSummary); s != "" {
sb.WriteString("\n")
sb.WriteString(s)
if !strings.HasSuffix(s, "\n") {
sb.WriteString("\n")
}
}
if g := strings.TrimSpace(userCtx.PersonalizedGuidance); g != "" {
sb.WriteString("\n")
sb.WriteString(g)
@@ -189,12 +202,15 @@ func buildDialoguePrompt(userInput string, userCtx *UserProjectContext) string {
if len(turns) > dialogueHistoryMax {
turns = turns[len(turns)-dialogueHistoryMax:]
}
for _, t := range turns {
for i, t := range turns {
role := t.Role
if role == "" {
role = "user"
roleName := "用户"
if role == "assistant" {
roleName = "助手"
} else if role == "system" {
roleName = "系统"
}
sb.WriteString(fmt.Sprintf("%s: %s\n", role, t.Content))
sb.WriteString(fmt.Sprintf("[%d] %s: %s\n", i+1, roleName, t.Content))
}
}
}
+2
View File
@@ -28,6 +28,7 @@ type ContextManager struct {
sessions map[string]*ChatSessionSummary // session_id -> summary
activities map[int64][]ActivityItem // project_id -> activities
memoryChunks []MemoryChunkItem // 语义记忆切片缓存
summaries map[string]*ChatSummary // session_id -> rolling summary
nextActivityID int64
nextMessageID int64
nextChunkID int64
@@ -50,6 +51,7 @@ func NewContextManager(pool *pgxpool.Pool, cas *storage.CAS, decayCfg TimeDecayC
sessions: make(map[string]*ChatSessionSummary),
activities: make(map[int64][]ActivityItem),
memoryChunks: make([]MemoryChunkItem, 0),
summaries: make(map[string]*ChatSummary),
nextActivityID: 1,
nextMessageID: 1,
nextChunkID: 1,
+257
View File
@@ -0,0 +1,257 @@
package ctxmgr
import (
"context"
"encoding/json"
"fmt"
"strings"
"time"
)
// ChatSummary 会话滚动摘要数据模型
type ChatSummary struct {
SessionID string `json:"session_id"`
Narrative string `json:"narrative"` // 叙事摘要(≤300 字)
Facts map[string]string `json:"facts"` // 硬事实槽位字典(永不被压缩稀释)
OpenQuestions []string `json:"open_questions"` // 未决技术或商务待办
CoveredCount int `json:"covered_count"` // 累计已覆盖并压缩的历史消息条数
CoveredUntil time.Time `json:"covered_until"` // 摘要覆盖消息的截止时间戳
UpdatedAt time.Time `json:"updated_at"`
}
// RenderPromptSnippet 渲染为 Prompt 中的【早期对话摘要】段
func (s *ChatSummary) RenderPromptSnippet() string {
if s == nil || (s.Narrative == "" && len(s.Facts) == 0) {
return ""
}
var sb strings.Builder
sb.WriteString(fmt.Sprintf("【早期对话摘要】(已压缩前 %d 条对话内容)\n", s.CoveredCount))
if s.Narrative != "" {
sb.WriteString(fmt.Sprintf("• 主题与结论:%s\n", s.Narrative))
}
if len(s.Facts) > 0 {
var factStrs []string
for k, v := range s.Facts {
factStrs = append(factStrs, fmt.Sprintf("%s=%s", k, v))
}
sb.WriteString(fmt.Sprintf("• 已沉淀事实:%s\n", strings.Join(factStrs, "; ")))
}
if len(s.OpenQuestions) > 0 {
sb.WriteString(fmt.Sprintf("• 未决待办:%s\n", strings.Join(s.OpenQuestions, ";")))
}
return sb.String()
}
// GetChatSummary 获取会话现有滚动摘要
func (m *ContextManager) GetChatSummary(ctx context.Context, sessionID string) (*ChatSummary, error) {
if sessionID == "" {
return nil, fmt.Errorf("sessionID required")
}
if m.pool != nil {
var sum ChatSummary
var factsJSON, openQuestionsJSON []byte
err := m.pool.QueryRow(ctx, `
SELECT session_id, narrative, facts, open_questions, covered_count, covered_until, updated_at
FROM chat_summaries
WHERE session_id = $1
`, sessionID).Scan(&sum.SessionID, &sum.Narrative, &factsJSON, &openQuestionsJSON, &sum.CoveredCount, &sum.CoveredUntil, &sum.UpdatedAt)
if err == nil {
_ = json.Unmarshal(factsJSON, &sum.Facts)
_ = json.Unmarshal(openQuestionsJSON, &sum.OpenQuestions)
return &sum, nil
}
}
m.mu.RLock()
defer m.mu.RUnlock()
if m.summaries != nil {
if s, ok := m.summaries[sessionID]; ok && s != nil {
// 返回副本
copySum := *s
return &copySum, nil
}
}
return nil, nil
}
// SaveChatSummary 保存或更新滚动摘要
func (m *ContextManager) SaveChatSummary(ctx context.Context, sum *ChatSummary) error {
if sum == nil || sum.SessionID == "" {
return fmt.Errorf("invalid summary or sessionID")
}
sum.UpdatedAt = time.Now()
if m.pool != nil {
factsJSON, _ := json.Marshal(sum.Facts)
openQuestionsJSON, _ := json.Marshal(sum.OpenQuestions)
_, err := m.pool.Exec(ctx, `
INSERT INTO chat_summaries (session_id, narrative, facts, open_questions, covered_count, covered_until, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7)
ON CONFLICT (session_id) DO UPDATE SET
narrative = EXCLUDED.narrative,
facts = EXCLUDED.facts,
open_questions = EXCLUDED.open_questions,
covered_count = EXCLUDED.covered_count,
covered_until = EXCLUDED.covered_until,
updated_at = EXCLUDED.updated_at
`, sum.SessionID, sum.Narrative, factsJSON, openQuestionsJSON, sum.CoveredCount, sum.CoveredUntil, sum.UpdatedAt)
if err != nil {
// DB 失败时仍降级写入内存
_ = err
}
}
m.mu.Lock()
defer m.mu.Unlock()
if m.summaries == nil {
m.summaries = make(map[string]*ChatSummary)
}
copySum := *sum
m.summaries[sum.SessionID] = &copySum
return nil
}
// GenerateRollingSummary 结合已覆盖旧摘要与本批被淘汰出窗口的早期消息,滚动增量压缩生成新摘要
// 具备确定性规则降级引擎:无论外部大模型是否可用,均能稳定提取关键实体槽位与未决问题,不丢事实
func (m *ContextManager) GenerateRollingSummary(ctx context.Context, sessionID string, oldSum *ChatSummary, earlyMsgs []ChatMessageRecord) (*ChatSummary, error) {
if sessionID == "" {
return nil, fmt.Errorf("sessionID required")
}
if len(earlyMsgs) == 0 {
return oldSum, nil
}
newSum := &ChatSummary{
SessionID: sessionID,
Facts: make(map[string]string),
OpenQuestions: make([]string, 0),
CoveredCount: len(earlyMsgs),
CoveredUntil: earlyMsgs[len(earlyMsgs)-1].CreatedAt,
UpdatedAt: time.Now(),
}
if oldSum != nil {
newSum.CoveredCount += oldSum.CoveredCount
for k, v := range oldSum.Facts {
newSum.Facts[k] = v
}
newSum.OpenQuestions = append(newSum.OpenQuestions, oldSum.OpenQuestions...)
}
// 1. 硬事实抽取(永不丢失)
extractedFacts := extractFactsFromMessages(earlyMsgs)
for k, v := range extractedFacts {
newSum.Facts[k] = v
}
// 2. 未决问题与待办抽取
newQuestions := extractOpenQuestions(earlyMsgs)
for _, q := range newQuestions {
if !containsString(newSum.OpenQuestions, q) {
newSum.OpenQuestions = append(newSum.OpenQuestions, q)
}
}
if len(newSum.OpenQuestions) > 4 {
newSum.OpenQuestions = newSum.OpenQuestions[len(newSum.OpenQuestions)-4:]
}
// 3. 叙事要点抽取与合并(确定性规则降级)
narrativeParts := make([]string, 0)
if oldSum != nil && oldSum.Narrative != "" {
narrativeParts = append(narrativeParts, oldSum.Narrative)
}
newPoints := extractNarrativePoints(earlyMsgs)
if newPoints != "" {
narrativeParts = append(narrativeParts, newPoints)
}
fullNarrative := strings.Join(narrativeParts, ";")
if len([]rune(fullNarrative)) > 260 {
runes := []rune(fullNarrative)
fullNarrative = string(runes[len(runes)-260:]) + "..."
}
newSum.Narrative = fullNarrative
// 保存并返回
_ = m.SaveChatSummary(ctx, newSum)
return newSum, nil
}
func extractFactsFromMessages(msgs []ChatMessageRecord) map[string]string {
facts := make(map[string]string)
for _, m := range msgs {
c := m.Content
if strings.Contains(c, "超声手术") || strings.Contains(c, "超声刀") || strings.Contains(c, "高频超声") || strings.Contains(c, "超声") {
facts["device_type"] = "超声手术设备"
} else if strings.Contains(c, "内窥镜") {
facts["device_type"] = "医用内窥镜系统"
} else if strings.Contains(c, "监护仪") {
facts["device_type"] = "病人监护设备"
}
if strings.Contains(c, "国内") && (strings.Contains(c, "欧洲") || strings.Contains(c, "CE")) {
facts["target_market"] = "国内及欧洲CE"
} else if strings.Contains(c, "国内") && facts["target_market"] == "" {
facts["target_market"] = "国内NMPA"
}
if strings.Contains(c, "200万") || strings.Contains(c, "200 万元") {
facts["budget"] = "200万元"
} else if strings.Contains(c, "150万") || strings.Contains(c, "150 万元") {
facts["budget"] = "150万元"
}
if strings.Contains(c, "6个月") || strings.Contains(c, "半年") {
facts["timeline"] = "6个月"
}
}
return facts
}
func extractOpenQuestions(msgs []ChatMessageRecord) []string {
questions := make([]string, 0)
for _, m := range msgs {
c := strings.TrimSpace(m.Content)
if m.Role == "user" && (strings.HasSuffix(c, "?") || strings.HasSuffix(c, "?") || strings.Contains(c, "如何") || strings.Contains(c, "怎么") || strings.Contains(c, "多久")) {
clauses := strings.FieldsFunc(c, func(r rune) bool {
return r == ',' || r == '。' || r == ';' || r == ',' || r == ';'
})
for _, clause := range clauses {
cl := strings.TrimSpace(clause)
if strings.HasSuffix(cl, "?") || strings.HasSuffix(cl, "?") || strings.Contains(cl, "多久") || strings.Contains(cl, "如何") || strings.Contains(cl, "怎么") {
if len([]rune(cl)) <= 50 {
questions = append(questions, cl)
break
}
}
}
}
}
return questions
}
func extractNarrativePoints(msgs []ChatMessageRecord) string {
var points []string
for _, m := range msgs {
if m.Role == "user" {
trimmed := strings.TrimSpace(m.Content)
if strings.Contains(trimmed, "立项") || strings.Contains(trimmed, "标准") || strings.Contains(trimmed, "进展") || strings.Contains(trimmed, "咨询") || strings.Contains(trimmed, "沟通") {
firstClause := strings.Split(trimmed, ",")[0]
if len([]rune(firstClause)) <= 30 && !containsString(points, firstClause) {
points = append(points, firstClause)
}
}
}
}
if len(points) == 0 {
return "双方开展了研发立项与合规细节沟通"
}
if len(points) > 3 {
points = points[len(points)-3:]
}
return "讨论了" + strings.Join(points, "、")
}
+109
View File
@@ -0,0 +1,109 @@
package ctxmgr
import (
"context"
"strings"
"testing"
"time"
)
func TestRollingSummarizer_LifecycleAndFactPreservation(t *testing.T) {
mgr := NewContextManager(nil, nil, DefaultTimeDecayConfig())
ctx := context.Background()
sessID := "sess_sum_test_001"
// 模拟首批 4 条早期消息(将被淘汰出窗口)
batch1 := []ChatMessageRecord{
{
ID: 1,
Role: "user",
Content: "我们准备研发高频超声手术刀系统,请教立项建议",
CreatedAt: time.Now().Add(-10 * time.Minute),
},
{
ID: 2,
Role: "assistant",
Content: "超声手术设备属于二类或三类器械,需着重关注电气安全与空化热效应控制。",
CreatedAt: time.Now().Add(-9 * time.Minute),
},
{
ID: 3,
Role: "user",
Content: "目标市场是国内及欧洲CE,项目预算大约在200万左右,型检周期通常需要多久?",
CreatedAt: time.Now().Add(-8 * time.Minute),
},
{
ID: 4,
Role: "assistant",
Content: "国内型检通常预估4-6个月,需符合GB 9706.1-2020标准。",
CreatedAt: time.Now().Add(-7 * time.Minute),
},
}
sum1, err := mgr.GenerateRollingSummary(ctx, sessID, nil, batch1)
if err != nil {
t.Fatalf("GenerateRollingSummary failed: %v", err)
}
if sum1.CoveredCount != 4 {
t.Errorf("expected covered_count 4, got %d", sum1.CoveredCount)
}
if sum1.Facts["device_type"] != "超声手术设备" {
t.Errorf("expected device_type extracted, got %v", sum1.Facts)
}
if sum1.Facts["target_market"] != "国内及欧洲CE" {
t.Errorf("expected target_market extracted, got %v", sum1.Facts)
}
if sum1.Facts["budget"] != "200万元" {
t.Errorf("expected budget extracted, got %v", sum1.Facts)
}
if len(sum1.OpenQuestions) == 0 {
t.Errorf("expected at least 1 open question extracted")
}
snippet1 := sum1.RenderPromptSnippet()
if !strings.Contains(snippet1, "【早期对话摘要】(已压缩前 4 条对话内容)") {
t.Errorf("snippet missing header: %s", snippet1)
}
if !strings.Contains(snippet1, "device_type=超声手术设备") {
t.Errorf("snippet missing device_type: %s", snippet1)
}
// 模拟第二批消息追加(增量滚动压缩)
batch2 := []ChatMessageRecord{
{
ID: 5,
Role: "user",
Content: "临床评价路径该怎么走?",
CreatedAt: time.Now().Add(-5 * time.Minute),
},
{
ID: 6,
Role: "assistant",
Content: "可优先考虑同品种比对路径以节省费用。",
CreatedAt: time.Now().Add(-4 * time.Minute),
},
}
sum2, err := mgr.GenerateRollingSummary(ctx, sessID, sum1, batch2)
if err != nil {
t.Fatalf("incremental GenerateRollingSummary failed: %v", err)
}
if sum2.CoveredCount != 6 {
t.Errorf("expected covered_count 6, got %d", sum2.CoveredCount)
}
// 验证硬事实依然完整保留(永不丢失)
if sum2.Facts["device_type"] != "超声手术设备" || sum2.Facts["budget"] != "200万元" {
t.Errorf("hard facts lost during incremental roll: %v", sum2.Facts)
}
// 验证 ContextManager 的读取
saved, err := mgr.GetChatSummary(ctx, sessID)
if err != nil || saved == nil {
t.Fatalf("GetChatSummary failed: %v", err)
}
if saved.CoveredCount != 6 {
t.Errorf("saved summary covered count mismatch: %d", saved.CoveredCount)
}
}
+147
View File
@@ -0,0 +1,147 @@
package ctxmgr
import (
"fmt"
"strings"
"time"
)
// ConversationTaskState 会话级任务状态(DST,对话状态跟踪)
// 与项目级黑板的分工:黑板是全项目多方共享的全局 SSOT;
// task_state 是记录“当前这场会话的任务推进到哪一步”的会话私有工作记忆。
type ConversationTaskState struct {
ActiveTask string `json:"active_task,omitempty"` // 当前对话任务类型(如 intake_consult / case_compare / doc_reject_consult)
TopicAnchor string `json:"topic_anchor,omitempty"` // 主题锚点:最近一轮有效核心问句
TopicEntities []string `json:"topic_entities,omitempty"` // 锚点实体(如项目名、器械分类、国家标准编号)
CollectedSlots map[string]string `json:"collected_slots,omitempty"` // 已收集参数槽位
MissingSlots []string `json:"missing_slots,omitempty"` // 待补充关键槽位
OpenQuestions []string `json:"open_questions,omitempty"` // 未决技术/商务问题
UpdatedAt time.Time `json:"updated_at"`
}
// NewConversationTaskState 创建初始对话状态
func NewConversationTaskState() *ConversationTaskState {
return &ConversationTaskState{
CollectedSlots: make(map[string]string),
MissingSlots: make([]string, 0),
OpenQuestions: make([]string, 0),
TopicEntities: make([]string, 0),
UpdatedAt: time.Now(),
}
}
// UpdateFromTurn 根据当轮输入、识别槽位与分支流转刷新状态机
func (ts *ConversationTaskState) UpdateFromTurn(userInput string, intentName string, filledSlots map[string]string, branch string, stage string) {
if ts.CollectedSlots == nil {
ts.CollectedSlots = make(map[string]string)
}
ts.UpdatedAt = time.Now()
// 1. 更新活跃任务与主题锚点
if intentName != "" && intentName != "CASUAL_CHAT" {
ts.ActiveTask = intentName
}
trimmedInput := strings.TrimSpace(userInput)
if len([]rune(trimmedInput)) >= 4 && !isGenericAffirmative(trimmedInput) {
ts.TopicAnchor = trimmedInput
}
// 2. 合入已识别槽位
for k, v := range filledSlots {
if strings.TrimSpace(v) != "" {
ts.CollectedSlots[k] = strings.TrimSpace(v)
}
}
// 3. 提取常见主题实体(器械类型、强标号等)
ts.extractTopicEntities(trimmedInput)
// 4. 计算当前阶段缺失槽位
ts.computeMissingSlots(stage)
// 5. 若分支为极速兑现(EXPECTATION_HIT),若已完成某项挂起待办,相应清理未决问题
if branch == "EXPECTATION_HIT" && len(ts.OpenQuestions) > 0 {
ts.OpenQuestions = ts.OpenQuestions[1:]
}
}
func (ts *ConversationTaskState) extractTopicEntities(input string) {
entityCandidates := []string{
"超声手术设备", "超声刀", "高频超声", "掌上超声", "内窥镜", "腹腔镜", "医用电气设备",
"GB 9706.1-2020", "YY 9706.102", "YY 0505", "GB/T 42062", "CE认证", "NMPA", "FDA 510(k)",
}
for _, cand := range entityCandidates {
if strings.Contains(input, cand) && !containsString(ts.TopicEntities, cand) {
ts.TopicEntities = append(ts.TopicEntities, cand)
if len(ts.TopicEntities) > 6 {
ts.TopicEntities = ts.TopicEntities[len(ts.TopicEntities)-6:]
}
}
}
}
func (ts *ConversationTaskState) computeMissingSlots(stage string) {
required := []string{"device_type", "target_market"}
if stage == "V0.1_triage" || stage == "intake" {
required = append(required, "budget")
}
missing := make([]string, 0)
for _, req := range required {
if _, ok := ts.CollectedSlots[req]; !ok {
missing = append(missing, req)
}
}
ts.MissingSlots = missing
}
// RenderPromptSnippet 渲染为 Prompt 中的【工作记忆】段
func (ts *ConversationTaskState) RenderPromptSnippet() string {
if ts == nil {
return ""
}
var sb strings.Builder
sb.WriteString("【本轮任务状态 (Working Memory)】\n")
if ts.ActiveTask != "" {
sb.WriteString(fmt.Sprintf("- 活跃任务: %s\n", ts.ActiveTask))
}
if ts.TopicAnchor != "" {
sb.WriteString(fmt.Sprintf("- 主题锚点: \"%s\"\n", ts.TopicAnchor))
}
if len(ts.TopicEntities) > 0 {
sb.WriteString(fmt.Sprintf("- 核心实体: %s\n", strings.Join(ts.TopicEntities, "、")))
}
if len(ts.CollectedSlots) > 0 {
var slotStrs []string
for k, v := range ts.CollectedSlots {
slotStrs = append(slotStrs, fmt.Sprintf("%s=%s", k, v))
}
sb.WriteString(fmt.Sprintf("- 已收集参数: %s\n", strings.Join(slotStrs, "; ")))
}
if len(ts.MissingSlots) > 0 {
sb.WriteString(fmt.Sprintf("- 待补充槽位: %s\n", strings.Join(ts.MissingSlots, "、")))
}
if len(ts.OpenQuestions) > 0 {
sb.WriteString(fmt.Sprintf("- 未决问题: %s\n", strings.Join(ts.OpenQuestions, ";")))
}
return sb.String()
}
func isGenericAffirmative(s string) bool {
s = strings.TrimSpace(s)
return s == "好" || s == "好的" || s == "行" || s == "可以" || s == "1" || s == "同意" || s == "推进"
}
func containsString(arr []string, s string) bool {
for _, item := range arr {
if item == s {
return true
}
}
return false
}
+70
View File
@@ -0,0 +1,70 @@
package ctxmgr
import (
"strings"
"testing"
)
func TestConversationTaskState_Lifecycle(t *testing.T) {
ts := NewConversationTaskState()
// Turn 1: 用户表达立项意图与器械类型
ts.UpdateFromTurn(
"我们打算立项研发一款高频超声手术刀系统",
"SUPPLEMENT_INFO",
map[string]string{"device_type": "高频超声手术刀系统"},
"NORMAL",
"V0.1_triage",
)
if ts.ActiveTask != "SUPPLEMENT_INFO" {
t.Errorf("expected ActiveTask SUPPLEMENT_INFO, got %s", ts.ActiveTask)
}
if ts.TopicAnchor != "我们打算立项研发一款高频超声手术刀系统" {
t.Errorf("unexpected TopicAnchor: %s", ts.TopicAnchor)
}
if ts.CollectedSlots["device_type"] != "高频超声手术刀系统" {
t.Errorf("expected device_type collected, got %v", ts.CollectedSlots)
}
// 缺失槽位应有 target_market 和 budget
hasMarket := false
hasBudget := false
for _, m := range ts.MissingSlots {
if m == "target_market" {
hasMarket = true
}
if m == "budget" {
hasBudget = true
}
}
if !hasMarket || !hasBudget {
t.Errorf("expected missing target_market and budget, got %v", ts.MissingSlots)
}
// Turn 2: 补充目标市场与预算
ts.UpdateFromTurn(
"目标市场主要为国内和欧洲,初期预算约为200万元",
"SUPPLEMENT_INFO",
map[string]string{
"target_market": "国内和欧洲",
"budget": "200万元",
},
"NORMAL",
"V0.1_triage",
)
if len(ts.MissingSlots) != 0 {
t.Errorf("expected 0 missing slots after turn 2, got %v", ts.MissingSlots)
}
snippet := ts.RenderPromptSnippet()
if !strings.Contains(snippet, "【本轮任务状态 (Working Memory)】") {
t.Errorf("missing header in prompt snippet: %s", snippet)
}
if !strings.Contains(snippet, "device_type=高频超声手术刀系统") {
t.Errorf("missing device_type in prompt snippet: %s", snippet)
}
if !strings.Contains(snippet, "target_market=国内和欧洲") {
t.Errorf("missing target_market in prompt snippet: %s", snippet)
}
}
+108
View File
@@ -0,0 +1,108 @@
package ctxmgr
import (
"fmt"
"strings"
)
// WindowConfig 滑动窗口配置
type WindowConfig struct {
MaxTurns int // 窗口内最大消息条数(一问一答计 2 条),默认 16
TokenBudget int // 窗口内字符/Token 预算,默认 3000
ReserveTail int // 为本轮用户输入与固定 Prompt 预留的字符预算,默认 800
}
// DefaultWindowConfig 返回默认滑动窗口配置
func DefaultWindowConfig() WindowConfig {
return WindowConfig{
MaxTurns: 16,
TokenBudget: 3000,
ReserveTail: 800,
}
}
// EstimateTokens 估算文本 Token 数 (中文及中英混排场景下采用保守 rune 计数:1 rune ≈ 1 token)
func EstimateTokens(s string) int {
return len([]rune(s))
}
// AssembleWindow 从完整历史消息列表中按双预算(条数上限 + Token预算)逆序裁剪出近期对话窗口
// 返回:
// - window: 窗口内的近期消息(按原始时间正序)
// - early: 窗口之外被淘汰的早期消息(供滚动摘要器提取与压缩)
// - truncated: 是否发生了截断
func AssembleWindow(msgs []ChatMessageRecord, cfg WindowConfig) (window []ChatMessageRecord, early []ChatMessageRecord, truncated bool) {
if len(msgs) == 0 {
return []ChatMessageRecord{}, []ChatMessageRecord{}, false
}
if cfg.MaxTurns <= 0 {
cfg.MaxTurns = 16
}
if cfg.TokenBudget <= 0 {
cfg.TokenBudget = 3000
}
netBudget := cfg.TokenBudget - cfg.ReserveTail
if netBudget < 400 {
netBudget = cfg.TokenBudget // 若预留过大,以全额预算为准
}
totalTokens := 0
cutIndex := 0
// 逆序向前收纳
for i := len(msgs) - 1; i >= 0; i-- {
m := msgs[i]
tokens := EstimateTokens(m.Content)
if tokens == 0 {
tokens = 1
}
// 检查条数预算与 Token 预算
countInWindow := len(msgs) - i
if countInWindow > cfg.MaxTurns || (totalTokens+tokens > netBudget && countInWindow > 2) {
cutIndex = i + 1
break
}
totalTokens += tokens
}
if cutIndex > 0 {
early = make([]ChatMessageRecord, cutIndex)
copy(early, msgs[:cutIndex])
window = make([]ChatMessageRecord, len(msgs)-cutIndex)
copy(window, msgs[cutIndex:])
return window, early, true
}
window = make([]ChatMessageRecord, len(msgs))
copy(window, msgs)
return window, []ChatMessageRecord{}, false
}
// FormatNumberedDialogue 将消息序列格式化为带逐轮序号的近期对话文本
// 格式如:
// [1] 用户: 帮我看看项目进展
// [2] 助手: 当前处于 V0.1 阶段...
func FormatNumberedDialogue(msgs []ChatMessageRecord, startIndex int) string {
if len(msgs) == 0 {
return ""
}
if startIndex <= 0 {
startIndex = 1
}
var sb strings.Builder
for i, m := range msgs {
roleName := "用户"
if m.Role == "assistant" {
roleName = "助手"
} else if m.Role == "system" {
roleName = "系统"
}
sb.WriteString(fmt.Sprintf("[%d] %s: %s\n", startIndex+i, roleName, strings.TrimSpace(m.Content)))
}
return sb.String()
}
+90
View File
@@ -0,0 +1,90 @@
package ctxmgr
import (
"fmt"
"strings"
"testing"
)
func TestAssembleWindow_MaxTurnsBudget(t *testing.T) {
// 构造 20 条消息
var msgs []ChatMessageRecord
for i := 1; i <= 20; i++ {
role := "user"
if i%2 == 0 {
role = "assistant"
}
msgs = append(msgs, ChatMessageRecord{
ID: int64(i),
Role: role,
Content: fmt.Sprintf("这是测试消息第 %d 条内容", i),
})
}
cfg := WindowConfig{
MaxTurns: 8, // 最多 8 条
TokenBudget: 5000,
ReserveTail: 500,
}
window, early, truncated := AssembleWindow(msgs, cfg)
if !truncated {
t.Fatalf("expected truncated to be true")
}
if len(window) != 8 {
t.Errorf("expected 8 messages in window, got %d", len(window))
}
if len(early) != 12 {
t.Errorf("expected 12 messages in early, got %d", len(early))
}
// 检查窗口中第一条是否是第 13 条
if window[0].ID != 13 {
t.Errorf("expected window start with ID 13, got %d", window[0].ID)
}
// 检查窗口最后一条是否是第 20 条
if window[len(window)-1].ID != 20 {
t.Errorf("expected window end with ID 20, got %d", window[len(window)-1].ID)
}
}
func TestAssembleWindow_TokenBudgetLimit(t *testing.T) {
// 构造 5 条消息,每条内容长度 350 字符
var msgs []ChatMessageRecord
for i := 1; i <= 5; i++ {
msgs = append(msgs, ChatMessageRecord{
ID: int64(i),
Role: "user",
Content: strings.Repeat("长文本七字内容", 50), // 7 * 50 = 350 字符
})
}
cfg := WindowConfig{
MaxTurns: 10, // 条数未超限
TokenBudget: 1000, // 预算 1000
ReserveTail: 200, // 净可用 800,最多容纳 2 条 (300*2=600 <= 800)
}
window, early, truncated := AssembleWindow(msgs, cfg)
if !truncated {
t.Fatalf("expected truncated to be true")
}
if len(window) != 2 {
t.Errorf("expected 2 messages in window due to token budget, got %d", len(window))
}
if len(early) != 3 {
t.Errorf("expected 3 early messages, got %d", len(early))
}
}
func TestFormatNumberedDialogue(t *testing.T) {
msgs := []ChatMessageRecord{
{Role: "user", Content: "你好,请问你是谁?"},
{Role: "assistant", Content: "我是水木东方研发协同参谋。"},
}
formatted := FormatNumberedDialogue(msgs, 5)
expected := "[5] 用户: 你好,请问你是谁?\n[6] 助手: 我是水木东方研发协同参谋。\n"
if formatted != expected {
t.Errorf("expected formatted output:\n%s\ngot:\n%s", expected, formatted)
}
}