Spaces:
Sleeping
Sleeping
File size: 4,499 Bytes
116524e | 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 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | """OpenClawToTraceStep — convert raw JSONL events to a structured trace dict."""
from __future__ import annotations
from typing import Any
from ...core.context import ACEStepContext
class OpenClawToTraceStep:
"""Convert raw OpenClaw JSONL events into a structured trace dict.
This step receives ``ctx.trace`` as a ``list[dict]`` of raw JSONL events
(placed by ``LoadTracesStep``) and converts them into the trace dict
format expected by ``ReflectStep``::
{
"question": str, # reconstructed conversation
"reasoning": str, # full execution trace (thinking + tool calls)
"answer": str, # last assistant text
"skill_ids": list, # always [] for OpenClaw
"feedback": str, # session summary
"ground_truth": None,
}
Follows the same pattern as ``BrowserToTrace``, ``LangChainToTrace``,
and ``ClaudeCodeToTrace``.
"""
requires = frozenset({"trace"})
provides = frozenset({"trace"})
def __call__(self, ctx: ACEStepContext) -> ACEStepContext:
events: list[dict[str, Any]] = ctx.trace # type: ignore[assignment]
if not events:
return ctx
trace_dict = _events_to_trace(events)
return ctx.replace(trace=trace_dict)
def _events_to_trace(events: list[dict[str, Any]]) -> dict[str, Any]:
"""Convert a list of OpenClaw JSONL events into the standardised trace dict."""
user_messages: list[str] = []
assistant_texts: list[str] = []
reasoning_parts: list[str] = []
model = ""
total_tokens = 0
for event in events:
etype = event.get("type")
if etype == "session":
model = event.get("cwd", "")
continue
if etype == "custom":
data = event.get("data", {})
if data.get("modelId"):
model = data["modelId"]
continue
if etype != "message":
continue
msg = event.get("message", {})
role = msg.get("role")
content_blocks = msg.get("content", [])
# Track token usage
usage = msg.get("usage", {})
total_tokens += usage.get("totalTokens", 0)
# Track model
if msg.get("model"):
model = msg["model"]
if role == "user":
for block in content_blocks:
if block.get("type") == "text":
user_messages.append(block["text"])
elif role == "assistant":
for block in content_blocks:
btype = block.get("type")
if btype == "thinking":
reasoning_parts.append(f"[thinking] {block.get('thinking', '')}")
elif btype == "text":
text = block.get("text", "")
assistant_texts.append(text)
reasoning_parts.append(f"[response] {text}")
elif btype == "toolCall":
name = block.get("name", "unknown")
args = block.get("arguments", {})
reasoning_parts.append(f"[tool:{name}] {args}")
elif role == "toolResult":
tool_name = msg.get("toolName", "unknown")
for block in content_blocks:
if block.get("type") == "text":
text = block["text"]
# Truncate long tool results
if len(text) > 500:
text = text[:500] + "..."
reasoning_parts.append(f"[tool_result:{tool_name}] {text}")
# Build the conversation as the "question"
question = "\n\n".join(f"User: {m}" for m in user_messages) if user_messages else ""
# Last assistant text as the "answer"
answer = assistant_texts[-1] if assistant_texts else ""
# Build feedback summary
n_user = len(user_messages)
n_assistant = len(assistant_texts)
feedback = (
f"OpenClaw session: {n_user} user messages, {n_assistant} assistant responses"
)
if model:
feedback += f", model: {model}"
if total_tokens:
feedback += f", {total_tokens} tokens"
return {
"question": question,
"reasoning": "\n".join(reasoning_parts),
"answer": answer,
"skill_ids": [],
"feedback": feedback,
"ground_truth": None,
}
|