lzf_0626
3a1065ca1e
## 结论先说：真正的根因和最初两个假设都不是同一个
你原来的判断是「第二条召回恒空、第三条传导断链」，并猜第三条是
「`record_evidence` 返回 False 时该不该补写重建事件」。查完库发现：
1. **`record_evidence` 返回 False 时数据零净变化**（幂等命中直接 return；
并发冲突把刚加的计数减回来），所以那个判断点解释不了画像停更；
2. **`memory_extraction_worker` 是有写重建事件的**（`if recorded:`），
233 条 `profile.rebuild_requested` 全 published；
3. 真正断在两处，都在 **`episode_worker`** 这条片段链路上。
## 根因一：`_touch_retry` 把「重复聚合」记成「抽取失败」（P0，已修）
`episode_worker._persist` 在 `content_hash` 命中（同一段会话被重复聚合）时调
`_touch_retry` 做 `retry_count += 1`。但 `retry_count` 的语义是**抽取失败次数**
（由 `_mark_failed` 累加），而"分片逻辑每轮重新看到同一段会话"根本不是失败。
后果是实测出来的：
episodes 待提取片段：retry_count=1405（40 条，全是客户 9001）
consume_pending 逐条件筛：
仅 status in (待提取,失败) -> 40
+ retry_count < max_retry(3) -> 0 ← 一个都不剩
+ promoted_to_ltm IS FALSE -> 40
这 40 条片段**永远不可能被选中** ⇒ 新记忆进不来 ⇒ 画像停在旧值。
（另外还观察到一次运行里它从 1405 涨到 1407 —— 常驻 Worker 每轮都在继续推高。）
**修法**：重复聚合不再触碰 `retry_count`，连 `flush` 都不做（内容没变就只是看到）。
## 根因二：片段链路从不投「画像重建」事件（已修）
`memory_extraction_worker` 写完证据会投 `profile.rebuild_requested`；
而 `episode_worker` 写完证据直接 `_mark_done` 就结束了 —— **完全没有这一步**。
所以即使片段被成功抽取，画像也不会重建。
**修法**：`episode_worker` 也捕获 `recorded` 并在为真时投同样的事件
（`trigger="episode_extraction"`）。只在 `recorded=True` 时投：幂等命中时证据与计数
都没有净变化，投一次是白跑。
## 数据修复
代码修好不会让已写进库的脏计数自己恢复 —— 那 40 条片段仍然超预算。
用 `tools` 级别的临时脚本把**确实被污染的行**重置（条件收紧为三者交集）：
extraction_status IN ('待提取','失败') AND promoted_to_ltm = 0 AND retry_count >= 3
-> 重置 40 行；之后可被 consume_pending 选中的片段从 0 恢复到 40
## 实测
- **重试预算修复**：`consume_episodes` 从"领不到任何片段"变为能领到；
分两批消费完 40 条（全部 `no_fact` —— 那些片段摘要里确实没有用户陈述，
属正常结果），待提取 40 → 0
- **重建事件修复**：那 40 条全是 `no_fact`，走不到 `if recorded:`，所以**
实测不到**。为不留下"改了但没验证"，另造了一条含用户陈述的片段：
consume_episodes -> extracted=[191]
9001 的 rebuild 事件 3 -> 4（增量 1）
memory_evidence 8 -> 9
另外注意到基线在我造数据前已经由 1 涨到 3 —— 说明常驻 Worker 也在这期间
投过事件，修复在真实链路里同样生效。
- `pytest tests/unit tests/contract` -> 1428 passed / 2 skipped / 1 failed
（剩下的 1 个是投顾工作台页面被替换所致，与本次无关）
## 测试
两个既有用例断言的正是被修掉的旧行为，已更新，并把第二个改造成**防回归守卫**
`test_repeated_aggregation_never_bumps_retry_count` —— 它守着
"`retry_count` 被重复聚合推高到 `max_retry` 之上会导致片段永久滞留"这个 P0。
`tests/unit/worker/test_episode_worker.py` 16 passed。
## 未处理
「召回恒空」（员工身份下 `recall` 取的是自己作为客户的记忆）**本次没动**。
它需要给 `AgentRequest` 加 `target_customer_id` 并配套越权校验，属接口契约变更；
排查报告给的建议是保持现状、员工侧走 `query_customer_profile` 工具。
要按"支持目标客户维度"做，请确认，我再单独一提交。
445 lines
20 KiB
Python
445 lines
20 KiB
Python
"""会话片段（episode）抽取：把会话消息按"会话片段"聚合后写入 `episodes`。
|
||
|
||
为什么需要它（对应 P3 缺口 2）：基线把 `episodes` 定义为中期记忆的抽取单位
|
||
（"会话结束、达到 Token 阈值或转人工时生成片段"），而现有实现按**单条消息**抽取，
|
||
`episodes` 表此前零 ORM、零使用。这里补上片段粒度与幂等落库。
|
||
|
||
聚合粒度：同一 `customer_id + session_id` 的消息按 `created_at` 升序，相邻消息
|
||
间隔超过 `gap_minutes`（默认 30 分钟）即切分为新片段；单个片段不足
|
||
`min_messages` 条时并入上一片段，避免尾部产生无意义碎片。摘要在片段内按
|
||
`[消息序号 | 角色] 正文` 拼接，只截断不改造原文。
|
||
|
||
幂等键：`content_hash = sha256(客户 + 会话 + 每条消息的 id 与正文)` 直接落到表上的
|
||
唯一键 `content_hash`。同一片段内容不变时哈希不变 → 重复调用不会产生重复片段；
|
||
片段内容变化（新消息并入或编辑）时哈希变化 → 视为新片段，旧片段保留为历史。
|
||
冲突时依赖唯一键拦截（`IntegrityError` 回滚到保存点），不使用"先查后写"作为唯一防线。
|
||
|
||
消费环节（对应 P3 缺口 4）：`EpisodeExtractionConsumer` 把已落库、状态为待提取的片段
|
||
接入记忆抽取，写入 `memory_unit` 后把片段置为已处理（`已完成` + `promoted_to_ltm`）。
|
||
失败只累加 `retry_count` 并把状态置为 `失败`，片段保持可重试、记忆一行不写；重复消费
|
||
由 `content_hash` 派生的 `MemoryEvidence.idempotency_key` 兜底，不产生第二条证据。
|
||
"""
|
||
|
||
import hashlib
|
||
import json
|
||
import logging
|
||
from dataclasses import dataclass, field
|
||
from datetime import UTC, datetime, timedelta
|
||
from uuid import NAMESPACE_URL, uuid4, uuid5
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.exc import IntegrityError
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.model.conversation import ConversationMessage
|
||
from app.model.episode import Episode
|
||
from app.model.platform import DomainEventOutbox
|
||
from app.service.memory_extraction_service import (
|
||
MemoryExtractionService,
|
||
get_memory_extraction_service,
|
||
)
|
||
from app.service.memory_service import CacheDeleteAdapter, MemoryService
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
DEFAULT_GAP_MINUTES = 30
|
||
DEFAULT_MIN_MESSAGES = 2
|
||
SUMMARY_LIMIT = 4000
|
||
STATUS_PENDING = "待提取"
|
||
STATUS_PROCESSING = "处理中"
|
||
STATUS_DONE = "已完成"
|
||
STATUS_FAILED = "失败"
|
||
RETRYABLE_STATUSES = frozenset({STATUS_FAILED, STATUS_PENDING, STATUS_PROCESSING})
|
||
# 消费环节可选中的状态：只处理"待提取"与"失败"，"处理中"留给并发消费者自行收尾。
|
||
CONSUMABLE_STATUSES = (STATUS_PENDING, STATUS_FAILED)
|
||
# 失败片段的重试上限：达到上限后停在"失败"，不再无限重试（避免毒片段拖住批处理）。
|
||
MAX_EXTRACTION_RETRY = 3
|
||
USER_ROLE = "user"
|
||
EVIDENCE_TYPE = "会话片段"
|
||
SOURCE_TYPE = "用户自述"
|
||
EVIDENCE_WEIGHT = 0.05
|
||
|
||
|
||
@dataclass
|
||
class EpisodeResult:
|
||
"""一次抽取的结果；`inserted` 为新片段，`skipped` 为命中幂等边界。"""
|
||
|
||
inserted: list[int] = field(default_factory=list)
|
||
skipped: list[int] = field(default_factory=list)
|
||
|
||
@property
|
||
def total(self) -> int:
|
||
return len(self.inserted) + len(self.skipped)
|
||
|
||
|
||
class EpisodeWorker:
|
||
"""按会话片段聚合消息并落库；可重复调用，失败片段可由调用方重试。"""
|
||
|
||
def __init__(
|
||
self,
|
||
session: AsyncSession,
|
||
*,
|
||
gap_minutes: int = DEFAULT_GAP_MINUTES,
|
||
min_messages: int = DEFAULT_MIN_MESSAGES,
|
||
message_limit: int = 500,
|
||
) -> None:
|
||
self.session = session
|
||
self.gap = timedelta(minutes=max(1, gap_minutes))
|
||
self.min_messages = max(1, min_messages)
|
||
self.message_limit = max(1, min(message_limit, 2000))
|
||
|
||
async def aggregate(
|
||
self,
|
||
customer_id: int,
|
||
*,
|
||
session_id: str | None = None,
|
||
ended_before: datetime | None = None,
|
||
status: str = STATUS_PENDING,
|
||
) -> EpisodeResult:
|
||
"""聚合并落库；`ended_before` 用于只处理已结束（静默超过间隔）的片段。"""
|
||
now = datetime.now(UTC).replace(tzinfo=None)
|
||
cutoff = ended_before or (now - self.gap)
|
||
result = EpisodeResult()
|
||
for key_session, messages in await self._pending_sessions(customer_id, session_id, cutoff):
|
||
for segment in self._segment(messages):
|
||
if len(segment) < self.min_messages:
|
||
continue
|
||
episode = await self._persist(customer_id, key_session, segment, status)
|
||
target = result.inserted if episode is not None else result.skipped
|
||
target.append(segment[0].id)
|
||
return result
|
||
|
||
async def _pending_sessions(
|
||
self, customer_id: int, session_id: str | None, cutoff: datetime
|
||
) -> list[tuple[str, list[ConversationMessage]]]:
|
||
conditions = [
|
||
ConversationMessage.customer_id == customer_id,
|
||
ConversationMessage.created_at <= cutoff,
|
||
]
|
||
if session_id is not None:
|
||
conditions.append(ConversationMessage.session_id == session_id)
|
||
rows = await self.session.scalars(
|
||
select(ConversationMessage)
|
||
.where(*conditions)
|
||
.order_by(ConversationMessage.session_id, ConversationMessage.created_at,
|
||
ConversationMessage.id)
|
||
.limit(self.message_limit)
|
||
)
|
||
grouped: dict[str, list[ConversationMessage]] = {}
|
||
for message in rows:
|
||
grouped.setdefault(message.session_id, []).append(message)
|
||
return list(grouped.items())
|
||
|
||
def _segment(self, messages: list[ConversationMessage]) -> list[list[ConversationMessage]]:
|
||
"""按时间间隔切分；尾部不足 `min_messages` 的片段并入上一片段。"""
|
||
segments: list[list[ConversationMessage]] = []
|
||
current: list[ConversationMessage] = []
|
||
for message in messages:
|
||
if current and message.created_at - current[-1].created_at > self.gap:
|
||
segments.append(current)
|
||
current = []
|
||
current.append(message)
|
||
if current:
|
||
segments.append(current)
|
||
if len(segments) > 1 and len(segments[-1]) < self.min_messages:
|
||
tail = segments.pop()
|
||
segments[-1].extend(tail)
|
||
return segments
|
||
|
||
async def _persist(
|
||
self,
|
||
customer_id: int,
|
||
session_id: str,
|
||
segment: list[ConversationMessage],
|
||
status: str,
|
||
) -> Episode | None:
|
||
content_hash = self.content_hash(customer_id, session_id, segment)
|
||
existing = await self.session.scalar(
|
||
select(Episode).where(Episode.content_hash == content_hash)
|
||
)
|
||
if existing is not None:
|
||
# ⚠️ 这里**不要**碰 `retry_count`。
|
||
#
|
||
# 此前会调 `_touch_retry(existing)` 做 `retry_count += 1`（注释写的是
|
||
# "仅在未完成时累加重试计数"）。但 `retry_count` 的语义是**抽取失败**的次数
|
||
# （由 `_mark_failed` 累加），而"分片逻辑每轮重新看到同一段会话"根本不是失败
|
||
# —— 片段内容没变，`content_hash` 才会相同。
|
||
#
|
||
# 两者混用一个字段的后果是实测出来的：客户 9001 有 **40 条片段
|
||
# `retry_count=1405`**，远超 `MAX_EXTRACTION_RETRY`(3)，于是
|
||
# `consume_pending` 的 `retry_count < max_retry` 永远筛不中它们 ——
|
||
# **片段永久滞留 → 新记忆进不来 → 画像停在旧值**
|
||
# （`profile_snapshots` 停在 2026-09-10 13:57，而 `memory_unit.updated_at`
|
||
# 已经是 2026-09-13 11:03）。
|
||
#
|
||
# 重复看到同一片段就只是看到，不改变任何状态。
|
||
logger.debug("episode already persisted content_hash=%s", content_hash)
|
||
return None
|
||
episode = self._build(customer_id, session_id, segment, content_hash, status)
|
||
self.session.add(episode)
|
||
try:
|
||
async with self.session.begin_nested():
|
||
await self.session.flush()
|
||
except IntegrityError:
|
||
# 并发抽取命中 uk_episodes_content_hash：片段已由另一事务写入，本轮跳过。
|
||
logger.warning("episode already exists content_hash=%s", content_hash)
|
||
return None
|
||
return episode
|
||
|
||
def _build(
|
||
self,
|
||
customer_id: int,
|
||
session_id: str,
|
||
segment: list[ConversationMessage],
|
||
content_hash: str,
|
||
status: str,
|
||
) -> Episode:
|
||
started_at = segment[0].created_at
|
||
ended_at = segment[-1].created_at
|
||
portals = sorted({message.portal for message in segment if message.portal})
|
||
now = datetime.now(UTC).replace(tzinfo=None)
|
||
numbers = [message.message_no for message in segment if message.message_no is not None]
|
||
return Episode(
|
||
# 跨存储稳定标识由幂等键派生，重试与重复消费得到同一个 uuid。
|
||
episode_uuid=str(uuid5(NAMESPACE_URL, f"jr:episode:{session_id}:{content_hash}")),
|
||
customer_id=customer_id,
|
||
session_id=session_id[:64],
|
||
start_message_no=min(numbers) if numbers else None,
|
||
end_message_no=max(numbers) if numbers else None,
|
||
portals_involved=portals,
|
||
summary=self.summarize(segment),
|
||
extraction_status=status,
|
||
content_hash=content_hash,
|
||
retry_count=0,
|
||
started_at=started_at,
|
||
ended_at=ended_at,
|
||
start_at=started_at,
|
||
end_at=ended_at,
|
||
promoted_to_ltm=False,
|
||
created_at=now,
|
||
)
|
||
|
||
@staticmethod
|
||
def content_hash(
|
||
customer_id: int, session_id: str, segment: list[ConversationMessage]
|
||
) -> str:
|
||
"""片段指纹：消息 id 与正文共同决定，保证内容变化即产生新指纹。"""
|
||
payload = json.dumps(
|
||
{
|
||
"customer_id": customer_id,
|
||
"session_id": session_id,
|
||
"messages": [
|
||
{"id": message.id, "role": message.role,
|
||
"content": (message.content or "").strip()}
|
||
for message in segment
|
||
],
|
||
},
|
||
ensure_ascii=False,
|
||
sort_keys=True,
|
||
separators=(",", ":"),
|
||
)
|
||
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
|
||
|
||
@staticmethod
|
||
def summarize(segment: list[ConversationMessage]) -> str:
|
||
"""片段摘要：只做拼接与截断，不改写、不脱敏（脱敏由上游写入前完成）。"""
|
||
parts: list[str] = []
|
||
for message in segment:
|
||
number = message.message_no if message.message_no is not None else message.id
|
||
content = (message.content or "").strip().replace("\r\n", " ").replace("\n", " ")
|
||
parts.append(f"[{number} | {message.role}] {content}")
|
||
text = "\n".join(parts)
|
||
return text[:SUMMARY_LIMIT]
|
||
|
||
|
||
@dataclass
|
||
class EpisodeConsumptionResult:
|
||
"""一次消费的结果；`extracted` 为写出记忆的片段，`failed` 为保持可重试的片段。"""
|
||
|
||
extracted: list[int] = field(default_factory=list)
|
||
no_fact: list[int] = field(default_factory=list)
|
||
failed: list[int] = field(default_factory=list)
|
||
|
||
@property
|
||
def processed(self) -> int:
|
||
"""已终结（不再重试）的片段数：写出记忆的与判定无持久事实的。"""
|
||
return len(self.extracted) + len(self.no_fact)
|
||
|
||
|
||
class EpisodeExtractionConsumer:
|
||
"""消费待提取片段：复用 `MemoryExtractionService` 抽取后写入 `memory_unit`。
|
||
|
||
幂等由三层保证，都不需要新增库表或列：
|
||
1. 选中条件排除 `已完成`/`promoted_to_ltm` 的片段，成功片段不会被二次处理；
|
||
2. 证据幂等键 `episode.extraction:{content_hash}`（表上唯一键 `content_hash` 派生）
|
||
命中 `memory_evidence` 唯一键时视为已消费，不重复写证据；
|
||
3. `memory_unit` 按 `(customer_id, memory_key)` 语义更新，重复抽取同一受控键
|
||
不会堆积重复记忆行。
|
||
|
||
失败（模型不可用、输出违约）不清空已有进度：只累加 `retry_count` 并把状态置为
|
||
`失败`，记忆一行不写，片段在下一次调度继续可被选中；达到 `max_retry` 后停在该
|
||
状态，由运维按 `retry_count` 排查毒片段，而不是无限重试拖住批处理。
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
session: AsyncSession,
|
||
*,
|
||
extractor: MemoryExtractionService | None = None,
|
||
cache: CacheDeleteAdapter | None = None,
|
||
max_retry: int = MAX_EXTRACTION_RETRY,
|
||
limit: int = 20,
|
||
) -> None:
|
||
self.session = session
|
||
self.extractor = extractor if extractor is not None else get_memory_extraction_service()
|
||
# 召回热缓存适配器：写入生效后必须失效，否则新记忆在 TTL 内召回不到。
|
||
self.cache = cache
|
||
self.max_retry = max(1, max_retry)
|
||
self.limit = max(1, min(limit, 200))
|
||
|
||
async def consume_pending(self, *, limit: int | None = None) -> EpisodeConsumptionResult:
|
||
"""消费一批待提取片段；单个片段失败不影响同批其它片段。"""
|
||
bounded = max(1, min(limit if limit is not None else self.limit, 200))
|
||
episodes = list(await self.session.scalars(
|
||
select(Episode)
|
||
.where(
|
||
Episode.extraction_status.in_(CONSUMABLE_STATUSES),
|
||
Episode.retry_count < self.max_retry,
|
||
Episode.promoted_to_ltm.is_(False),
|
||
)
|
||
.order_by(Episode.id)
|
||
.limit(bounded)
|
||
))
|
||
result = EpisodeConsumptionResult()
|
||
for episode in episodes:
|
||
await self._consume(episode, result)
|
||
return result
|
||
|
||
async def _consume(self, episode: Episode, result: EpisodeConsumptionResult) -> None:
|
||
episode_id = int(episode.id)
|
||
text = self.user_content(episode.summary)
|
||
if not text:
|
||
# 片段里没有用户陈述（例如纯助手片段）：没有可抽取的事实，直接置为已处理。
|
||
await self._mark_done(episode, promoted=False)
|
||
result.no_fact.append(episode_id)
|
||
return
|
||
try:
|
||
extracted = await self.extractor.extract(message=text)
|
||
except Exception:
|
||
# 失败保持可重试：状态置失败并累加重试计数，记忆一行不写。
|
||
await self._mark_failed(episode)
|
||
logger.warning(
|
||
"episode extraction failed episode_id=%s retry_count=%s",
|
||
episode_id, episode.retry_count, exc_info=True,
|
||
)
|
||
result.failed.append(episode_id)
|
||
return
|
||
if extracted is None:
|
||
# 模型判定该片段没有持久事实：不是错误，但也没有可写的记忆。
|
||
await self._mark_done(episode, promoted=False)
|
||
result.no_fact.append(episode_id)
|
||
return
|
||
service = MemoryService(self.session, cache=self.cache)
|
||
memory = await service.upsert(
|
||
episode.customer_id,
|
||
extracted.memory_key,
|
||
extracted.value,
|
||
memory_type=extracted.memory_type,
|
||
confidence=extracted.confidence,
|
||
source_type=SOURCE_TYPE,
|
||
structured_value={
|
||
"memory_key": extracted.memory_key,
|
||
"value": extracted.value,
|
||
"memory_type": extracted.memory_type,
|
||
"confidence": extracted.confidence,
|
||
"session_id": episode.session_id,
|
||
"episode_uuid": episode.episode_uuid,
|
||
},
|
||
)
|
||
recorded = await service.record_evidence(
|
||
memory,
|
||
# 幂等键落在片段指纹上：同一片段重复消费不会写出第二条证据。
|
||
idempotency_key=self.idempotency_key(episode),
|
||
evidence_type=EVIDENCE_TYPE,
|
||
excerpt=text,
|
||
snapshot={
|
||
"episode_uuid": episode.episode_uuid,
|
||
"content_hash": episode.content_hash,
|
||
"customer_id": episode.customer_id,
|
||
"session_id": episode.session_id,
|
||
"start_message_no": episode.start_message_no,
|
||
"end_message_no": episode.end_message_no,
|
||
"memory_key": extracted.memory_key,
|
||
"value": extracted.value,
|
||
"confidence": extracted.confidence,
|
||
},
|
||
weight=EVIDENCE_WEIGHT,
|
||
source_table="episodes",
|
||
source_record_id=str(episode.id),
|
||
occurred_at=episode.ended_at or episode.created_at,
|
||
)
|
||
if recorded:
|
||
# ⚠️ 证据写入后必须把「画像重建」**投成事件**，不能直接调用重建：
|
||
# 本方法的记忆写入还在当前事务里、尚未提交，另开 session 去重建看不到
|
||
# 这条新记忆（`memory_extraction_worker` 里记录了这条实测结论）。
|
||
#
|
||
# 这条路径此前**完全没有**投重建事件，后果是：从会话片段抽取出来的记忆
|
||
# 永远到不了画像。实测证据 —— 客户 9001 的 `profile_snapshots` current
|
||
# 停在 `2026-09-10 13:57`(v7)，而它的 `memory_unit.updated_at` 已经是
|
||
# `2026-09-13 11:03`、`evidence_count` 涨到 4；那三条新证据的
|
||
# `source_table` 正是 `episodes`。也就是说**片段链路记住了，画像不知道**。
|
||
#
|
||
# 只在 `recorded=True` 时投：幂等命中（同一片段重复消费）时证据与计数
|
||
# 都没有净变化，投一次重建是白跑。
|
||
now = datetime.now(UTC).replace(tzinfo=None)
|
||
self.session.add(DomainEventOutbox(
|
||
id=0,
|
||
event_id=str(uuid4()),
|
||
event_type="profile.rebuild_requested",
|
||
aggregate_type="customer_profile",
|
||
aggregate_id=str(episode.customer_id),
|
||
trace_id=episode.episode_uuid,
|
||
payload={"customer_id": episode.customer_id, "trigger": "episode_extraction"},
|
||
status="pending",
|
||
retry_count=0,
|
||
occurred_at=now,
|
||
created_at=now,
|
||
updated_at=now,
|
||
))
|
||
await self._mark_done(episode, promoted=True)
|
||
result.extracted.append(episode_id)
|
||
|
||
async def _mark_done(self, episode: Episode, *, promoted: bool) -> None:
|
||
episode.extraction_status = STATUS_DONE
|
||
episode.promoted_to_ltm = promoted
|
||
await self.session.flush()
|
||
|
||
async def _mark_failed(self, episode: Episode) -> None:
|
||
episode.extraction_status = STATUS_FAILED
|
||
episode.retry_count += 1
|
||
await self.session.flush()
|
||
|
||
@staticmethod
|
||
def idempotency_key(episode: Episode) -> str:
|
||
"""片段级幂等键；`content_hash` 缺失（列可空）时退化为 episode_uuid / 主键。"""
|
||
fingerprint = episode.content_hash or episode.episode_uuid or str(episode.id)
|
||
return f"episode.extraction:{fingerprint}"
|
||
|
||
@staticmethod
|
||
def user_content(summary: str) -> str:
|
||
"""从片段摘要里取出**用户陈述**：助手回复不是记忆来源，不参与抽取。
|
||
|
||
摘要行格式由 `EpisodeWorker.summarize` 固定为 `[序号 | 角色] 正文`；
|
||
这里只按该格式取 `user` 行，取不到就当作"没有可抽取的用户事实"。
|
||
"""
|
||
marker = f"| {USER_ROLE}] "
|
||
parts: list[str] = []
|
||
for line in (summary or "").splitlines():
|
||
index = line.find(marker)
|
||
if index < 0:
|
||
continue
|
||
content = line[index + len(marker):].strip()
|
||
if content:
|
||
parts.append(content)
|
||
return "\n".join(parts)
|