Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 14 additions & 2 deletions scripts/beam/beam_ingestion.py
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,19 @@ def parse_time_anchor(time_anchor_str):
return parse_beam_time(time_anchor_str)


def session_time_anchor(messages, default="January-01-2024"):
"""Return the first non-empty ``time_anchor`` in a session, else ``default``.

BEAM-10M conversation 7 session 0 opens with null anchors; its first
anchor (index 244) is March-05-2020. Every other 10M session starts with
an anchor, so they keep the date of their first message.
"""
for msg in messages:
if isinstance(msg, dict) and msg.get("time_anchor"):
return msg["time_anchor"]
return default


_BEAM_RETAIN_MISSION = (
"Extract ALL factual claims the user makes about themselves, their project, "
"and their experience — including NEGATIVE statements (e.g. 'I have never done X', "
Expand Down Expand Up @@ -172,8 +185,7 @@ def ingest_conversation(conv, version, frame, success_records, f, clear=False):
if not session:
continue

first_msg = session[0]
session_date = parse_time_anchor(first_msg.get("time_anchor", "January-01-2024"))
session_date = parse_time_anchor(session_time_anchor(session))
session_id = f"{user_id}_session_{sess_idx}"

try:
Expand Down
7 changes: 3 additions & 4 deletions scripts/beam/beam_streaming.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
ingest_session,
load_beam_data,
parse_time_anchor,
session_time_anchor,
)
from beam.beam_search import parse_probing_questions # noqa: E402
from beam.beam_common import ( # noqa: E402
Expand Down Expand Up @@ -155,9 +156,7 @@ def build_graphiti_session_chunks(
) -> list[BeamGraphitiChunk]:
conv_id = str(conv["conversation_id"])
scale = conv.get("_scale", "unknown")
session_date = parse_time_anchor(
messages[0].get("time_anchor", "January-01-2024") if messages else "January-01-2024"
)
session_date = parse_time_anchor(session_time_anchor(messages))
timestamp = session_date.isoformat()

body_lines = [
Expand Down Expand Up @@ -310,7 +309,7 @@ def add_conversation_per_session(
session_chars = sum(len(msg.get("content", "")) for msg in messages)
total_messages += len(messages)
total_chars += session_chars
session_date = parse_time_anchor(messages[0].get("time_anchor", "January-01-2024"))
session_date = parse_time_anchor(session_time_anchor(messages))
session_id = f"{user_id}_session_{session_idx}"

if checkpoint_id in added_sessions:
Expand Down
20 changes: 20 additions & 0 deletions scripts/tests/unit/test_beam_time_anchor.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
"""Session date selection for BEAM ingestion (null leading time_anchor)."""

from datetime import datetime, timezone

from beam.beam_ingestion import parse_time_anchor, session_time_anchor


def test_first_message_anchor_is_used():
msgs = [{"time_anchor": "March-15-2024"}, {"time_anchor": "April-01-2024"}]
assert session_time_anchor(msgs) == "March-15-2024"


def test_null_leading_anchor_uses_first_non_null():
msgs = [{"time_anchor": None}] * 3 + [{"time_anchor": "March-05-2020"}, {"time_anchor": "June-01-2020"}]
assert parse_time_anchor(session_time_anchor(msgs)) == datetime(2020, 3, 5, tzinfo=timezone.utc)


def test_missing_or_empty_anchors_fall_back_to_default():
assert session_time_anchor([{"role": "user"}, {"time_anchor": ""}]) == "January-01-2024"
assert session_time_anchor([]) == "January-01-2024"
Loading