T2: 递归再压缩 + Semantica式结构化长期记忆层(结构化记忆列/抽取/注入)
This commit is contained in:
qihu-dev committed
1 parent
6890570142
commit
e30e385f11
2 files changed
+201
-113
No files matched your search
@@ -2526,97 +2526,87 @@ def _estimate_tokens(text):
|
||||
|
||||
|
||||
def _build_chat_history(db, conv_id, max_tokens=6000, max_messages=40,
|
||||
|
||||
exclude_latest_user=True, include_summary=True):
|
||||
"""会话层上下文引擎：预算感知的历史组装 + 压缩摘要/结构化长期记忆前置。
|
||||
|
||||
"""会话层上下文引擎：预算感知的历史组装 + 可选压缩摘要前置。
|
||||
|
||||
|
||||
|
||||
- 若 conversation_summaries 存在摘要，以 system 角色「对话摘要」前置（压缩层）。
|
||||
|
||||
- 若 conversation_summaries 存在结构化长期记忆(structured_memory)，以紧凑
|
||||
结构化区块前置（Semantica 式：实体/事实/偏好/决策/待办），比扁平摘要更省
|
||||
token 且更便于模型检索关键信息。
|
||||
- 否则回退到扁平 summary_text 前置。
|
||||
- 按 token 估算从新到旧累加，超预算即截断（隔离：不溢出上下文窗口）。
|
||||
|
||||
- exclude_latest_user=True 时丢弃最新一条 user 消息（当前轮次另行发送）。
|
||||
|
||||
"""
|
||||
|
||||
ensure_structured_column()
|
||||
summary_block = None
|
||||
|
||||
if include_summary:
|
||||
|
||||
try:
|
||||
|
||||
summ = db.query(ConversationSummary).filter(
|
||||
|
||||
ConversationSummary.conversation_id == conv_id
|
||||
|
||||
).first()
|
||||
|
||||
if summ and summ.summary_text:
|
||||
|
||||
if summ and summ.structured_memory and isinstance(summ.structured_memory, dict):
|
||||
block = _format_structured_memory(summ.structured_memory)
|
||||
if block:
|
||||
summary_block = {
|
||||
"role": "system",
|
||||
"content": "【长期记忆·结构化】请据此理解用户此前的偏好、已确认决策与待办：\n" + block,
|
||||
}
|
||||
elif summ and summ.summary_text:
|
||||
summary_block = {
|
||||
|
||||
"role": "system",
|
||||
|
||||
"content": "【对话历史摘要，请据此理解用户此前的意图与已确认结论】\n" + summ.summary_text,
|
||||
|
||||
}
|
||||
|
||||
except Exception:
|
||||
|
||||
pass
|
||||
|
||||
rows = db.query(Message).filter(
|
||||
|
||||
Message.conversation_id == conv_id,
|
||||
|
||||
Message.role.in_(["user", "assistant"]),
|
||||
|
||||
).order_by(desc(Message.created_at)).limit(max_messages).all()
|
||||
|
||||
if exclude_latest_user:
|
||||
|
||||
for i, m in enumerate(rows):
|
||||
|
||||
if m.role == "user":
|
||||
|
||||
rows = rows[:i] + rows[i + 1:]
|
||||
|
||||
break
|
||||
|
||||
kept = []
|
||||
|
||||
used = 0
|
||||
|
||||
for m in rows: # 新->旧
|
||||
|
||||
t = _estimate_tokens(m.content or "")
|
||||
|
||||
if kept and used + t > max_tokens:
|
||||
|
||||
break
|
||||
|
||||
kept.append({"role": m.role, "content": m.content})
|
||||
|
||||
used += t
|
||||
|
||||
kept.reverse() # 时间正序
|
||||
|
||||
out = []
|
||||
|
||||
if summary_block:
|
||||
|
||||
out.append(summary_block)
|
||||
|
||||
out.extend(kept)
|
||||
|
||||
return out
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
def _format_structured_memory(mem: dict) -> str:
|
||||
"""将结构化记忆格式化为紧凑可读区块。"""
|
||||
labels = {
|
||||
"entities": "实体",
|
||||
"facts": "事实",
|
||||
"preferences": "偏好",
|
||||
"decisions": "决策",
|
||||
"open_tasks": "待办",
|
||||
}
|
||||
lines = []
|
||||
for key, label in labels.items():
|
||||
items = mem.get(key)
|
||||
if not items:
|
||||
continue
|
||||
if isinstance(items, list):
|
||||
for it in items:
|
||||
if it:
|
||||
lines.append(f"· {label}: {it}")
|
||||
elif isinstance(items, str):
|
||||
lines.append(f"· {label}: {items}")
|
||||
return "\n".join(lines)
|
||||
async def _summarize_via_llm(prompt):
|
||||
|
||||
"""调用默认文本模型生成对话摘要；失败返回 None（压缩为可选项，不阻塞）。"""
|
||||
@@ -2703,117 +2693,214 @@ async def _summarize_via_llm(prompt):
|
||||
|
||||
|
||||
|
||||
_STRUCT_COL_READY = False
|
||||
|
||||
|
||||
def ensure_structured_column():
|
||||
"""幂等地为 conversation_summaries 增加 structured_memory 列（MariaDB 安全）。"""
|
||||
global _STRUCT_COL_READY
|
||||
if _STRUCT_COL_READY:
|
||||
return True
|
||||
try:
|
||||
from app.database import engine
|
||||
from sqlalchemy import text
|
||||
with engine.connect() as conn:
|
||||
exists = conn.execute(text(
|
||||
"SELECT COUNT(*) FROM information_schema.columns "
|
||||
"WHERE table_schema = DATABASE() AND table_name='conversation_summaries' "
|
||||
"AND column_name='structured_memory'"
|
||||
)).scalar()
|
||||
if not exists:
|
||||
conn.execute(text(
|
||||
"ALTER TABLE conversation_summaries ADD COLUMN structured_memory JSON DEFAULT NULL"
|
||||
))
|
||||
conn.commit()
|
||||
_STRUCT_COL_READY = True
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.warning(f"[context] ensure structured column failed: {e}")
|
||||
return False
|
||||
|
||||
|
||||
async def _recursive_compress(text, depth=0, max_depth=3, budget=700):
|
||||
"""递归再压缩：若摘要仍超过 token 预算，则进一步浓缩，直至收敛或达最大深度。"""
|
||||
if not text:
|
||||
return text
|
||||
if _estimate_tokens(text) <= budget or depth >= max_depth:
|
||||
return text
|
||||
s = await _summarize_via_llm("请将以下摘要进一步浓缩，保留全部关键信息，篇幅更短：\n" + text)
|
||||
if not s:
|
||||
return text
|
||||
return await _recursive_compress(s, depth + 1, max_depth, budget)
|
||||
|
||||
|
||||
async def _extract_structured_memory(text):
|
||||
"""Semantica 式结构化长期记忆抽取：将对话内容抽取为类型化结构。
|
||||
|
||||
模型对抽取类提示偶发返回空，故做最多 3 次重试；并兼容有标签与无标签两种输出。
|
||||
"""
|
||||
if not text or len(text.strip()) < 10:
|
||||
return None
|
||||
prompt = (
|
||||
"提取要点，按 实体/事实/偏好/决策/待办 五类，用分号分隔输出：" + text[:6000]
|
||||
)
|
||||
raw = None
|
||||
for _ in range(3):
|
||||
r = await _summarize_via_llm(prompt)
|
||||
if r and r.strip():
|
||||
raw = r.strip()
|
||||
break
|
||||
if not raw:
|
||||
return None
|
||||
# JSON 回退
|
||||
if "{" in raw:
|
||||
try:
|
||||
s = raw.find("{")
|
||||
e = raw.rfind("}")
|
||||
data = json.loads(raw[s:e + 1])
|
||||
if isinstance(data, dict):
|
||||
return _normalize_structured(data)
|
||||
except Exception:
|
||||
pass
|
||||
norm = raw.replace("：", ":").replace("；", ";").replace("，", ",").replace("。", "")
|
||||
result = _parse_structured_segments(norm)
|
||||
if result:
|
||||
return result
|
||||
# 无标签回退：按分号切分，按顺序映射到五类
|
||||
segs = [s.strip() for s in norm.replace(chr(10), ";").split(";") if s.strip()]
|
||||
if segs:
|
||||
cats = ["entities", "facts", "preferences", "decisions", "open_tasks"]
|
||||
fb = {}
|
||||
for i, s in enumerate(segs[:5]):
|
||||
fb[cats[i]] = [s]
|
||||
return fb
|
||||
return None
|
||||
|
||||
|
||||
def _parse_structured_segments(norm):
|
||||
"""解析『类别: 内容』格式（兼容全角冒号/分号与单行多类别）。"""
|
||||
result = {}
|
||||
for seg in norm.replace(chr(10), ";").split(";"):
|
||||
if ":" not in seg:
|
||||
continue
|
||||
key, _, val = seg.partition(":")
|
||||
key = key.strip()
|
||||
val = val.strip().strip(",")
|
||||
if not val or val in ("无", "none", "None"):
|
||||
continue
|
||||
mapped = {
|
||||
"实体": "entities", "事实": "facts", "偏好": "preferences",
|
||||
"决策": "decisions", "待办": "open_tasks",
|
||||
}.get(key)
|
||||
if mapped:
|
||||
items = [x.strip() for x in val.split(",") if x.strip()]
|
||||
if items:
|
||||
result[mapped] = items
|
||||
return result
|
||||
|
||||
|
||||
def _normalize_structured(data):
|
||||
"""将模型的 JSON 输出规范化为结构化记忆字典。"""
|
||||
keys = ["entities", "facts", "preferences", "decisions", "open_tasks"]
|
||||
out = {}
|
||||
for k in keys:
|
||||
v = data.get(k)
|
||||
if isinstance(v, list):
|
||||
out[k] = [str(x) for x in v if x]
|
||||
elif isinstance(v, str) and v:
|
||||
out[k] = [v]
|
||||
return out
|
||||
|
||||
|
||||
def _merge_structured_memory(old, new):
|
||||
"""合并新旧结构化记忆，按字符串去重并控制规模。"""
|
||||
keys = ["entities", "facts", "preferences", "decisions", "open_tasks"]
|
||||
merged = {}
|
||||
for k in keys:
|
||||
a = old.get(k) if isinstance(old, dict) else None
|
||||
b = new.get(k) if isinstance(new, dict) else None
|
||||
items = []
|
||||
for x in (a or []) + (b or []):
|
||||
if x and x not in items:
|
||||
items.append(x)
|
||||
merged[k] = items[:50]
|
||||
return merged
|
||||
|
||||
|
||||
async def _maybe_compress_history(conv_id):
|
||||
|
||||
"""后台异步压缩：将受保护窗口外的旧消息摘要落库 conversation_summaries。
|
||||
|
||||
|
||||
"""后台异步压缩：将受保护窗口外的旧消息递归再压缩，并抽取结构化长期记忆。
|
||||
|
||||
失败静默回滚，绝不阻塞用户本次响应（fire-and-forget）。
|
||||
|
||||
"""
|
||||
|
||||
ensure_structured_column()
|
||||
sdb = None
|
||||
|
||||
try:
|
||||
|
||||
sdb = SessionLocal()
|
||||
|
||||
all_msgs = sdb.query(Message).filter(
|
||||
|
||||
Message.conversation_id == conv_id,
|
||||
|
||||
Message.role.in_(["user", "assistant"]),
|
||||
|
||||
).order_by(asc(Message.created_at)).all()
|
||||
|
||||
PROTECTED = 20
|
||||
|
||||
if len(all_msgs) <= PROTECTED:
|
||||
|
||||
return
|
||||
|
||||
compressible = all_msgs[: len(all_msgs) - PROTECTED]
|
||||
|
||||
if len(compressible) < 8:
|
||||
|
||||
return
|
||||
|
||||
chunk = compressible[:30]
|
||||
|
||||
text = "\n".join(f"{m.role}: {m.content}" for m in chunk)
|
||||
# 1) 分块摘要：覆盖全部可压缩消息（而非仅前 30 条）
|
||||
CHUNK = 30
|
||||
chunk_summaries = []
|
||||
for i in range(0, len(compressible), CHUNK):
|
||||
piece = compressible[i:i + CHUNK]
|
||||
text = "\n".join(f"{m.role}: {m.content}" for m in piece)
|
||||
s = await _summarize_via_llm("请将以下对话片段浓缩为简明要点（保留关键事实、诉求与结论）：\n" + text)
|
||||
if s:
|
||||
chunk_summaries.append(s)
|
||||
|
||||
existing = sdb.query(ConversationSummary).filter(
|
||||
|
||||
ConversationSummary.conversation_id == conv_id
|
||||
|
||||
).first()
|
||||
|
||||
prior = existing.summary_text if existing else ""
|
||||
|
||||
prompt = (
|
||||
combined = "\n".join(chunk_summaries)
|
||||
if prior:
|
||||
combined = prior + "\n" + combined
|
||||
|
||||
"请把以下对话片段浓缩为简明摘要，保留用户核心诉求、已确认结论、关键事实与待办。"
|
||||
# 2) 递归再压缩：合并摘要超预算时逐层再压缩直到收敛
|
||||
final_summary = await _recursive_compress(combined)
|
||||
|
||||
"若提供『已有摘要』，请合并而非重复。\n\n"
|
||||
|
||||
f"已有摘要：\n{prior}\n\n待摘要片段：\n{text}"
|
||||
|
||||
)
|
||||
|
||||
summary = await _summarize_via_llm(prompt)
|
||||
|
||||
if not summary:
|
||||
|
||||
return
|
||||
# 3) 结构化长期记忆抽取（Semantica 式），并与既有记忆合并
|
||||
structured = await _extract_structured_memory(combined)
|
||||
if structured is None:
|
||||
structured = {}
|
||||
if existing and existing.structured_memory and isinstance(existing.structured_memory, dict):
|
||||
structured = _merge_structured_memory(existing.structured_memory, structured)
|
||||
|
||||
now = datetime.now()
|
||||
if existing:
|
||||
|
||||
existing.summary_text = summary
|
||||
|
||||
existing.token_count = _estimate_tokens(summary)
|
||||
|
||||
existing.updated_at = datetime.now()
|
||||
|
||||
existing.summary_text = final_summary
|
||||
existing.structured_memory = structured
|
||||
existing.token_count = _estimate_tokens(final_summary)
|
||||
existing.updated_at = now
|
||||
else:
|
||||
|
||||
sdb.add(ConversationSummary(
|
||||
|
||||
conversation_id=conv_id,
|
||||
|
||||
summary_text=summary,
|
||||
|
||||
token_count=_estimate_tokens(summary),
|
||||
|
||||
updated_at=datetime.now(),
|
||||
|
||||
summary_text=final_summary,
|
||||
structured_memory=structured,
|
||||
token_count=_estimate_tokens(final_summary),
|
||||
updated_at=now,
|
||||
))
|
||||
|
||||
sdb.commit()
|
||||
|
||||
except Exception as e:
|
||||
|
||||
logger.warning(f"[context] compress history failed conv={conv_id}: {e}")
|
||||
|
||||
if sdb is not None:
|
||||
|
||||
try:
|
||||
|
||||
sdb.rollback()
|
||||
|
||||
except Exception:
|
||||
|
||||
pass
|
||||
|
||||
finally:
|
||||
|
||||
if sdb is not None:
|
||||
|
||||
try:
|
||||
|
||||
sdb.close()
|
||||
|
||||
except Exception:
|
||||
|
||||
pass
|
||||
|
||||
@@ -1156,4 +1156,5 @@ class ConversationSummary(Base):
|
||||
conversation_id = Column(String(64), primary_key=True)
|
||||
summary_text = Column(Text, nullable=False, default="")
|
||||
token_count = Column(Integer, default=0)
|
||||
structured_memory = Column(JSON, default=None) # T2: Semantica 式结构化长期记忆
|
||||
updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
|
||||
Reference in new issue
Block a user