| """Log user interactions to a HuggingFace Hub dataset.""" |
|
|
| import json |
| import os |
| import uuid |
| from pathlib import Path |
|
|
| from huggingface_hub import HfApi, HfFileSystem |
|
|
| DATASET_REPO = "EarthSpeciesProject/naturelm-audio-space-logs" |
| SPLIT = "test" |
| TESTING = os.getenv("TESTING", "0") == "1" |
|
|
| _hf_token = os.getenv("HF_TOKEN", None) |
| _api = HfApi(token=_hf_token) |
| _fs = HfFileSystem(token=_hf_token) |
|
|
| _METADATA_PATH = f"datasets/{DATASET_REPO}/{SPLIT}/metadata.jsonl" |
|
|
|
|
| def log_interaction( |
| audio: str | Path, |
| user_text: str, |
| model_response: str, |
| session_id: str = "", |
| model_version: str = "", |
| ) -> None: |
| """Log a single user/model exchange (audio + text) to the Hub dataset. |
| |
| Parameters |
| ---------- |
| audio : str | Path |
| Local path to the audio file uploaded by the user. |
| user_text : str |
| The user's query text. |
| model_response : str |
| The model's generated response. |
| session_id : str |
| Unique identifier for the user session. |
| model_version : str |
| Version string of the model that produced the response. |
| """ |
| data_id = str(uuid.uuid4()) |
|
|
| if TESTING: |
| data_id = "test-" + data_id |
| session_id = "test-" + session_id |
|
|
| suffix = Path(audio).suffix |
| audio_repo_path = f"{SPLIT}/audio/{session_id}{suffix}" |
|
|
| if not _fs.exists(f"datasets/{DATASET_REPO}/{audio_repo_path}"): |
| _api.upload_file( |
| path_or_fileobj=str(audio), |
| path_in_repo=audio_repo_path, |
| repo_id=DATASET_REPO, |
| repo_type="dataset", |
| ) |
|
|
| record = { |
| "user_message": user_text, |
| "model_response": model_response, |
| "file_name": f"audio/{session_id}{suffix}", |
| "original_fn": Path(audio).name, |
| "id": data_id, |
| "session_id": session_id, |
| "model_version": model_version, |
| } |
|
|
| line = json.dumps(record) + "\n" |
|
|
| |
| if _fs.exists(_METADATA_PATH): |
| with _fs.open(_METADATA_PATH, "r") as f: |
| lines = f.readlines() |
| lines.append(line) |
| with _fs.open(_METADATA_PATH, "w") as f: |
| f.writelines(lines) |
| else: |
| with _fs.open(_METADATA_PATH, "w") as f: |
| f.write(line) |
|
|
