Spaces:
Sleeping
Sleeping
File size: 3,463 Bytes
910dadd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | """Export contimp-app traces from LangFuse as observation-level JSONL.
Produces rows shaped like LangFuse's Tracing-tab batch export (one JSON object
per observation, trace-level fields duplicated, input/output JSON-encoded) —
the format the oumi platform's LangFuse importer consumes.
uv run python scripts/export_traces.py --tag pr-area --limit 50 --out /tmp/export.jsonl
Auth via LANGFUSE_PUBLIC_KEY / LANGFUSE_SECRET_KEY / LANGFUSE_BASE_URL env vars.
"""
import argparse
import base64
import json
import os
import urllib.parse
import urllib.request
def api_get(path: str, params: dict | None = None) -> dict:
base = os.environ.get("LANGFUSE_BASE_URL", "https://cloud.langfuse.com").rstrip("/")
auth = base64.b64encode(
f"{os.environ['LANGFUSE_PUBLIC_KEY']}:{os.environ['LANGFUSE_SECRET_KEY']}".encode()
).decode()
query = f"?{urllib.parse.urlencode(params)}" if params else ""
req = urllib.request.Request(
f"{base}/api/public{path}{query}", headers={"Authorization": f"Basic {auth}"}
)
return json.loads(urllib.request.urlopen(req, timeout=60).read())
def encode(value: object) -> str | None:
if value is None:
return None
return value if isinstance(value, str) else json.dumps(value, ensure_ascii=False)
def export_trace(trace_id: str) -> list[dict]:
trace = api_get(f"/traces/{trace_id}")
rows = []
for obs in trace.get("observations", []):
rows.append({
"id": obs.get("id"),
"traceId": trace["id"],
"traceName": trace.get("name"),
"sessionId": trace.get("sessionId"),
"userId": trace.get("userId"),
"traceTags": trace.get("tags"),
"traceMetadata": trace.get("metadata"),
"traceScores": [
{"name": s.get("name"), "value": s.get("value"), "comment": s.get("comment")}
for s in trace.get("scores", [])
],
"type": obs.get("type"),
"name": obs.get("name"),
"startTime": obs.get("startTime"),
"endTime": obs.get("endTime"),
"parentObservationId": obs.get("parentObservationId"),
"model": obs.get("model"),
"input": encode(obs.get("input")),
"output": encode(obs.get("output")),
"metadata": obs.get("metadata"),
})
return rows
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--tag", help="filter traces by tag (e.g. a task id)")
parser.add_argument("--limit", type=int, default=100)
parser.add_argument("--out", default="/tmp/langfuse_export.jsonl")
args = parser.parse_args()
params: dict = {"limit": min(args.limit, 100), "page": 1}
if args.tag:
params["tags"] = args.tag
trace_ids: list[str] = []
while len(trace_ids) < args.limit:
page = api_get("/traces", params)
trace_ids += [t["id"] for t in page["data"]]
if params["page"] >= page["meta"]["totalPages"]:
break
params["page"] += 1
trace_ids = trace_ids[: args.limit]
with open(args.out, "w") as f:
n_rows = 0
for trace_id in trace_ids:
for row in export_trace(trace_id):
f.write(json.dumps(row, ensure_ascii=False) + "\n")
n_rows += 1
print(f"exported {len(trace_ids)} traces ({n_rows} observations) to {args.out}")
if __name__ == "__main__":
main()
|