309a051038
数据库与模型：新增多版迁移（章节证据快照、对话血缘、记忆事实/时间线 lineage 等），把「成稿 ↔ 对话/记忆」的溯源信息落到表结构里。 业务链路：会话与 WS、回忆录/故事流水线、记忆写入与 enrichment 等跟着接上线索与快照；新增章节证据快照与评测侧 EvalTraceService 等模块，方便组评审用的证据包。 内部评测：自动化 run 与手工 memoir 评审共用可追溯证据；rubric/ judge 相关脚本与文档有配套调整。 app-eval-web：Memoir/实验详情里能展开看证据摘要与 evidence_trace（含对话轮次 id）；Vite 代理与 development.sh 注入的 API 端口与当前默认内部评测端口一致，避免改端口后页面连错服务。 工程杂项：GitHub Actions / 仓库说明有更新；各适配器与支付/配额/plan 等多处为小改动或跟随主改动的收尾；新增/扩充了?
91 lines
2.8 KiB
Python
91 lines
2.8 KiB
Python
"""conversation.lineage_schemas 单元测试。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
from app.features.conversation.lineage_schemas import (
|
|
DialogueLineage,
|
|
aggregate_lineage_from_segments,
|
|
merge_dialogue_lineages,
|
|
primary_user_message_id_from_lineage,
|
|
)
|
|
|
|
|
|
def test_for_single_turn_sets_primary() -> None:
|
|
ln = DialogueLineage.for_single_turn(
|
|
conversation_id="c1",
|
|
user_message_id="u1",
|
|
assistant_message_id="a1",
|
|
segment_ids=["s1"],
|
|
)
|
|
assert ln.primary_user_message_id == "u1"
|
|
assert ln.turns[0].assistant_message_id == "a1"
|
|
assert ln.segment_ids == ["s1"]
|
|
|
|
|
|
def test_primary_user_message_id_from_lineage_dict() -> None:
|
|
d = {
|
|
"schema_version": 1,
|
|
"conversation_id": "c",
|
|
"turns": [{"user_message_id": "x", "assistant_message_id": None}],
|
|
}
|
|
assert primary_user_message_id_from_lineage(d) == "x"
|
|
|
|
|
|
def test_merge_dialogue_lineages_dedupes_user_messages() -> None:
|
|
a = DialogueLineage.for_single_turn(
|
|
conversation_id="c1",
|
|
user_message_id="u1",
|
|
assistant_message_id="a1",
|
|
segment_ids=["s1"],
|
|
)
|
|
b = DialogueLineage.for_single_turn(
|
|
conversation_id="c1",
|
|
user_message_id="u1",
|
|
assistant_message_id="a2",
|
|
segment_ids=["s2"],
|
|
)
|
|
c = DialogueLineage.for_single_turn(
|
|
conversation_id="c1",
|
|
user_message_id="u2",
|
|
assistant_message_id="a3",
|
|
segment_ids=["s3"],
|
|
)
|
|
m = merge_dialogue_lineages([a, b, c], conversation_id="c1")
|
|
assert m is not None
|
|
assert [t.user_message_id for t in m.turns] == ["u1", "u2"]
|
|
|
|
|
|
def test_aggregate_lineage_from_segments_orders_and_merges() -> None:
|
|
segs = [
|
|
SimpleNamespace(
|
|
id="s2",
|
|
conversation_id="conv",
|
|
lineage_json=DialogueLineage.for_single_turn(
|
|
conversation_id="conv",
|
|
user_message_id="u2",
|
|
assistant_message_id="a2",
|
|
segment_ids=["s2"],
|
|
).model_dump(mode="json"),
|
|
user_message_id=None,
|
|
),
|
|
SimpleNamespace(
|
|
id="s1",
|
|
conversation_id="conv",
|
|
lineage_json=DialogueLineage.for_single_turn(
|
|
conversation_id="conv",
|
|
user_message_id="u1",
|
|
assistant_message_id="a1",
|
|
segment_ids=["s1"],
|
|
).model_dump(mode="json"),
|
|
user_message_id=None,
|
|
),
|
|
]
|
|
ordered = ["s1", "s2"]
|
|
order_map = {sid: i for i, sid in enumerate(ordered)}
|
|
segs.sort(key=lambda s: order_map[str(s.id)])
|
|
out = aggregate_lineage_from_segments(segs, conversation_id_fallback="conv")
|
|
assert out is not None
|
|
assert [t["user_message_id"] for t in out["turns"]] == ["u1", "u2"]
|