feat(chat): 低信息短答主动续话；修复本地 dev 环境与迁移链

- interview_turn_plan: 识别低信息短回复，引导 AI 承接后主动追问新话题 - development.sh / docker-compose.dev: Postgres/Redis 端口与 .env 对齐，补充宿主机端口监听检查 - Alembic: 补回 0016 memory pipeline status、0017 segment narrative defer - app-expo: api/ws URL 去掉末尾斜杠，避免 WS 双斜杠；更新 .env.staging Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-11 12:06:17 +08:00
parent 55cfbc7f80
commit 705fe951b3
8 changed files with 427 additions and 15 deletions
--- a/api/alembic/versions/0016_memory_pipeline_status.py
+++ b/api/alembic/versions/0016_memory_pipeline_status.py
@@ -0,0 +1,109 @@
+"""Memory pipeline status columns.
+
+Revision ID: 0016_memory_pipeline_status
+Revises: 0015_memory_single_chain
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "0016_memory_pipeline_status"
+down_revision: Union[str, None] = "0015_memory_single_chain"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def _has_column(table: str, column: str) -> bool:
+    bind = op.get_bind()
+    return any(c["name"] == column for c in sa.inspect(bind).get_columns(table))
+
+
+def upgrade() -> None:
+    if not _has_column("memory_sources", "embedding_status"):
+        op.add_column(
+            "memory_sources",
+            sa.Column("embedding_status", sa.String(), nullable=True),
+        )
+    if not _has_column("memory_sources", "embedding_error"):
+        op.add_column(
+            "memory_sources",
+            sa.Column("embedding_error", sa.Text(), nullable=True),
+        )
+    if not _has_column("memory_sources", "enrichment_status"):
+        op.add_column(
+            "memory_sources",
+            sa.Column("enrichment_status", sa.String(), nullable=True),
+        )
+    if not _has_column("memory_sources", "enrichment_error"):
+        op.add_column(
+            "memory_sources",
+            sa.Column("enrichment_error", sa.Text(), nullable=True),
+        )
+    if not _has_column("memory_chunks", "embedding_status"):
+        op.add_column(
+            "memory_chunks",
+            sa.Column("embedding_status", sa.String(), nullable=True),
+        )
+    if not _has_column("memory_chunks", "embedding_error"):
+        op.add_column(
+            "memory_chunks",
+            sa.Column("embedding_error", sa.Text(), nullable=True),
+        )
+
+    op.execute(
+        """
+        UPDATE memory_sources
+        SET
+          embedding_status = COALESCE(embedding_status, 'success'),
+          enrichment_status = COALESCE(enrichment_status, 'pending')
+        """
+    )
+    op.execute(
+        """
+        UPDATE memory_chunks
+        SET embedding_status = COALESCE(
+          embedding_status,
+          CASE WHEN embedding IS NULL THEN 'pending' ELSE 'success' END
+        )
+        """
+    )
+    op.create_index(
+        "ix_memory_sources_user_embedding_status",
+        "memory_sources",
+        ["user_id", "embedding_status"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_memory_sources_user_enrichment_status",
+        "memory_sources",
+        ["user_id", "enrichment_status"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_memory_chunks_user_embedding_status",
+        "memory_chunks",
+        ["user_id", "embedding_status"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_memory_chunks_user_embedding_status", table_name="memory_chunks")
+    op.drop_index(
+        "ix_memory_sources_user_enrichment_status", table_name="memory_sources"
+    )
+    op.drop_index("ix_memory_sources_user_embedding_status", table_name="memory_sources")
+    for column in ("embedding_error", "embedding_status"):
+        if _has_column("memory_chunks", column):
+            op.drop_column("memory_chunks", column)
+    for column in (
+        "enrichment_error",
+        "enrichment_status",
+        "embedding_error",
+        "embedding_status",
+    ):
+        if _has_column("memory_sources", column):
+            op.drop_column("memory_sources", column)
--- a/api/alembic/versions/0017_segment_narrative_defer.py
+++ b/api/alembic/versions/0017_segment_narrative_defer.py
@@ -0,0 +1,75 @@
+"""segments：Phase2 低置信路由延迟元数据
+
+Revision ID: 0017_segment_narrative_defer
+Revises: 0016_memory_pipeline_status
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "0017_segment_narrative_defer"
+down_revision: Union[str, None] = "0016_memory_pipeline_status"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def _column_names(table_name: str) -> set[str]:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    return {column["name"] for column in inspector.get_columns(table_name)}
+
+
+def upgrade() -> None:
+    columns = _column_names("segments")
+    if "narrative_deferred_until" not in columns:
+        op.add_column(
+            "segments",
+            sa.Column(
+                "narrative_deferred_until",
+                sa.DateTime(timezone=True),
+                nullable=True,
+            ),
+        )
+    if "narrative_defer_count" not in columns:
+        op.add_column(
+            "segments",
+            sa.Column(
+                "narrative_defer_count",
+                sa.Integer(),
+                nullable=False,
+                server_default=sa.text("0"),
+            ),
+        )
+    if "narrative_defer_reason" not in columns:
+        op.add_column(
+            "segments",
+            sa.Column(
+                "narrative_defer_reason",
+                sa.String(),
+                nullable=True,
+            ),
+        )
+    if "narrative_last_attempt_at" not in columns:
+        op.add_column(
+            "segments",
+            sa.Column(
+                "narrative_last_attempt_at",
+                sa.DateTime(timezone=True),
+                nullable=True,
+            ),
+        )
+
+
+def downgrade() -> None:
+    columns = _column_names("segments")
+    for column in (
+        "narrative_last_attempt_at",
+        "narrative_defer_reason",
+        "narrative_defer_count",
+        "narrative_deferred_until",
+    ):
+        if column in columns:
+            op.drop_column("segments", column)
--- a/api/app/agents/chat/interview_turn_plan.py
+++ b/api/app/agents/chat/interview_turn_plan.py
@@ -14,7 +14,7 @@ from dataclasses import dataclass
 from typing import Literal

 from app.agents.chat.prompts_conversation import SLOT_NAME_MAP
-from app.agents.stage_constants import STAGE_SLOT_KEYS
+from app.agents.stage_constants import STAGE_KEYWORD_WEIGHTS, STAGE_SLOT_KEYS

 InterviewTurnMode = Literal[
    "emotion_first",
@@ -66,6 +66,7 @@ class InterviewTurnPlan:
    secondary_focus: FocusPrimary | None = None
    focus_summary: str = ""
    focus_source: FocusSource = "rule"
+    low_information_reply: bool = False

    # ---- 语义属性：供 prompt_layers / interview_agent 等调用方消费，禁止重复立法 ----

@@ -323,6 +324,17 @@ _ASSISTANT_IDENTITY_QUESTION_MARKERS: tuple[str, ...] = (
    "你也是",
 )

+_LOW_INFORMATION_REPLY_MAX_CHARS = 8
+_LOW_INFORMATION_REPLY_CHARS: frozenset[str] = frozenset(
+    # 这不是短语白名单，而是一组低信息的应声/确认/语气字符。
+    # 只要短回复里出现不在此集合中的字，就会被视为有潜在叙事信号。
+    "嗯唔呃哦噢喔啊呀呢嘛吧哈"
+    "对是的了好行可允许以"
+    "没错确实不否无有还太很挺就"
+    "什么这样那样当当然"
+)
+_ACK_STRIP_CHARS = " \t\r\n，。！？!?、,.；;：:~～…"
+

 def _is_emotion_heavy(text: str) -> bool:
    t = (text or "").strip()
@@ -351,6 +363,32 @@ def _is_ambiguous_or_needs_slow_pace(text: str) -> bool:
    return False


+def _normalized_short_reply(text: str) -> str:
+    return "".join(ch for ch in (text or "").strip() if ch not in _ACK_STRIP_CHARS)
+
+
+def _has_substantive_short_reply_signal(compact: str) -> bool:
+    """短回复里的叙事实质信号：地点/人事词、阶段关键词、年份数字等。"""
+    if any(ch.isdigit() for ch in compact):
+        return True
+    if any(("a" <= ch.lower() <= "z") for ch in compact):
+        return True
+    for weighted_keywords in STAGE_KEYWORD_WEIGHTS.values():
+        if any(keyword and keyword in compact for keyword, _ in weighted_keywords):
+            return True
+    return any(ch not in _LOW_INFORMATION_REPLY_CHARS for ch in compact)
+
+
+def _is_low_information_reply(text: str) -> bool:
+    """识别短、低信息、没有新增叙事素材的确认/应声回复。"""
+    compact = _normalized_short_reply(text)
+    if not compact:
+        return False
+    if len(compact) > _LOW_INFORMATION_REPLY_MAX_CHARS:
+        return False
+    return not _has_substantive_short_reply_signal(compact)
+
+
 def _is_too_vague_for_memoir_push(text: str) -> bool:
    """过短或仍含糊时，不进入 memoir_push。"""
    t = (text or "").strip()
@@ -396,6 +434,7 @@ def plan_interview_turn(
    )
    um = (normalized_user_message or "").strip()
    asks_assistant_identity = _is_asking_assistant_identity_or_life(um)
+    low_information_reply = _is_low_information_reply(um)
    reply_shape: ReplyShape = "flexible"
    if any(
        k in um
@@ -403,6 +442,45 @@ def plan_interview_turn(
    ):
        reply_shape = "ack_then_question"

+    if low_information_reply:
+        slot = primary_empty_slot(current_stage, empty_slots)
+        if slot:
+            readable = SLOT_NAME_MAP.get(slot, slot)
+            return InterviewTurnPlan(
+                mode="memoir_push",
+                anchor_slot_key=slot,
+                anchor_slot_readable=readable,
+                anchor_snippet=snippet,
+                anchor_source_kind=anchor_source_kind,
+                assistant_identity_question=asks_assistant_identity,
+                memory_usage=mem_use,
+                reply_shape="ack_then_question",
+                primary_focus=_focus_primary_for_mode("memoir_push"),
+                focus_summary=(
+                    f"用户只做了简短确认；短接一句后，不澄清“{um}”，"
+                    f"主动从「{readable}」打开一个具体、好回答的新回忆话题"
+                ),
+                focus_source="rule",
+                low_information_reply=True,
+            )
+        return InterviewTurnPlan(
+            mode="follow_user_only",
+            anchor_slot_key=None,
+            anchor_slot_readable="（本阶段主要叙述槽已有素材）请回到上文最近的人/事/地方或情绪线，主动打开一个纵深问题",
+            anchor_snippet=snippet,
+            anchor_source_kind=anchor_source_kind,
+            assistant_identity_question=asks_assistant_identity,
+            memory_usage=mem_use,
+            reply_shape="ack_then_question",
+            primary_focus=_focus_primary_for_mode("follow_user_only"),
+            focus_summary=(
+                f"用户只做了简短确认；短接一句后，不澄清“{um}”，"
+                "回到上文最近的具体线索，主动递一个新的回忆追问"
+            ),
+            focus_source="rule",
+            low_information_reply=True,
+        )
+
    if _is_emotion_heavy(normalized_user_message):
        slot = primary_empty_slot(current_stage, empty_slots)
        readable = (
@@ -534,6 +612,17 @@ def format_interview_turn_directive_block(plan: InterviewTurnPlan) -> str:
    else:
        shape_block = ""

+    ack_block = ""
+    if plan.low_information_reply:
+        ack_block = (
+            "- **低信息短回复处理**：用户本轮只是简短确认、应声或泛泛回应，没有新增叙事素材。"
+            "不要把这个短答本身当成需要澄清的内容，不要反复追问「你是说……吗」，"
+            "也不要停在原地等用户继续补充。\n"
+            "- 先用半句自然接住，再主动从「主追问方向」、上文最近的具体人/事/地方，"
+            "或过往记忆线索里挑一个**具体、好回答**的新回忆话题；若挂钩线索为空，"
+            "允许直接借当前阶段未聊方向起问，但禁止编造用户没说过的细节。\n"
+        )
+
    if plan.mode == "emotion_first":
        mode_rules = (
            "- **情绪优先**：本轮以承接、并肩与安全感为主，**不要**为推进「叙述槽大纲」而追加信息搜集型追问。\n"
@@ -557,13 +646,29 @@ def format_interview_turn_directive_block(plan: InterviewTurnPlan) -> str:
            "- **跟话头**：本轮禁止问卷式首开、禁止重启式盘点；顺着用户刚展开的画面、人物或情绪自然往下。\n"
            "- 若带问句，最多**一个**，且必须**从用户原词或下面摘录**长出来，禁止空泛「还有吗」。"
        )
+        if plan.low_information_reply:
+            mode_rules = (
+                "- **跟话头 + 主动续话**：用户本轮只是简短确认，没有新素材；"
+                "不要围着短答本身澄清，也不要重复上一问等对方补充。\n"
+                "- 若带问句，最多**一个**，优先从上文最近的具体人/事/地方或下面摘录长出来；"
+                "若无可用摘录，就从当前阶段已聊内容的纵深处打开一个新回忆话题。"
+            )
    else:
-        mode_rules = (
-            "- **回忆推进（memoir_push）**：对用户可见回复中须有**恰好一个**开放式回忆追问，\n"
-            "  且意图明显在补足下面「主追问方向」；问句必须挂住**用户本轮原词**或**挂钩摘录**（二者至少其一，且**优先原词**）。\n"
-            "- **禁止**把用户仍含糊的表达改写成确定事实或关系；禁止用日常寒暄替代该追问；"
-            "仍遵守全篇「最多一个问句」「禁止晚会硬切」。"
-        )
+        if plan.low_information_reply:
+            mode_rules = (
+                "- **回忆推进（memoir_push）**：用户本轮只是简短确认，"
+                "对用户可见回复中须有**恰好一个**开放式回忆追问，主动把话头往下递。\n"
+                "  追问可挂住**上文最近具体细节**、**主追问方向**或**挂钩摘录**；"
+                "不要要求从低信息短答里抽词。\n"
+                "- 禁止用日常寒暄替代该追问；仍遵守全篇「最多一个问句」「禁止晚会硬切」。"
+            )
+        else:
+            mode_rules = (
+                "- **回忆推进（memoir_push）**：对用户可见回复中须有**恰好一个**开放式回忆追问，\n"
+                "  且意图明显在补足下面「主追问方向」；问句必须挂住**用户本轮原词**或**挂钩摘录**（二者至少其一，且**优先原词**）。\n"
+                "- **禁止**把用户仍含糊的表达改写成确定事实或关系；禁止用日常寒暄替代该追问；"
+                "仍遵守全篇「最多一个问句」「禁止晚会硬切」。"
+            )

    focus_block = _focus_directive_lines(plan)

@@ -576,7 +681,7 @@ def format_interview_turn_directive_block(plan: InterviewTurnPlan) -> str:

    return f"""## 本轮编排指令（硬规则，优先于后文一般性建议）
 {mode_rules}
-{focus_block}{mem_block}{perspective_block}{shape_block}- **主追问方向（叙述槽）**：{plan.anchor_slot_readable}
+{focus_block}{ack_block}{mem_block}{perspective_block}{shape_block}- **主追问方向（叙述槽）**：{plan.anchor_slot_readable}
 - **挂钩线索**{anchor_label}：{snippet_line}
 """

--- a/api/development.sh
+++ b/api/development.sh
@@ -212,6 +212,66 @@ get_effective_database_url() {
  return 1
 }

+get_effective_redis_url() {
+  if [[ -n "${REDIS_URL:-}" ]]; then
+    printf '%s\n' "${REDIS_URL}"
+    return 0
+  fi
+
+  if [[ -f "${ROOT_DIR}/.env" ]]; then
+    local line
+    line="$(sed -n 's/^REDIS_URL=//p' "${ROOT_DIR}/.env" | sed -n '1p')"
+    line="${line%\"}"
+    line="${line#\"}"
+    line="${line%\'}"
+    line="${line#\'}"
+    if [[ -n "${line}" ]]; then
+      printf '%s\n' "${line}"
+      return 0
+    fi
+  fi
+
+  return 1
+}
+
+extract_url_port() {
+  local url="$1"
+  local default_port="$2"
+
+  if [[ "${url}" =~ :([0-9]+)(/|\?|$) ]]; then
+    printf '%s\n' "${BASH_REMATCH[1]}"
+    return 0
+  fi
+
+  printf '%s\n' "${default_port}"
+}
+
+wait_host_infra_ready() {
+  local database_url redis_url pg_port redis_port
+
+  if ! database_url="$(get_effective_database_url)"; then
+    print_warn "无法解析 DATABASE_URL，跳过宿主机 PostgreSQL 端口检查"
+  else
+    pg_port="$(extract_url_port "${database_url}" "5432")"
+    if wait_for_tcp_listener "$$" "${pg_port}" 12; then
+      print_ok "宿主机 PostgreSQL 端口已监听 (:${pg_port})"
+    else
+      print_warn "宿主机 PostgreSQL 端口未监听 (:${pg_port})；请检查 .env 与 docker-compose.dev.yml 端口映射"
+    fi
+  fi
+
+  if ! redis_url="$(get_effective_redis_url)"; then
+    print_warn "无法解析 REDIS_URL，跳过宿主机 Redis 端口检查"
+  else
+    redis_port="$(extract_url_port "${redis_url}" "6379")"
+    if wait_for_tcp_listener "$$" "${redis_port}" 12; then
+      print_ok "宿主机 Redis 端口已监听 (:${redis_port})"
+    else
+      print_warn "宿主机 Redis 端口未监听 (:${redis_port})；请检查 .env 与 docker-compose.dev.yml 端口映射"
+    fi
+  fi
+}
+
 warn_database_url_host_pitfall() {
  local database_url
  local host
@@ -562,6 +622,7 @@ main() {
  ensure_venv
  ensure_dotenv_from_development
  check_env_file
+  wait_host_infra_ready
  run_migrations
  start_services

--- a/api/docker-compose.dev.yml
+++ b/api/docker-compose.dev.yml
@@ -7,7 +7,7 @@ services:
    image: pgvector/pgvector:pg17
    container_name: life-echo-postgres-dev
    ports:
-      - "5432:5432"
+      - "48291:5432"
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
@@ -26,7 +26,7 @@ services:
    image: redis:7-alpine
    container_name: life-echo-redis-dev
    ports:
-      - "6379:6379"
+      - "48307:6379"
    volumes:
      - redis_data_dev:/data
    command: redis-server --appendonly yes
--- a/api/tests/test_interview_turn_plan.py
+++ b/api/tests/test_interview_turn_plan.py
@@ -181,13 +181,66 @@ def test_plan_clarify_first_when_very_short():
    p = plan_interview_turn(
        current_stage="childhood",
        empty_slots=["place"],
-        normalized_user_message="还好吧",
+        normalized_user_message="有点说不清",
        memory_evidence_text="",
        stage_switched_this_turn=False,
    )
    assert p.mode == "clarify_first"


+def test_plan_low_information_reply_pushes_next_topic_when_slots_remain():
+    p = plan_interview_turn(
+        current_stage="childhood",
+        empty_slots=["place", "people"],
+        normalized_user_message="嗯。",
+        memory_evidence_text="",
+        stage_switched_this_turn=False,
+    )
+    assert p.mode == "memoir_push"
+    assert p.anchor_slot_key == "place"
+    assert p.reply_shape == "ack_then_question"
+    assert p.low_information_reply is True
+
+
+def test_low_information_directive_asks_for_proactive_topic_not_clarification():
+    p = plan_interview_turn(
+        current_stage="childhood",
+        empty_slots=["place"],
+        normalized_user_message="对",
+        memory_evidence_text="",
+        stage_switched_this_turn=False,
+    )
+    block = format_interview_turn_directive_block(p)
+    assert "低信息短回复处理" in block
+    assert "不要把这个短答本身当成需要澄清的内容" in block
+    assert "恰好一个" in block
+    assert "主动" in block
+
+
+def test_plan_low_information_reply_uses_follow_when_no_slots_remain():
+    p = plan_interview_turn(
+        current_stage="childhood",
+        empty_slots=[],
+        normalized_user_message="是的",
+        memory_evidence_text="",
+        stage_switched_this_turn=False,
+    )
+    assert p.mode == "follow_user_only"
+    assert p.reply_shape == "ack_then_question"
+    assert p.low_information_reply is True
+
+
+def test_short_substantive_reply_is_not_treated_as_low_information():
+    p = plan_interview_turn(
+        current_stage="childhood",
+        empty_slots=["place", "people"],
+        normalized_user_message="上海",
+        memory_evidence_text="",
+        stage_switched_this_turn=False,
+    )
+    assert p.low_information_reply is False
+
+
 def test_plan_memoir_push():
    p = plan_interview_turn(
        current_stage="childhood",
@@ -232,3 +285,4 @@ def test_plan_follow_when_no_empty_slots():
        stage_switched_this_turn=False,
    )
    assert p.mode == "follow_user_only"
+    assert p.low_information_reply is True