Spaces:
Running on Zero
Running on Zero
File size: 9,252 Bytes
56f6a56 fdbf570 56f6a56 fdbf570 56f6a56 fdbf570 56f6a56 | 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 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 | """OpenClaude-specific prompting and message normalization for the Space.
The Space owns this adapter so notebook clients can connect directly to its
OpenAI-compatible endpoint. No conversation state is stored in the process;
all decisions are reconstructed from the request history.
"""
from __future__ import annotations
import os
import re
from collections.abc import Mapping
from typing import Any
from tool_calls import normalize_openai_tool_arguments
TOOL_PROTOCOL_MARKER = "OPENAI TOOL CALL FORMAT — MANDATORY"
TOOL_RECAP_CHARACTERS = int(os.getenv("TOOL_RECAP_CHARACTERS", "6000"))
SYSTEM_REMINDER_RE = re.compile(
r"<system-reminder\b[^>]*>.*?</system-reminder>",
re.DOTALL | re.IGNORECASE,
)
def _content_text(content: Any) -> str:
if isinstance(content, str):
return content
if isinstance(content, list):
return "\n".join(
str(block.get("text", ""))
for block in content
if isinstance(block, Mapping)
and block.get("type") in {"text", "input_text"}
)
return "" if content is None else str(content)
def _tool_name(call: Mapping[str, Any]) -> str | None:
function = call.get("function")
if not isinstance(function, Mapping):
return None
name = function.get("name")
return name if isinstance(name, str) and name else None
def _is_continuation_nudge(text: str) -> bool:
folded = text.casefold()
return (
"<system-reminder>" in folded
or (
"continue with the task" in folded
and "resume your thought" in folded
)
)
def _strip_system_reminders(text: str) -> str:
cleaned = SYSTEM_REMINDER_RE.sub("", str(text))
return re.sub(r"\n{3,}", "\n\n", cleaned).strip()
def _bound_recap(text: str) -> str:
"""Keep evidence recaps bounded so one tool result cannot dominate context."""
limit = max(256, TOOL_RECAP_CHARACTERS)
if len(text) <= limit:
return text
head = limit * 2 // 3
tail = limit - head
return (
text[:head]
+ f"\n...[{len(text) - limit} characters omitted]...\n"
+ text[-tail:]
)
def _read_recap(content: str) -> str:
lines: list[str] = []
for raw_line in _strip_system_reminders(content).splitlines():
line = raw_line.strip()
if not line or line.startswith("<system-reminder"):
continue
match = re.match(r"^\d+→\s*(.*)$", line)
if match:
line = match.group(1).strip()
if line:
lines.append(line)
return _bound_recap("\n".join(lines).strip())
def _tool_recap(tool_name: str, content: str) -> str:
cleaned = _strip_system_reminders(content)
if not cleaned:
return f"{tool_name} completed without textual output."
return f"{tool_name} result:\n{_bound_recap(cleaned)}"
def normalize_openclaude_messages(messages: object) -> list[dict[str, Any]]:
"""Preserve native tool history and add bounded evidence recaps.
OpenClaude may return parallel results in a different order from the calls.
Results are therefore matched by ``tool_call_id`` rather than by position.
The recap is emitted only after the whole result batch, so parallel tool
messages remain contiguous for Qwen's chat template.
"""
if not isinstance(messages, list):
raise ValueError("messages must be a list")
normalized: list[dict[str, Any]] = []
pending_by_id: dict[str, str] = {}
pending_order: list[str] = []
pending_recaps: list[str] = []
generated_call_number = 0
def flush_recaps() -> None:
if not pending_recaps:
return
normalized.append(
{
"role": "user",
"content": "[Tool results received]\n"
+ "\n\n".join(pending_recaps),
}
)
pending_recaps.clear()
for raw_message in messages:
if not isinstance(raw_message, Mapping):
raise ValueError("each message must be an object")
message = dict(raw_message)
raw_role = str(message.get("role", "user")).casefold()
content = _content_text(message.get("content"))
if raw_role != "tool":
flush_recaps()
if raw_role in {"system", "developer"}:
normalized.append({"role": "system", "content": content})
continue
if raw_role == "assistant":
calls: list[dict[str, Any]] = []
raw_calls = message.get("tool_calls")
if not isinstance(raw_calls, list):
raw_calls = []
for raw_call in raw_calls:
if not isinstance(raw_call, Mapping):
continue
name = _tool_name(raw_call)
if not name:
continue
generated_call_number += 1
call_id = raw_call.get("id")
if not isinstance(call_id, str) or not call_id:
call_id = f"call_normalized_{generated_call_number}"
if call_id in pending_by_id:
raise ValueError(f"duplicate tool_call id: {call_id}")
function = raw_call.get("function")
arguments = (
function.get("arguments", {})
if isinstance(function, Mapping)
else {}
)
calls.append(
{
"id": call_id,
"type": "function",
"function": {
"name": name,
"arguments": normalize_openai_tool_arguments(
arguments
),
},
}
)
pending_by_id[call_id] = name
pending_order.append(call_id)
if content and (
"[tool results received]" in content.casefold()
or _is_continuation_nudge(content)
):
continue
normalized.append(
{
"role": "assistant",
"content": content if content else None,
**({"tool_calls": calls} if calls else {}),
}
)
continue
if raw_role == "tool":
call_id = message.get("tool_call_id")
tool_name: str | None = None
if isinstance(call_id, str) and call_id:
tool_name = pending_by_id.pop(call_id, None)
if tool_name is None:
explicit_name = message.get("name")
if isinstance(explicit_name, str) and explicit_name:
tool_name = explicit_name
else:
raise ValueError(
"tool result references unknown tool_call_id: "
f"{call_id}"
)
if call_id in pending_order:
pending_order.remove(call_id)
elif pending_order:
call_id = pending_order.pop(0)
tool_name = pending_by_id.pop(call_id)
else:
explicit_name = message.get("name")
if not isinstance(explicit_name, str) or not explicit_name:
raise ValueError("tool result is missing tool_call_id")
tool_name = explicit_name
call_id = None
entry: dict[str, Any] = {
"role": "tool",
"name": tool_name,
"content": content,
}
if isinstance(call_id, str) and call_id:
entry["tool_call_id"] = call_id
normalized.append(entry)
recap = (
_read_recap(content)
if tool_name.casefold() == "read"
else _tool_recap(tool_name, content)
)
if recap:
pending_recaps.append(recap)
continue
original_content = content
content = _strip_system_reminders(content)
if original_content and not content:
continue
if _is_continuation_nudge(content):
continue
normalized.append({"role": "user", "content": content})
flush_recaps()
return normalized
def has_tool_protocol(messages: object) -> bool:
if not isinstance(messages, list):
return False
return any(
isinstance(message, Mapping)
and TOOL_PROTOCOL_MARKER in _content_text(message.get("content"))
for message in messages
)
def add_system_instruction(
messages: list[dict[str, Any]], instruction: str | None
) -> list[dict[str, Any]]:
"""Insert request-local instructions near the current user turn."""
if not instruction:
return messages
prepared = list(messages)
insert_at = 0
for index in range(len(prepared) - 1, -1, -1):
if prepared[index].get("role") == "user":
insert_at = index
break
prepared.insert(insert_at, {"role": "system", "content": instruction})
return prepared
|