Adds conversation memory for a single line of data to MemOS and returns the data with a persistent user_id.
(
line_data: tuple,
mem_client,
num_irrelevant_turns: int,
lib: str,
version: str,
success_records,
f,
)
| 28 | |
| 29 | |
| 30 | def add_memory_for_line( |
| 31 | line_data: tuple, |
| 32 | mem_client, |
| 33 | num_irrelevant_turns: int, |
| 34 | lib: str, |
| 35 | version: str, |
| 36 | success_records, |
| 37 | f, |
| 38 | ) -> dict: |
| 39 | """ |
| 40 | Adds conversation memory for a single line of data to MemOS and returns the data with a persistent user_id. |
| 41 | """ |
| 42 | i, line = line_data |
| 43 | user_id = f"{lib}_user_pref_eval_{i}_{version}" |
| 44 | |
| 45 | try: |
| 46 | original_data = json.loads(line) |
| 47 | conversation = original_data.get("conversation", []) |
| 48 | |
| 49 | if num_irrelevant_turns == 10: |
| 50 | conversation = conversation + irre_10 |
| 51 | elif num_irrelevant_turns == 300: |
| 52 | conversation = conversation + irre_300 |
| 53 | |
| 54 | start_time_add = time.monotonic() |
| 55 | |
| 56 | for idx, _ in enumerate(conversation[::2]): |
| 57 | msg_idx = idx * 2 |
| 58 | record_id = f"{lib}_user_pref_eval_{i}_{version}_{msg_idx!s}" |
| 59 | |
| 60 | if record_id not in success_records: |
| 61 | mem_client.add( |
| 62 | messages=conversation[msg_idx : msg_idx + 2], |
| 63 | user_id=user_id, |
| 64 | ) |
| 65 | f.write(f"{record_id}\n") |
| 66 | f.flush() |
| 67 | |
| 68 | end_time_add = time.monotonic() |
| 69 | add_duration = end_time_add - start_time_add |
| 70 | |
| 71 | original_data["user_id"] = user_id |
| 72 | original_data["metrics"] = {"add_memories_duration_seconds": add_duration} |
| 73 | return original_data |
| 74 | |
| 75 | except Exception as e: |
| 76 | print(f"Error adding memory for line {i + 1} (user_id: {user_id}): {e}") |
| 77 | return None |
| 78 | |
| 79 | |
| 80 | def search_memory_for_line(line_data: tuple, mem_client, top_k_value: int) -> dict: |