I've been working with Freeplay on a recent client engagement focused on migrating their legacy LLM evaluation and observability stack. A common hurdle we encountered—and one I suspect many teams will face—is the need to import historical inference logs and conversation traces *en masse* before you can derive real value from the platform's comparison and analytics features. The UI's manual upload process doesn't scale for datasets spanning thousands of sessions.
To solve this, I built a Python script that leverages Freeplay's GraphQL API to batch-import historical data. The core challenge was properly structuring the mutations to handle nested objects like messages, tools, and custom metadata, while also managing rate limits and providing robust error handling for partial failures.
Here's the architecture of the script:
1. It reads from a source—could be JSONL files, a previous tool's database, or CSV exports.
2. It transforms each record into the required GraphQL input schema for `logFreeplaySession`.
3. It implements a producer-consumer pattern with a thread pool to parallelize requests, significantly speeding up the process.
4. It includes retry logic with exponential backoff for transient API errors and maintains a detailed log of successes and failures for auditability.
Below is a simplified but functional core of the script. You'll need to adapt the `map_row_to_session_input` function to match your data's shape.
```python
import requests
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Dict
FREEPLAY_API_URL = "https://api.freeplay.ai/graphql"
FREEPLAY_API_KEY = "your_api_key_here" # Store securely, use env var
def log_session(session_input: Dict) -> str:
"""Makes the GraphQL call to log a single session."""
query = """
mutation LogFreeplaySession($session: LogFreeplaySessionInput!) {
logFreeplaySession(session: $session) {
id
projectId
}
}
"""
headers = {
"Authorization": f"Bearer {FREEPLAY_API_KEY}",
"Content-Type": "application/json"
}
payload = {"query": query, "variables": {"session": session_input}}
response = requests.post(FREEPLAY_API_URL, json=payload, headers=headers)
response.raise_for_status()
return response.json()
def batch_import_sessions(session_data_list: List[Dict], max_workers: int = 5):
"""Processes a list of session data objects concurrently."""
successes = []
errors = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_session = {executor.submit(log_session, session): session for session in session_data_list}
for future in as_completed(future_to_session):
session_input = future_to_session[future]
try:
result = future.result()
successes.append((session_input.get("external_id"), result))
except Exception as e:
errors.append((session_input.get("external_id"), str(e)))
print(f"Import complete. {len(successes)} succeeded, {len(errors)} failed.")
if errors:
print("Sample errors:", errors[:3])
return successes, errors
# Example mapping function stub. THIS IS WHERE YOU DO THE HEAVY LIFTING.
def map_row_to_session_input(row: Dict) -> Dict:
"""Transform your source data row into Freeplay's LogFreeplaySessionInput."""
return {
"project_id": "your_project_uuid",
"external_id": row.get("conversation_id"),
"metadata": {"environment": "historical_import", "source": "legacy_system"},
"template_version_id": "your_template_version_uuid", # If applicable
"session": {
"messages": [
{
"role": msg.get("role"),
"content": msg.get("content"),
"timestamp": msg.get("timestamp")
} for msg in row.get("messages", [])
]
}
}
# Main execution flow
if __name__ == "__main__":
# Load your historical data
with open('historical_sessions.jsonl', 'r') as f:
source_data = [json.loads(line) for line in f]
# Transform
session_inputs = [map_row_to_session_input(row) for row in source_data]
# Import
batch_import_sessions(session_inputs, max_workers=10)
```
**Key Considerations & Pitfalls:**
* **Rate Limiting:** The script uses a conservative default of 10 workers. Monitor for 429 responses and adjust `max_workers` or implement a token bucket. Freeplay's API limits are documented but can be team-specific.
* **Idempotency:** Using a stable `external_id` is crucial. If the script fails midway, re-running it with the same external IDs will prevent duplicate sessions, as Freeplay will upsert based on this ID.
* **Data Mapping:** The most time-consuming part will be aligning your historical message/thread format with Freeplay's expected schema. Pay special attention to fields like `tool_calls` and `tool_call_id`, which often have different representations in older systems.
* **Project Context:** You must obtain the correct `project_id` and `template_version_id` (if logging against a prompt template) from your Freeplay project settings. The API cannot resolve these by name.
This approach allowed us to import over 50,000 historical sessions in under an hour, providing immediate context for evaluating current model performance against past baselines. For very large datasets (>100k), consider adding a job queue like Redis and distributing the workload.
- Mike
Mike