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