450 lines
15 KiB
Go
450 lines
15 KiB
Go
package ctxmgr
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"log/slog"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/jackc/pgx/v5"
|
||
)
|
||
|
||
// ChatSummary 会话滚动摘要数据模型
|
||
type ChatSummary struct {
|
||
SessionID string `json:"session_id"`
|
||
Narrative string `json:"narrative"` // 叙事摘要（≤300 字）；PU6 shadow 下始终为 deterministic
|
||
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"`
|
||
Source string `json:"source,omitempty"` // deterministic（served）
|
||
FreshnessNote string `json:"freshness_note,omitempty"` // shadow 失败或未持久化时的可解释回退说明
|
||
Shadow *SemanticShadowReport `json:"shadow,omitempty"` // advisory only；REPLACE 未接线
|
||
VisibilityDigest string `json:"visibility_digest,omitempty"` // 授权可见集；未计算时为空
|
||
FromMessageID int64 `json:"from_message_id,omitempty"`
|
||
ToMessageID int64 `json:"to_message_id,omitempty"`
|
||
PreviousRevision int `json:"previous_revision,omitempty"`
|
||
Revision int `json:"revision,omitempty"`
|
||
ProtocolVersion string `json:"protocol_version,omitempty"`
|
||
ModelVersion string `json:"model_version,omitempty"`
|
||
// Persisted 为 true 表示配置的存储接受了这次写入。有数据库时只有 INSERT 成功才为 true。
|
||
Persisted bool `json:"persisted,omitempty"`
|
||
}
|
||
|
||
const summaryNotDurableNote = "摘要未写入数据库，仅本进程可见"
|
||
|
||
// 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 cached, ok := m.Cache().Get(LayerC5Summary, sessionID); ok {
|
||
if sum, ok := cached.(*ChatSummary); ok && sum != nil {
|
||
copySum := *sum
|
||
return ©Sum, nil
|
||
}
|
||
}
|
||
|
||
if m.pool != nil {
|
||
sum, err := m.loadChatSummary(ctx, sessionID)
|
||
if err == nil && sum != nil {
|
||
m.Cache().Set(LayerC5Summary, sessionID, sum, 15)
|
||
return sum, nil
|
||
}
|
||
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
|
||
slog.Warn("chat summary load failed", "session_id", sessionID, "err", err)
|
||
}
|
||
}
|
||
|
||
m.mu.RLock()
|
||
defer m.mu.RUnlock()
|
||
if m.summaries != nil {
|
||
if s, ok := m.summaries[sessionID]; ok && s != nil {
|
||
// 返回副本
|
||
copySum := *s
|
||
m.Cache().Set(LayerC5Summary, sessionID, ©Sum, 15)
|
||
return ©Sum, 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()
|
||
|
||
// 主动失效 C5 摘要缓存与 C4 上下文缓存
|
||
m.Cache().Invalidate(LayerC5Summary, sum.SessionID)
|
||
m.Cache().InvalidateLayer(LayerC4Context)
|
||
|
||
if m.pool == nil {
|
||
sum.Persisted = true
|
||
m.rememberSummary(sum)
|
||
return nil
|
||
}
|
||
|
||
factsJSON, err := json.Marshal(sum.Facts)
|
||
if err != nil {
|
||
return fmt.Errorf("chat summary facts: %w", err)
|
||
}
|
||
openQuestionsJSON, err := json.Marshal(sum.OpenQuestions)
|
||
if err != nil {
|
||
return fmt.Errorf("chat summary questions: %w", err)
|
||
}
|
||
var shadowJSON []byte
|
||
if sum.Shadow != nil {
|
||
shadowJSON, err = json.Marshal(sum.Shadow)
|
||
if err != nil {
|
||
return fmt.Errorf("chat summary shadow: %w", err)
|
||
}
|
||
}
|
||
_, err = m.pool.Exec(ctx, `
|
||
INSERT INTO chat_summaries (
|
||
session_id, narrative, facts, open_questions, covered_count, covered_until, updated_at,
|
||
source, visibility_digest, from_message_id, to_message_id, previous_revision, revision,
|
||
protocol_version, model_version, freshness_note, shadow_json
|
||
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)
|
||
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,
|
||
source = EXCLUDED.source,
|
||
visibility_digest = EXCLUDED.visibility_digest,
|
||
from_message_id = EXCLUDED.from_message_id,
|
||
to_message_id = EXCLUDED.to_message_id,
|
||
previous_revision = EXCLUDED.previous_revision,
|
||
revision = EXCLUDED.revision,
|
||
protocol_version = EXCLUDED.protocol_version,
|
||
model_version = EXCLUDED.model_version,
|
||
freshness_note = EXCLUDED.freshness_note,
|
||
shadow_json = EXCLUDED.shadow_json
|
||
`, sum.SessionID, sum.Narrative, factsJSON, openQuestionsJSON, sum.CoveredCount, sum.CoveredUntil, sum.UpdatedAt,
|
||
sum.Source, sum.VisibilityDigest, nullSummaryID(sum.FromMessageID), nullSummaryID(sum.ToMessageID),
|
||
sum.PreviousRevision, sum.Revision, sum.ProtocolVersion, sum.ModelVersion, sum.FreshnessNote, nullShadowJSON(shadowJSON))
|
||
if err != nil {
|
||
sum.Persisted = false
|
||
noteSummaryNotDurable(sum)
|
||
m.rememberSummary(sum)
|
||
slog.Warn("chat summary persist failed", "session_id", sum.SessionID, "err", err)
|
||
return fmt.Errorf("chat summary persist failed: %w", err)
|
||
}
|
||
sum.Persisted = true
|
||
m.rememberSummary(sum)
|
||
return nil
|
||
}
|
||
|
||
func nullShadowJSON(b []byte) any {
|
||
if len(b) == 0 {
|
||
return nil
|
||
}
|
||
return b
|
||
}
|
||
|
||
func (m *ContextManager) loadChatSummary(ctx context.Context, sessionID string) (*ChatSummary, error) {
|
||
var sum ChatSummary
|
||
var factsJSON, openQuestionsJSON, shadowJSON []byte
|
||
var fromID, toID *int64
|
||
err := m.pool.QueryRow(ctx, `
|
||
SELECT session_id, narrative, facts, open_questions, covered_count, covered_until, updated_at,
|
||
source, visibility_digest, from_message_id, to_message_id, previous_revision, revision,
|
||
protocol_version, model_version, freshness_note, shadow_json
|
||
FROM chat_summaries
|
||
WHERE session_id = $1
|
||
`, sessionID).Scan(
|
||
&sum.SessionID, &sum.Narrative, &factsJSON, &openQuestionsJSON, &sum.CoveredCount, &sum.CoveredUntil, &sum.UpdatedAt,
|
||
&sum.Source, &sum.VisibilityDigest, &fromID, &toID, &sum.PreviousRevision, &sum.Revision,
|
||
&sum.ProtocolVersion, &sum.ModelVersion, &sum.FreshnessNote, &shadowJSON,
|
||
)
|
||
if err != nil {
|
||
// Compatibility: older schemas without provenance columns.
|
||
if isUndefinedColumn(err) {
|
||
return m.loadChatSummaryLegacy(ctx, sessionID)
|
||
}
|
||
return nil, err
|
||
}
|
||
_ = json.Unmarshal(factsJSON, &sum.Facts)
|
||
_ = json.Unmarshal(openQuestionsJSON, &sum.OpenQuestions)
|
||
if len(shadowJSON) > 0 {
|
||
var report SemanticShadowReport
|
||
if json.Unmarshal(shadowJSON, &report) == nil {
|
||
sum.Shadow = &report
|
||
}
|
||
}
|
||
if fromID != nil {
|
||
sum.FromMessageID = *fromID
|
||
}
|
||
if toID != nil {
|
||
sum.ToMessageID = *toID
|
||
}
|
||
sum.Persisted = true
|
||
return &sum, nil
|
||
}
|
||
|
||
func (m *ContextManager) loadChatSummaryLegacy(ctx context.Context, sessionID string) (*ChatSummary, error) {
|
||
var sum ChatSummary
|
||
var factsJSON, openQuestionsJSON []byte
|
||
var fromID, toID *int64
|
||
err := m.pool.QueryRow(ctx, `
|
||
SELECT session_id, narrative, facts, open_questions, covered_count, covered_until, updated_at,
|
||
source, visibility_digest, from_message_id, to_message_id, previous_revision, revision,
|
||
protocol_version, model_version
|
||
FROM chat_summaries
|
||
WHERE session_id = $1
|
||
`, sessionID).Scan(
|
||
&sum.SessionID, &sum.Narrative, &factsJSON, &openQuestionsJSON, &sum.CoveredCount, &sum.CoveredUntil, &sum.UpdatedAt,
|
||
&sum.Source, &sum.VisibilityDigest, &fromID, &toID, &sum.PreviousRevision, &sum.Revision,
|
||
&sum.ProtocolVersion, &sum.ModelVersion,
|
||
)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
_ = json.Unmarshal(factsJSON, &sum.Facts)
|
||
_ = json.Unmarshal(openQuestionsJSON, &sum.OpenQuestions)
|
||
if fromID != nil {
|
||
sum.FromMessageID = *fromID
|
||
}
|
||
if toID != nil {
|
||
sum.ToMessageID = *toID
|
||
}
|
||
sum.Persisted = true
|
||
return &sum, nil
|
||
}
|
||
|
||
func isUndefinedColumn(err error) bool {
|
||
if err == nil {
|
||
return false
|
||
}
|
||
msg := strings.ToLower(err.Error())
|
||
return strings.Contains(msg, "freshness_note") || strings.Contains(msg, "shadow_json") || strings.Contains(msg, "undefined_column")
|
||
}
|
||
|
||
func (m *ContextManager) rememberSummary(sum *ChatSummary) {
|
||
m.mu.Lock()
|
||
defer m.mu.Unlock()
|
||
if m.summaries == nil {
|
||
m.summaries = make(map[string]*ChatSummary)
|
||
}
|
||
copySum := *sum
|
||
m.summaries[sum.SessionID] = ©Sum
|
||
}
|
||
|
||
func nullSummaryID(id int64) any {
|
||
if id <= 0 {
|
||
return nil
|
||
}
|
||
return id
|
||
}
|
||
|
||
func noteSummaryNotDurable(sum *ChatSummary) {
|
||
if sum == nil || strings.Contains(sum.FreshnessNote, summaryNotDurableNote) {
|
||
return
|
||
}
|
||
if sum.FreshnessNote == "" {
|
||
sum.FreshnessNote = summaryNotDurableNote
|
||
return
|
||
}
|
||
sum.FreshnessNote += "；" + summaryNotDurableNote
|
||
}
|
||
|
||
// 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
|
||
newSum.Source = "deterministic"
|
||
newSum.ProtocolVersion = "deterministic-v1"
|
||
if len(earlyMsgs) > 0 {
|
||
newSum.FromMessageID = earlyMsgs[0].ID
|
||
newSum.ToMessageID = earlyMsgs[len(earlyMsgs)-1].ID
|
||
}
|
||
if oldSum != nil {
|
||
prev := oldSum.Revision
|
||
if prev < 1 {
|
||
prev = 1
|
||
}
|
||
newSum.PreviousRevision = prev
|
||
newSum.Revision = prev + 1
|
||
} else {
|
||
newSum.Revision = 1
|
||
}
|
||
|
||
// Always bind the authorized-visible set digest (even when shadow is off).
|
||
minimized := MinimizeAuthorizedMessages(earlyMsgs)
|
||
newSum.VisibilityDigest = DigestAuthorizedVisibility(minimized)
|
||
|
||
// SEMANTIC_SUMMARY_SHADOW：后台可靠任务生成 advisory 语义摘要；请求路径立即返回确定性结果。
|
||
// SEMANTIC_SUMMARY_REPLACE 故意不接线，served Narrative 永不被语义结果替换。
|
||
scheduleSemanticShadow(ctx, m, newSum, earlyMsgs, oldSum)
|
||
|
||
// 写库失败仍返回确定性摘要，供本轮使用；错误留在 Persisted/FreshnessNote，不伪装成已持久化。
|
||
_ = 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, "、")
|
||
}
|