Spaces:
Running on Zero
Running on Zero
File size: 3,653 Bytes
8fb1ae9 13fe947 8fb1ae9 13fe947 8fb1ae9 13fe947 8fb1ae9 fbdb1e5 8fb1ae9 e12a049 8fb1ae9 e12a049 8fb1ae9 fbdb1e5 | 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 | from __future__ import annotations
import json
from typing import Any
from hackathon_advisor._text import utc_now
TRACE_SCHEMA_VERSION = 1
def build_trace_jsonl(session: dict[str, Any], metadata: dict[str, Any]) -> str:
trace = session.get("trace") or []
ideas = session.get("ideas") or []
records = [
{
"type": "trace_manifest",
"schema_version": TRACE_SCHEMA_VERSION,
"generated_at": utc_now(),
"app": "hackathon-advisor",
"index": {
"algorithm": metadata.get("index_algorithm", ""),
"snapshot_generated_at": metadata.get("snapshot_generated_at", ""),
"index_generated_at": metadata.get("index_generated_at", ""),
"snapshot_digest": metadata.get("snapshot_digest", ""),
},
"idea_count": len(ideas),
"turn_count": len(trace),
}
]
for index, event in enumerate(trace, start=1):
records.append(
{
"type": "agent_turn",
"schema_version": TRACE_SCHEMA_VERSION,
"turn_index": index,
"input": str(event.get("input") or ""),
"tools": _tools(event),
"verdict": str(event.get("verdict") or ""),
"overall": event.get("overall"),
"plan_steps": int(event.get("plan_steps") or 0),
"artifact_title": str(event.get("artifact_title") or ""),
"response": str(event.get("response") or ""),
"tool_resolution": _tool_resolution(event),
}
)
return "\n".join(json.dumps(record, ensure_ascii=False, sort_keys=True) for record in records) + "\n"
def trace_metadata(index: Any) -> dict[str, str]:
metadata = {
"snapshot_generated_at": index.generated_at,
"index_generated_at": index.index_generated_at,
"index_algorithm": index.index_algorithm,
"snapshot_digest": index.snapshot_digest,
}
embedding = getattr(index, "embedding_metadata", None)
if isinstance(embedding, dict):
metadata.update(
{
"embedding_model_repo": str(embedding.get("model_repo") or ""),
"embedding_model_file": str(embedding.get("model_file") or ""),
"embedding_runtime": str(embedding.get("runtime") or ""),
"embedding_build_source": str(embedding.get("build_source") or ""),
"embedding_dimensions": str(embedding.get("dimensions") or ""),
"embedding_builder": str(embedding.get("builder") or ""),
"embedding_modal_app": str(embedding.get("modal_app") or ""),
}
)
return metadata
def _tools(event: dict[str, Any]) -> list[dict[str, str]]:
tools = event.get("tools") or []
return [
{
"name": str(tool.get("name") or ""),
"summary": str(tool.get("summary") or ""),
}
for tool in tools
if isinstance(tool, dict)
]
def _tool_resolution(event: dict[str, Any]) -> dict[str, Any]:
resolution = event.get("tool_resolution") or {}
call = resolution.get("call") if isinstance(resolution, dict) else {}
return {
"status": str(resolution.get("status") or "") if isinstance(resolution, dict) else "",
"call": {
"name": str(call.get("name") or "") if isinstance(call, dict) else "",
"arguments": call.get("arguments") if isinstance(call, dict) else {},
},
"errors": list(resolution.get("errors") or []) if isinstance(resolution, dict) else [],
}
|